use table cleaner

airbytehq · Dec 17, 2024 · ea13d27 · ea13d27
1 parent 6204548
commit ea13d27
Showing 1 changed file with 9 additions and 2 deletions.
diff --git a/...ration/kotlin/io/airbyte/integrations/destination/iceberg/v2/IcebergDestinationCleaner.kt b/...ration/kotlin/io/airbyte/integrations/destination/iceberg/v2/IcebergDestinationCleaner.kt
@@ -7,6 +7,8 @@ package io.airbyte.integrations.destination.iceberg.v2
 import io.airbyte.cdk.load.test.util.DestinationCleaner
 import io.airbyte.cdk.load.test.util.IntegrationTest.Companion.isNamespaceOld
 import io.airbyte.cdk.load.test.util.IntegrationTest.Companion.randomizedNamespaceRegex
+import io.airbyte.integrations.destination.iceberg.v2.io.IcebergTableCleaner
+import io.airbyte.integrations.destination.iceberg.v2.io.IcebergUtil
 import org.apache.iceberg.catalog.Catalog
 import org.apache.iceberg.catalog.Namespace
 import org.apache.iceberg.catalog.SupportsNamespaces
@@ -18,9 +20,14 @@ class IcebergDestinationCleaner(private val catalog: Catalog) : DestinationClean
                 val namespace = it.level(0)
                 randomizedNamespaceRegex.matches(namespace) && isNamespaceOld(namespace)
             }
+
+        // we're passing explicit TableIdentifier to clearTable, so just use SimpleTableIdGenerator
+        val tableCleaner = IcebergTableCleaner(IcebergUtil(SimpleTableIdGenerator()))
+
         namespaces.forEach { namespace ->
-            catalog.listTables(namespace).forEach { table ->
-                catalog.dropTable(table, /* purge = */ true)
+            catalog.listTables(namespace).forEach { tableId ->
+                val table = catalog.loadTable(tableId)
+                tableCleaner.clearTable(catalog, tableId, table.io(), table.location())
             }
             catalog.dropNamespace(namespace)
         }