in progress2

src/main/java/com/snowflake/kafka/connector/internal/streaming/DirectTopicPartitionChannel.java

@@ -531,6 +531,8 @@ private void insertRowFallbackSupplier(Throwable ex)
    * exception.
    *
    * @param insertErrors errors from validation response. (Only if it has errors)
+   *
+   * IcebergDataTypeParser.deserializeIcebergType(channel.getTableSchema().get("TESTSTRUCT").getIcebergSchema())
    */
   private void handleInsertRowFailure(
       List<InsertValidationResponse.InsertError> insertErrors, SinkRecord kafkaSinkRecord) {

src/main/java/com/snowflake/kafka/connector/internal/streaming/schemaevolution/TableSchemaResolver.java

@@ -92,7 +92,7 @@ private TableSchema getTableSchemaFromRecordSchema(
                     Map.Entry::getKey, Map.Entry::getValue, (oldValue, newValue) -> newValue));
     return new TableSchema(columnsInferredFromSchema);
   }
-
+  // IcebergDataTypeParser.structFromJson(recordNode)
   private TableSchema getTableSchemaFromJson(SinkRecord record, Set<String> columnNamesSet) {
     JsonNode recordNode = RecordService.convertToJson(record.valueSchema(), record.value(), true);
     Map<String, ColumnInfos> columnsInferredFromJson =

src/main/java/com/snowflake/kafka/connector/internal/streaming/schemaevolution/iceberg/IcebergColumnSchema.java

@@ -0,0 +1,17 @@
+package com.snowflake.kafka.connector.internal.streaming.schemaevolution.iceberg;
+
+import net.snowflake.ingest.internal.apache.iceberg.types.Types;
+
+/** Wrapper class for Iceberg schema retrieved from channel */
+public class IcebergColumnSchema {
+
+    private final Types schema;
+
+    public IcebergColumnSchema(Types schema) {
+        this.schema = schema;
+    }
+
+    public Types getSchema() {
+        return schema;
+    }
+}

src/main/java/com/snowflake/kafka/connector/internal/streaming/schemaevolution/iceberg/IcebergColumnTree.java

@@ -0,0 +1,7 @@
+package com.snowflake.kafka.connector.internal.streaming.schemaevolution.iceberg;
+
+/** Class with object types compatible with Snowflake Iceberg table */
+public class IcebergColumnTree {
+
+
+}

src/main/java/com/snowflake/kafka/connector/internal/streaming/schemaevolution/iceberg/IcebergColumnTypeMapper.java

@@ -53,7 +53,7 @@ public String mapToColumnType(Schema.Type kafkaType, String schemaName) {
           return "BINARY";
         }
       case STRUCT:
-        return "OBJECT()";
+        return "OBJECT(k1 INT, k2 INT)";
       case ARRAY:
         throw new IllegalArgumentException("Arrays, struct and map not supported!");
       default:

src/main/java/com/snowflake/kafka/connector/internal/streaming/schemaevolution/iceberg/IcebergNode.java

@@ -0,0 +1,16 @@
+package com.snowflake.kafka.connector.internal.streaming.schemaevolution.iceberg;
+
+import net.snowflake.ingest.internal.apache.iceberg.types.Types;
+
+public class IcebergNode {
+
+    public String name;
+    // todo should be snowflake iceberg table types
+    public Types dataType;
+    // todo add is nullable field
+
+    public IcebergNode(String name, Types dataType) {
+        this.name = name;
+        this.dataType = dataType;
+    }
+}

src/test/java/com/snowflake/kafka/connector/streaming/iceberg/IcebergIngestionSchemaEvolutionIT.java

@@ -131,10 +131,12 @@ void shouldEvolveSchemaAndInsertRecords_withObjects(
     //                    "}";
     // reinsert record with extra field
     service.insert(Collections.singletonList(createKafkaRecord(testStruct, 2, false)));
-    rows = describeTable(tableName);
-    // assertThat(rows).hasSize(15);
+
     service.insert(Collections.singletonList(createKafkaRecord(testStruct, 2, false)));
     waitForOffset(3);
+    String skipStruct = "{ \"skipStruct\": {" + "\"k1\" : 1," + "\"k3\" : 2" + "} " + "}";
+    service.insert(Collections.singletonList(createKafkaRecord(skipStruct, 3, false)));
+    waitForOffset(4);
   }
 
   private void assertRecordsInTable() {

src/test/java/com/snowflake/kafka/connector/streaming/iceberg/TreeTest.java

@@ -0,0 +1,16 @@
+package com.snowflake.kafka.connector.streaming.iceberg;
+
+import net.snowflake.ingest.internal.apache.iceberg.types.Type;
+import net.snowflake.ingest.utils.IcebergDataTypeParser;
+import org.junit.jupiter.api.Test;
+
+public class TreeTest {
+
+    @Test
+    void testParsing() {
+        String plainIcebergSchema = "{\"type\":\"struct\",\"fields\":[{\"id\":23,\"name\":\"k1\",\"required\":false,\"type\":\"int\"},{\"id\":24,\"name\":\"k2\",\"required\":false,\"type\":\"int\"}]}";
+        Type typ = IcebergDataTypeParser.deserializeIcebergType(plainIcebergSchema);
+
+        System.out.println("stop debugger");
+    }
+}