Removing migrated_data_location from definitionMetadata (#637)

* Removing migrated_data_location from definitionMetadata

add smoke test and delete from db

add smoke test and delete from db

testing smoke test

fix test error

* test recreate

* test recreate

* test recreate

* address nit

---------

Co-authored-by: Giorgio Trettenero <[email protected]>

Author: gtrettenero

metacat-common-server/src/main/java/com/netflix/metacat/common/server/util/MetacatUtils.java

@@ -34,14 +34,14 @@ public class MetacatUtils {
     public static final String NAME_TAGS = "tags";
     private static final String DATA_DEPENDENCY_NODE = "data_dependency";
     private static final String VTTS_FIELD = "valid_thru_utc_ts";
+    public static final String MIGRATED_DATA_LOCATION = "migrated_data_location";
 
     /**
      * Iceberg common view field names.
      */
     public static final String COMMON_VIEW = "common_view";
     public static final String STORAGE_TABLE = "storage_table";
 
-
     /**
      * Default Ctor.
      */

metacat-functional-tests/src/functionalTest/groovy/com/netflix/metacat/MetacatSmokeSpec.groovy

@@ -2685,4 +2685,63 @@ class MetacatSmokeSpec extends Specification {
         assert parentChildRelV1.getChildren(catalogName, databaseName, child21).isEmpty()
         assert parentChildRelV1.getParents(catalogName, databaseName, child21) == [new ParentInfoDto("hive-metastore/iceberg_db/parent2", "CLONE", "p2_uuid")] as Set
     }
+
+    def "Test remove migrated_data_location before soft delete"() {
+        given:
+        def catalogName = 'hive-metastore'
+        def databaseName = 'smoke_db'
+        def tableName = 'test_table_migrated_data_location'
+        def uri = isLocalEnv ? String.format('file:/tmp/%s/%s', databaseName, tableName) : null
+        def tableDto = PigDataDtoProvider.getTable(catalogName, databaseName, tableName, 'gtret', uri)
+
+        // Add multiple metadata fields to verify they persist
+        tableDto.definitionMetadata.put('migrated_data_location', 's3://old/location')
+        tableDto.definitionMetadata.put('preserved_field', 'should remain')
+        tableDto.definitionMetadata.put('another_field', 'should also remain')
+
+        try {
+            api.createDatabase(catalogName, databaseName, new DatabaseCreateRequestDto())
+        } catch (Exception ignored) {
+        }
+
+        when:
+        def createdTable = api.createTable(catalogName, databaseName, tableName, tableDto)
+
+        then:
+        assert createdTable.definitionMetadata.has('migrated_data_location')
+        assert createdTable.definitionMetadata.has('preserved_field')
+        assert createdTable.definitionMetadata.has('another_field')
+
+        // Soft delete the table
+        when:
+        api.deleteTable(catalogName, databaseName, tableName)
+
+        // Recreate the table with the same name and verify migrated_data_location is gone
+        def newTableDto = PigDataDtoProvider.getTable(catalogName, databaseName, tableName, 'gtret', uri)
+        def recreatedTable = api.createTable(catalogName, databaseName, tableName, newTableDto)
+
+        then:
+        // Verify migrated_data_location is not present
+        !recreatedTable.definitionMetadata.has('migrated_data_location')
+        // Verify other fields are still present
+        recreatedTable.definitionMetadata.has('preserved_field')
+        recreatedTable.definitionMetadata.get('preserved_field').asText() == 'should remain'
+        recreatedTable.definitionMetadata.has('another_field')
+        recreatedTable.definitionMetadata.get('another_field').asText() == 'should also remain'
+
+        // Verify the definition metadata doesn't contain migrated_data_location
+        def definitionMetadatas = metadataApi.getDefinitionMetadataList(null, null, null, null, null, null, "$catalogName/$databaseName/$tableName", null)
+        definitionMetadatas.each { definition ->
+            assert !definition.getDefinitionMetadata().has("migrated_data_location")
+        }
+
+        // Verify no exceptions are thrown
+        noExceptionThrown()
+
+        cleanup:
+        try {
+            api.deleteTable(catalogName, databaseName, tableName)
+        } catch (Exception ignored) {
+        }
+    }
 }

metacat-main/src/main/java/com/netflix/metacat/main/services/impl/TableServiceImpl.java

@@ -431,6 +431,15 @@ public TableDto deleteAndReturn(final QualifiedName name, final boolean isMView)
             tagService.delete(name, false);
         } else {
             if (config.canSoftDeleteDataMetadata() && tableDto.isDataExternal()) {
+                // Remove the migrated_data_location field before soft deleting to allow for cleanup of old prefixes
+                if (tableDto.getDefinitionMetadata() != null
+                    && tableDto.getDefinitionMetadata().has(MetacatUtils.MIGRATED_DATA_LOCATION)) {
+                    tableDto.getDefinitionMetadata().remove(MetacatUtils.MIGRATED_DATA_LOCATION);
+                    userMetadataService.saveDefinitionMetadata(tableDto.getName(),
+                        metacatRequestContext.getUserName(),
+                        Optional.of(tableDto.getDefinitionMetadata()),
+                            false);
+                }
                 userMetadataService.softDeleteDataMetadata(metacatRequestContext.getUserName(),
                     Lists.newArrayList(tableDto.getDataUri()));
             }
@@ -453,6 +462,8 @@ private boolean hasTags(@Nullable final TableDto tableDto, final Set<String> has
         return false;
     }
 
+
+
     /**
      * Returns true
      * 1. If the system is configured to delete deifnition metadata.