fix mapper; remove expected record mapper

airbytehq · Dec 19, 2024 · df3e451 · df3e451
1 parent e8ba94c
commit df3e451
Show file tree

Hide file tree

Showing 4 changed files with 19 additions and 20 deletions.
diff --git a/...src/main/kotlin/io/airbyte/cdk/load/data/iceberg/parquet/IcebergParquetPipelineFactory.kt b/...src/main/kotlin/io/airbyte/cdk/load/data/iceberg/parquet/IcebergParquetPipelineFactory.kt
@@ -20,7 +20,8 @@ class IcebergParquetPipelineFactory : MapperPipelineFactory {
         MapperPipeline(
             stream.schema,
             listOf(
-                AirbyteSchemaNoopMapper() to SchemalessValuesToJsonString(),
+                // TODO not sure why base parquet was doing this as a noop
+                SchemalessTypesToStringType() to SchemalessValuesToJsonString(),
                 AirbyteSchemaNoopMapper() to NullOutOfRangeIntegers(),
                 MergeUnions() to AirbyteValueNoopMapper(),
                 UnionTypeToDisjointRecord() to UnionValueToDisjointRecord(),

diff --git a/...t/src/main/kotlin/io/airbyte/cdk/load/data/iceberg/parquet/SchemalessTypesToStringType.kt b/...t/src/main/kotlin/io/airbyte/cdk/load/data/iceberg/parquet/SchemalessTypesToStringType.kt
@@ -0,0 +1,15 @@
+package io.airbyte.cdk.load.data.iceberg.parquet
+
+import io.airbyte.cdk.load.data.AirbyteSchemaIdentityMapper
+import io.airbyte.cdk.load.data.ArrayTypeWithoutSchema
+import io.airbyte.cdk.load.data.ObjectTypeWithEmptySchema
+import io.airbyte.cdk.load.data.ObjectTypeWithoutSchema
+import io.airbyte.cdk.load.data.StringType
+import io.airbyte.cdk.load.data.UnknownType
+
+class SchemalessTypesToStringType: AirbyteSchemaIdentityMapper {
+    override fun mapArrayWithoutSchema(schema: ArrayTypeWithoutSchema) = StringType
+    override fun mapObjectWithEmptySchema(schema: ObjectTypeWithEmptySchema) = StringType
+    override fun mapObjectWithoutSchema(schema: ObjectTypeWithoutSchema) = StringType
+    override fun mapUnknown(schema: UnknownType) = StringType
+}
diff --git a/...-integration/kotlin/io/airbyte/integrations/destination/iceberg/v2/IcebergRecordMapper.kt b/...-integration/kotlin/io/airbyte/integrations/destination/iceberg/v2/IcebergRecordMapper.kt
diff --git a/...t-integration/kotlin/io/airbyte/integrations/destination/iceberg/v2/IcebergV2WriteTest.kt b/...t-integration/kotlin/io/airbyte/integrations/destination/iceberg/v2/IcebergV2WriteTest.kt
@@ -8,6 +8,7 @@ import com.fasterxml.jackson.databind.JsonNode
 import com.fasterxml.jackson.databind.ObjectMapper
 import io.airbyte.cdk.load.test.util.DestinationCleaner
 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 java.nio.file.Files
@@ -28,7 +29,7 @@ abstract class IcebergV2WriteTest(
         IcebergV2Specification::class.java,
         IcebergV2DataDumper,
         destinationCleaner,
-        IcebergRecordMapper,
+        NoopExpectedRecordMapper,
         isStreamSchemaRetroactive = true,
         supportsDedup = false,
         stringifySchemalessObjects = true,