Destination Iceberg v2: basic test scaffolds + related runtime classes (

#48544)

airbyte-cdk/bulk/core/load/src/testFixtures/kotlin/io/airbyte/cdk/load/check/CheckIntegrationTest.kt

@@ -25,7 +25,6 @@ import org.junit.jupiter.api.assertAll
 data class CheckTestConfig(val configPath: Path, val featureFlags: Set<FeatureFlag> = emptySet())
 
 open class CheckIntegrationTest<T : ConfigurationSpecification>(
-    val configurationClass: Class<T>,
     val successConfigFilenames: List<CheckTestConfig>,
     val failConfigFilenamesAndFailureReasons: Map<CheckTestConfig, Pattern>,
 ) :

airbyte-cdk/bulk/core/load/src/testFixtures/kotlin/io/airbyte/cdk/load/spec/SpecTest.kt

@@ -5,11 +5,10 @@
 package io.airbyte.cdk.load.spec
 
 import com.deblock.jsondiff.DiffGenerator
-import com.deblock.jsondiff.diff.JsonDiff
 import com.deblock.jsondiff.matcher.CompositeJsonMatcher
 import com.deblock.jsondiff.matcher.JsonMatcher
-import com.deblock.jsondiff.matcher.LenientJsonObjectPartialMatcher
 import com.deblock.jsondiff.matcher.StrictJsonArrayPartialMatcher
+import com.deblock.jsondiff.matcher.StrictJsonObjectPartialMatcher
 import com.deblock.jsondiff.matcher.StrictPrimitivePartialMatcher
 import com.deblock.jsondiff.viewer.OnlyErrorDiffViewer
 import io.airbyte.cdk.command.FeatureFlag
@@ -41,6 +40,8 @@ abstract class SpecTest :
         NoopDestinationCleaner,
         NoopExpectedRecordMapper,
     ) {
+    private val testResourcesPath = Path.of("src/test-integration/resources")
+
     @Test
     fun testSpecOss() {
         testSpec("expected-spec-oss.json")
@@ -55,9 +56,10 @@ abstract class SpecTest :
         expectedSpecFilename: String,
         vararg featureFlags: FeatureFlag,
     ) {
-        val expectedSpecPath = Path.of("src/test-integration/resources", expectedSpecFilename)
+        val expectedSpecPath = testResourcesPath.resolve(expectedSpecFilename)
 
         if (!Files.exists(expectedSpecPath)) {
+            Files.createDirectories(testResourcesPath)
             Files.createFile(expectedSpecPath)
         }
         val expectedSpec = Files.readString(expectedSpecPath)
@@ -80,15 +82,28 @@ abstract class SpecTest :
         val jsonMatcher: JsonMatcher =
             CompositeJsonMatcher(
                 StrictJsonArrayPartialMatcher(),
-                LenientJsonObjectPartialMatcher(),
+                StrictJsonObjectPartialMatcher(),
                 StrictPrimitivePartialMatcher(),
             )
-        val diff: JsonDiff =
-            DiffGenerator.diff(expectedSpec, Jsons.writeValueAsString(spec), jsonMatcher)
+        val diff =
+            OnlyErrorDiffViewer.from(
+                    DiffGenerator.diff(expectedSpec, Jsons.writeValueAsString(spec), jsonMatcher)
+                )
+                .toString()
         assertAll(
             "Spec snapshot test failed. Run this test locally and then `git diff <...>/$expectedSpecFilename` to see what changed, and commit the diff if that change was intentional.",
-            { Assertions.assertEquals("", OnlyErrorDiffViewer.from(diff).toString()) },
-            { Assertions.assertEquals(expectedSpec, actualSpecPrettyPrint) }
+            {
+                Assertions.assertTrue(
+                    diff.isEmpty(),
+                    "Detected semantic diff in JSON:\n" + diff.prependIndent("\t\t")
+                )
+            },
+            {
+                Assertions.assertTrue(
+                    expectedSpec == actualSpecPrettyPrint,
+                    "File contents did not equal generated spec, see git diff for details"
+                )
+            }
         )
     }
 }

airbyte-integrations/connectors/destination-dev-null/src/test-integration/kotlin/io/airbyte/integrations/destination/dev_null/DevNullCheckIntegrationTest.kt

