chore: dont need state manager in iceberg

airbyte-integrations/connectors/destination-iceberg-v2/src/main/kotlin/io/airbyte/integrations/destination/iceberg/v2/IcebergStreamLoader.kt

@@ -10,54 +10,35 @@ import io.airbyte.cdk.load.data.MapperPipeline
 import io.airbyte.cdk.load.message.Batch
 import io.airbyte.cdk.load.message.DestinationFile
 import io.airbyte.cdk.load.message.DestinationRecord
-import io.airbyte.cdk.load.state.DestinationStateManager
-import io.airbyte.cdk.load.state.object_storage.ObjectStorageDestinationState
+import io.airbyte.cdk.load.message.SimpleBatch
+import io.airbyte.cdk.load.state.StreamIncompleteResult
 import io.airbyte.cdk.load.write.StreamLoader
-import io.airbyte.cdk.load.write.object_storage.ObjectStorageStreamLoader
+import io.airbyte.integrations.destination.iceberg.v2.io.IcebergTableCleaner
 import io.airbyte.integrations.destination.iceberg.v2.io.IcebergTableWriterFactory
 import io.airbyte.integrations.destination.iceberg.v2.io.IcebergUtil
 import io.airbyte.integrations.destination.iceberg.v2.io.IcebergUtil.constructGenerationIdSuffix
 import io.github.oshai.kotlinlogging.KotlinLogging
-import java.util.concurrent.atomic.AtomicLong
 import org.apache.iceberg.Table
 
 @SuppressFBWarnings("NP_NONNULL_PARAM_VIOLATION", justification = "Kotlin async continuation")
 class IcebergStreamLoader(
     override val stream: DestinationStream,
     private val table: Table,
     private val icebergTableWriterFactory: IcebergTableWriterFactory,
-    private val destinationStateManager: DestinationStateManager<ObjectStorageDestinationState>,
     private val pipeline: MapperPipeline,
     private val stagingBranchName: String,
     private val mainBranchName: String
 ) : StreamLoader {
     private val log = KotlinLogging.logger {}
 
-    private val partNumber = AtomicLong(0L)
-
-    override suspend fun start() {
-        val state = destinationStateManager.getState(stream)
-        val maxPartNumber =
-            state.generations
-                .filter { it.generationId >= stream.minimumGenerationId }
-                .mapNotNull { it.objects.maxOfOrNull { obj -> obj.partNumber } }
-                .maxOrNull()
-        log.info { "Got max part number from destination state: $maxPartNumber" }
-        maxPartNumber?.let { partNumber.set(it + 1L) }
-    }
-
     override suspend fun processRecords(
         records: Iterator<DestinationRecord>,
         totalSizeBytes: Long
     ): Batch {
-        val partNumber = partNumber.getAndIncrement()
-        val state = destinationStateManager.getState(stream)
-
         icebergTableWriterFactory
             .create(table = table, generationId = constructGenerationIdSuffix(stream))
             .use { writer ->
                 log.info { "Writing records to branch $stagingBranchName" }
-                state.addObject(stream.generationId, stagingBranchName, partNumber)
                 records.forEach { record ->
                     val icebergRecord =
                         IcebergUtil.toRecord(
@@ -82,27 +63,25 @@ class IcebergStreamLoader(
                 log.info { "Finished writing records to $stagingBranchName" }
             }
 
-        return ObjectStorageStreamLoader.StagedObject(
-            remoteObject = stagingBranchName,
-            partNumber = partNumber
-        )
+        return SimpleBatch(Batch.State.PERSISTED)
     }
 
     override suspend fun processFile(file: DestinationFile): Batch {
         throw NotImplementedError("Destination Iceberg does not support universal file transfer.")
     }
 
-    override suspend fun processBatch(batch: Batch): Batch {
-        val stagedObject = batch as ObjectStorageStreamLoader.StagedObject<*>
-        log.info { "Moving staged object from $stagingBranchName to $mainBranchName" }
-        table.manageSnapshots().fastForwardBranch(mainBranchName, stagingBranchName).commit()
-
-        val state = destinationStateManager.getState(stream)
-        state.removeObject(stream.generationId, stagingBranchName)
-        state.addObject(stream.generationId, mainBranchName, stagedObject.partNumber)
-
-        val finalizedObject =
-            ObjectStorageStreamLoader.FinalizedObject(remoteObject = mainBranchName)
-        return finalizedObject
+    override suspend fun close(streamFailure: StreamIncompleteResult?) {
+        if (streamFailure == null) {
+            table.manageSnapshots().fastForwardBranch(mainBranchName, stagingBranchName).commit()
+            if (stream.minimumGenerationId > 0) {
+                val icebergTableCleaner = IcebergTableCleaner()
+                for (i in 0 until stream.minimumGenerationId) {
+                    icebergTableCleaner.deleteGenerationId(
+                        table,
+                        constructGenerationIdSuffix(i),
+                    )
+                }
+            }
+        }
     }
 }

airbyte-integrations/connectors/destination-iceberg-v2/src/main/kotlin/io/airbyte/integrations/destination/iceberg/v2/IcebergV2Writer.kt

@@ -8,8 +8,6 @@ import io.airbyte.cdk.load.command.DestinationStream
 import io.airbyte.cdk.load.data.iceberg.parquet.toIcebergSchema
 import io.airbyte.cdk.load.data.parquet.ParquetMapperPipelineFactory
 import io.airbyte.cdk.load.data.withAirbyteMeta
-import io.airbyte.cdk.load.state.DestinationStateManager
-import io.airbyte.cdk.load.state.object_storage.ObjectStorageDestinationState
 import io.airbyte.cdk.load.write.DestinationWriter
 import io.airbyte.cdk.load.write.StreamLoader
 import io.airbyte.integrations.destination.iceberg.v2.io.IcebergTableWriterFactory
@@ -21,7 +19,6 @@ import org.apache.iceberg.catalog.TableIdentifier
 
 @Singleton
 class IcebergV2Writer(
-    private val destinationStateManager: DestinationStateManager<ObjectStorageDestinationState>,
     private val icebergTableWriterFactory: IcebergTableWriterFactory,
     private val icebergConfiguration: IcebergV2Configuration,
 ) : DestinationWriter {
@@ -48,7 +45,6 @@ class IcebergV2Writer(
             stream = stream,
             table = table,
             icebergTableWriterFactory = icebergTableWriterFactory,
-            destinationStateManager = destinationStateManager,
             pipeline = pipeline,
             stagingBranchName = DEFAULT_STAGING_BRANCH,
             mainBranchName = icebergConfiguration.nessieServerConfiguration.mainBranchName,

airbyte-integrations/connectors/destination-iceberg-v2/src/main/kotlin/io/airbyte/integrations/destination/iceberg/v2/io/IcebergUtil.kt

@@ -45,12 +45,16 @@ object IcebergUtil {
     }
 
     fun constructGenerationIdSuffix(stream: DestinationStream): String {
-        if (stream.generationId < 0) {
+        return constructGenerationIdSuffix(stream.generationId)
+    }
+
+    fun constructGenerationIdSuffix(generationId: Long): String {
+        if (generationId < 0) {
             throw IllegalArgumentException(
-                "GenerationId must be non-negative. Provided: ${stream.generationId}",
+                "GenerationId must be non-negative. Provided: ${generationId}",
             )
         }
-        return "ab-generation-id-${stream.generationId}"
+        return "ab-generation-id-${generationId}"
     }
     /**
      * Builds an Iceberg [Catalog].