exclude fields from identifier fields

Author: edgao

airbyte-cdk/bulk/toolkits/load-iceberg-parquet/src/main/kotlin/io/airbyte/cdk/load/data/iceberg/parquet/AirbyteTypeToIcebergSchema.kt

@@ -107,15 +107,23 @@ fun ObjectType.toIcebergSchema(primaryKeys: List<List<String>>): Schema {
         // There's no _airbyte_data field, because we flatten the fields.
         // But we should leave the _airbyte_meta field as an actual object.
         val stringifyObjects = name != Meta.COLUMN_NAME_AB_META
+        val icebergType =
+            icebergTypeConverter.convert(field.type, stringifyObjects = stringifyObjects)
         fields.add(
             NestedField.of(
                 id,
                 isOptional,
                 name,
-                icebergTypeConverter.convert(field.type, stringifyObjects = stringifyObjects),
+                icebergType,
             ),
         )
-        if (isPrimaryKey) {
+        // Identifier fields must be primitive types, and cannot be float/double.
+        if (
+            isPrimaryKey &&
+                icebergType.isPrimitiveType &&
+                icebergType != Types.DoubleType.get() &&
+                icebergType != Types.FloatType.get()
+        ) {
             identifierFields.add(id)
         }
     }

airbyte-integrations/connectors/destination-s3-data-lake/build.gradle

@@ -6,7 +6,7 @@ plugins {
 airbyteBulkConnector {
     core = 'load'
     toolkits = ['load-iceberg-parquet', 'load-aws']
-    cdk = '0.334'
+    cdk = 'local'
 }
 
 application {