@@ -12,7 +12,6 @@ import org.junit.jupiter.api.Test
 
 class DevNullCheckIntegrationTest :
     CheckIntegrationTest<DevNullSpecificationOss>(
-        DevNullSpecificationOss::class.java,
         successConfigFilenames =
             listOf(
                 CheckTestConfig(DevNullTestUtils.loggingConfigPath),

airbyte-integrations/connectors/destination-iceberg-v2/gradle.properties

@@ -0,0 +1 @@
+testExecutionConcurrency=-1

airbyte-integrations/connectors/destination-iceberg-v2/metadata.yaml

@@ -2,7 +2,7 @@ data:
   connectorSubtype: file
   connectorType: destination
   definitionId: 37a928c1-2d5c-431a-a97d-ae236bd1ea0c
-  dockerImageTag: 0.1.0
+  dockerImageTag: 0.1.1
   dockerRepository: airbyte/destination-iceberg-v2
   githubIssueLabel: destination-iceberg-v2
   icon: s3.svg
@@ -22,15 +22,15 @@ data:
     ql: 100
   supportLevel: community
   supportsRefreshes: true
-#  connectorTestSuitesOptions:
-#    - suite: unitTests
-#    - suite: integrationTests
-#      testSecrets:
-#        - name: SECRET_DESTINATION-S3-V2-MINIMAL-REQUIRED-CONFIG
-#          fileName: s3_dest_v2_minimal_required_config.json
-#          secretStore:
-#            type: GSM
-#            alias: airbyte-connector-testing-secret-store
+  connectorTestSuitesOptions:
+    - suite: unitTests
+    - suite: integrationTests
+      testSecrets:
+        - name: SECRET_DESTINATION-S3-V2-MINIMAL-REQUIRED-CONFIG
+          fileName: s3_dest_v2_minimal_required_config.json
+          secretStore:
+            type: GSM
+            alias: airbyte-connector-testing-secret-store
 #        - name: SECRET_DESTINATION-S3-V2-JSONL-ROOT-LEVEL-FLATTENING
 #          fileName: s3_dest_v2_jsonl_root_level_flattening_config.json
 #          secretStore:

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

@@ -0,0 +1,15 @@
+/*
+ * Copyright (c) 2024 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.integrations.destination.iceberg_v2
+
+import io.airbyte.cdk.load.check.DestinationChecker
+import javax.inject.Singleton
+
+@Singleton
+class IcebergV2Checker : DestinationChecker<IcebergV2Configuration> {
+    override fun check(config: IcebergV2Configuration) {
+        // TODO validate the config
+    }
+}

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

@@ -0,0 +1,25 @@
+/*
+ * Copyright (c) 2024 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.integrations.destination.iceberg_v2
+
+import io.airbyte.cdk.load.command.DestinationConfiguration
+import io.airbyte.cdk.load.command.DestinationConfigurationFactory
+import io.airbyte.integrations.destination.iceberg.v2.IcebergV2Specification
+import javax.inject.Singleton
+
+// TODO put real fields here
+data class IcebergV2Configuration(val something: String) : DestinationConfiguration()
+
+@Singleton
+class IcebergV2ConfigurationFactory :
+    DestinationConfigurationFactory<IcebergV2Specification, IcebergV2Configuration> {
+    override fun makeWithoutExceptionHandling(
+        pojo: IcebergV2Specification
+    ): IcebergV2Configuration {
+        // TODO convert from the jackson-friendly IcebergV2Specification
+        //   to the programmer-friendly IcebergV2Configuration
+        return IcebergV2Configuration("hello world")
+    }
+}

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

@@ -0,0 +1,33 @@
+/*
+ * Copyright (c) 2024 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.integrations.destination.iceberg_v2
+
+import io.airbyte.cdk.load.command.DestinationStream
+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.message.SimpleBatch
+import io.airbyte.cdk.load.write.DestinationWriter
+import io.airbyte.cdk.load.write.StreamLoader
+import javax.inject.Singleton
+
+@Singleton
+class IcebergV2Writer : DestinationWriter {
+    override fun createStreamLoader(stream: DestinationStream): StreamLoader {
+        // TODO instantiate an actual IcebergStreamLoader
+        return object : StreamLoader {
+            override val stream = stream
+
+            override suspend fun processRecords(
+                records: Iterator<DestinationRecord>,
+                totalSizeBytes: Long
+            ) = SimpleBatch(state = Batch.State.COMPLETE)
+
+            override suspend fun processFile(file: DestinationFile): Batch {
+                throw NotImplementedError()
+            }
+        }
+    }
+}

airbyte-integrations/connectors/destination-iceberg-v2/src/test-integration/kotlin/io/airbyte/integrations/destination/iceberg.v2/IcebergV2CheckTest.kt

@@ -0,0 +1,17 @@
+/*
+ * Copyright (c) 2024 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.integrations.destination.iceberg.v2
+
+import io.airbyte.cdk.load.check.CheckIntegrationTest
+import io.airbyte.cdk.load.check.CheckTestConfig
+import java.nio.file.Path
+
+class IcebergV2CheckTest :
+    CheckIntegrationTest<IcebergV2Specification>(
+        successConfigFilenames =
+            listOf(CheckTestConfig(Path.of(IcebergV2TestUtil.SOME_RANDOM_S3_CONFIG))),
+        // TODO we maybe should add some configs that are expected to fail `check`
+        failConfigFilenamesAndFailureReasons = mapOf(),
+    )

airbyte-integrations/connectors/destination-iceberg-v2/src/test-integration/kotlin/io/airbyte/integrations/destination/iceberg.v2/IcebergV2SpecTest.kt

@@ -0,0 +1,9 @@
+/*
+ * Copyright (c) 2024 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.integrations.destination.iceberg.v2
+
+import io.airbyte.cdk.load.spec.SpecTest
+
+class IcebergV2SpecTest : SpecTest()

airbyte-integrations/connectors/destination-iceberg-v2/src/test-integration/kotlin/io/airbyte/integrations/destination/iceberg.v2/IcebergV2TestUtil.kt

@@ -0,0 +1,14 @@
+/*
+ * Copyright (c) 2024 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.integrations.destination.iceberg.v2
+
+import java.nio.file.Files
+import java.nio.file.Path
+
+object IcebergV2TestUtil {
+    // TODO this is just here as an example, we should remove it + add real configs
+    const val SOME_RANDOM_S3_CONFIG = "secrets/s3_dest_v2_minimal_required_config.json"
+    fun getConfig(configPath: String): String = Files.readString(Path.of(configPath))
+}

airbyte-integrations/connectors/destination-iceberg-v2/src/test-integration/kotlin/io/airbyte/integrations/destination/iceberg.v2/IcebergV2WriteTest.kt

@@ -0,0 +1,33 @@
+/*
+ * Copyright (c) 2024 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.integrations.destination.iceberg.v2
+
+import io.airbyte.cdk.load.test.util.FakeDataDumper
+import io.airbyte.cdk.load.test.util.NoopDestinationCleaner
+import io.airbyte.cdk.load.test.util.NoopExpectedRecordMapper
+import io.airbyte.cdk.load.write.BasicFunctionalityIntegrationTest
+import io.airbyte.cdk.load.write.StronglyTyped
+import org.junit.jupiter.api.Disabled
+
+abstract class IcebergV2WriteTest(path: String) :
+    BasicFunctionalityIntegrationTest(
+        IcebergV2TestUtil.getConfig(path),
+        IcebergV2Specification::class.java,
+        FakeDataDumper,
+        NoopDestinationCleaner,
+        NoopExpectedRecordMapper,
+        // TODO let's validate these - I'm making some assumptions about how iceberg works
+        isStreamSchemaRetroactive = true,
+        supportsDedup = false,
+        stringifySchemalessObjects = true,
+        promoteUnionToObject = true,
+        preserveUndeclaredFields = false,
+        commitDataIncrementally = false,
+        allTypesBehavior = StronglyTyped(),
+    )
+
+// TODO replace this with a real test class for an actual config
+@Disabled("nowhere even close to functional")
+class FakeIcebergWriteTest : IcebergV2WriteTest(IcebergV2TestUtil.SOME_RANDOM_S3_CONFIG)

airbyte-integrations/connectors/destination-iceberg-v2/src/test-integration/resources/expected-spec-cloud.json

@@ -0,0 +1,14 @@
+{
+  "documentationUrl" : "https://docs.airbyte.com/integrations/destinations/s3",
+  "connectionSpecification" : {
+    "$schema" : "http://json-schema.org/draft-07/schema#",
+    "title" : "Iceberg V2 Destination Spec",
+    "type" : "object",
+    "additionalProperties" : true,
+    "properties" : { }
+  },
+  "supportsIncremental" : true,
+  "supportsNormalization" : false,
+  "supportsDBT" : false,
+  "supported_destination_sync_modes" : [ "append" ]
+}

airbyte-integrations/connectors/destination-iceberg-v2/src/test-integration/resources/expected-spec-oss.json

@@ -0,0 +1,14 @@
+{
+  "documentationUrl" : "https://docs.airbyte.com/integrations/destinations/s3",
+  "connectionSpecification" : {
+    "$schema" : "http://json-schema.org/draft-07/schema#",
+    "title" : "Iceberg V2 Destination Spec",
+    "type" : "object",
+    "additionalProperties" : true,
+    "properties" : { }
+  },
+  "supportsIncremental" : true,
+  "supportsNormalization" : false,
+  "supportsDBT" : false,
+  "supported_destination_sync_modes" : [ "append" ]
+}

airbyte-integrations/connectors/destination-s3-v2/src/test-integration/kotlin/io/airbyte/integrations/destination/s3_v2/S3V2CheckTest.kt

@@ -12,7 +12,6 @@ import org.junit.jupiter.api.Test
 
 class S3V2CheckTest :
     CheckIntegrationTest<S3V2Specification>(
-        S3V2Specification::class.java,
         successConfigFilenames =
             listOf(
                 CheckTestConfig(