Store the previous_metadata_location for iceberg tables in polaris (N…

…etflix#495)

metacat-connector-polaris/src/functionalTest/resources/schema.sql

@@ -12,7 +12,8 @@ create table TBLS (
   id uuid default gen_random_uuid() not null primary key,
   db_name varchar(255) not null,
   tbl_name varchar(255) not null,
-  metadata_location varchar(1024),
+  previous_metadata_location varchar(8192),
+  metadata_location varchar(8192),
   constraint uniq_name unique(db_name, tbl_name),
   foreign key (db_name) references DBS(name) ON DELETE CASCADE ON UPDATE CASCADE
 );

metacat-connector-polaris/src/main/java/com/netflix/metacat/connector/polaris/store/entities/PolarisTableEntity.java

@@ -47,6 +47,11 @@ public class PolarisTableEntity {
     @Column(name = "tbl_name", nullable = false)
     private String tblName;
 
+    @Basic
+    @Setter
+    @Column(name = "previous_metadata_location", nullable = true, updatable = true)
+    private String previousMetadataLocation;
+
     @Basic
     @Setter
     @Column(name = "metadata_location", nullable = true, updatable = true)

metacat-connector-polaris/src/main/java/com/netflix/metacat/connector/polaris/store/repos/PolarisTableRepository.java

@@ -87,8 +87,9 @@ Slice<PolarisTableEntity> findAllTablesByDbNameAndTablePrefix(
      * @return number of rows that are updated.
      */
     @Modifying(flushAutomatically = true, clearAutomatically = true)
-    @Query("UPDATE PolarisTableEntity t SET t.metadataLocation = :newLocation, t.version = t.version + 1 "
-        + "WHERE t.metadataLocation = :expectedLocation AND t.dbName = :dbName AND t.tblName = :tableName")
+    @Query("UPDATE PolarisTableEntity t SET t.metadataLocation = :newLocation, "
+            + "t.previousMetadataLocation = t.metadataLocation, t.version = t.version + 1 "
+            + "WHERE t.metadataLocation = :expectedLocation AND t.dbName = :dbName AND t.tblName = :tableName")
     @Transactional
     int updateMetadataLocation(
         @Param("dbName") final String dbName,

metacat-connector-polaris/src/test/java/com/netflix/metacat/connector/polaris/store/PolarisStoreConnectorTest.java

@@ -203,7 +203,9 @@ public void updateMetadataLocation() {
         // successful update should happen.
         updatedSuccess = polarisConnector.updateTableMetadataLocation(dbName, tblName, metadataLocation, newLocation);
         Assert.assertTrue(updatedSuccess);
-
+        final PolarisTableEntity updatedEntity = polarisConnector.
+                getTable(dbName, tblName).orElseThrow(() -> new RuntimeException("Expected to find saved entity"));
+        Assert.assertEquals(updatedEntity.getPreviousMetadataLocation(), metadataLocation);
 
         // after the successful update, the same call should fail, since the current metadataLocation has changed.
         updatedSuccess = polarisConnector.updateTableMetadataLocation(dbName, tblName, metadataLocation, newLocation);

metacat-connector-polaris/src/test/resources/h2db/schema.sql

@@ -12,6 +12,7 @@ create table TBLS (
   id IDENTITY not null primary key,
   db_name varchar(255) not null,
   tbl_name varchar(255) not null,
+  previous_metadata_location varchar(1024),
   metadata_location varchar(1024),
   constraint uniq_name unique(db_name, tbl_name),
   foreign key (db_name) references DBS(name) ON DELETE CASCADE ON UPDATE CASCADE

metacat-functional-tests/metacat-test-cluster/datastores/crdb/sql/schema.sql

@@ -12,7 +12,8 @@ create table TBLS (
                       id uuid default gen_random_uuid() not null primary key,
                       db_name varchar(255) not null,
                       tbl_name varchar(255) not null,
-                      metadata_location varchar(1024),
+                      previous_metadata_location varchar(8192),
+                      metadata_location varchar(8192),
                       constraint uniq_name unique(db_name, tbl_name),
                       foreign key (db_name) references DBS(name) ON DELETE CASCADE ON UPDATE CASCADE
 );