Kafka Connect: Support VARIANT when record convert

kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/RecordConverter.java

@@ -35,6 +35,7 @@
 import java.time.format.DateTimeParseException;
 import java.time.temporal.Temporal;
 import java.util.Base64;
+import java.util.Collection;
 import java.util.Date;
 import java.util.List;
 import java.util.Locale;
@@ -52,6 +53,7 @@
 import org.apache.iceberg.mapping.NameMapping;
 import org.apache.iceberg.mapping.NameMappingParser;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.apache.iceberg.types.Type;
 import org.apache.iceberg.types.Type.PrimitiveType;
@@ -64,6 +66,12 @@
 import org.apache.iceberg.util.ByteBuffers;
 import org.apache.iceberg.util.DateTimeUtil;
 import org.apache.iceberg.util.UUIDUtil;
+import org.apache.iceberg.variants.ShreddedObject;
+import org.apache.iceberg.variants.ValueArray;
+import org.apache.iceberg.variants.Variant;
+import org.apache.iceberg.variants.VariantMetadata;
+import org.apache.iceberg.variants.VariantValue;
+import org.apache.iceberg.variants.Variants;
 import org.apache.kafka.connect.data.Struct;
 import org.apache.kafka.connect.errors.ConnectException;
 
@@ -142,6 +150,8 @@ private Object convertValue(
         return convertTimeValue(value);
       case TIMESTAMP:
         return convertTimestampValue(value, (TimestampType) type);
+      case VARIANT:
+        return convertVariantValue(value);
     }
     throw new UnsupportedOperationException("Unsupported type: " + type.typeId());
   }
@@ -464,6 +474,119 @@ protected Temporal convertTimestampValue(Object value, TimestampType type) {
     return convertLocalDateTime(value);
   }
 
+  protected Variant convertVariantValue(Object value) {
+    if (value instanceof Variant) {
+      return (Variant) value;
+    } else if (value instanceof ByteBuffer) {
+      return Variant.from((ByteBuffer) value);
+    }
+
+    List<String> allFieldNames =
+        collectFieldNames(value).stream().distinct().sorted().collect(Collectors.toList());
+    VariantMetadata metadata =
+        allFieldNames.isEmpty() ? Variants.emptyMetadata() : Variants.metadata(allFieldNames);
+    VariantValue variantValue = objectToVariantValue(value, metadata);
+    return Variant.of(metadata, variantValue);
+  }
+
+  /**
+   * Collects all field names (map keys) from the entire object tree. Used to build a single
+   * VariantMetadata for the whole Variant (required for nested maps).
+   */
+  private static List<String> collectFieldNames(Object value) {
+    if (value == null) {
+      return Lists.newArrayList();
+    }
+    if (value instanceof Map) {
+      Map<?, ?> map = (Map<?, ?>) value;
+      List<String> names = Lists.newArrayList();
+      map.keySet().stream().map(Object::toString).forEach(names::add);
+      for (Object v : map.values()) {
+        names.addAll(collectFieldNames(v));
+      }
+      return names;
+    }
+    if (value instanceof Collection) {
+      List<String> names = Lists.newArrayList();
+      for (Object element : (Collection<?>) value) {
+        names.addAll(collectFieldNames(element));
+      }
+      return names;
+    }
+    return Lists.newArrayList();
+  }
+
+  /**
+   * Recursively converts a Java object to a VariantValue using the given shared metadata for all
+   * nested maps. Handles primitives, List (array), and Map (object); map keys become field names.
+   */
+  private static VariantValue objectToVariantValue(Object value, VariantMetadata metadata) {
+    if (value == null) {
+      return Variants.ofNull();
+    }
+    if (value instanceof Boolean) {
+      return Variants.of((Boolean) value);
+    }
+    if (value instanceof Number) {
+      return numberToVariantValue((Number) value);
+    }
+    if (value instanceof String) {
+      return Variants.of((String) value);
+    }
+    if (value instanceof ByteBuffer) {
+      return Variants.of((ByteBuffer) value);
+    }
+    if (value instanceof byte[]) {
+      return Variants.of(ByteBuffer.wrap((byte[]) value));
+    }
+    if (value instanceof BigDecimal) {
+      return Variants.of((BigDecimal) value);
+    }
+    if (value instanceof UUID) {
+      return Variants.ofUUID((UUID) value);
+    }
+    if (value instanceof Collection) {
+      ValueArray array = Variants.array();
+      for (Object element : (Collection<?>) value) {
+        array.add(objectToVariantValue(element, metadata));
+      }
+      return array;
+    }
+    if (value instanceof Map) {
+      Map<?, ?> map = (Map<?, ?>) value;
+      ShreddedObject object = Variants.object(metadata);
+      map.forEach((k, v) -> object.put(k.toString(), objectToVariantValue(v, metadata)));
+      return object;
+    }
+    throw new IllegalArgumentException("Cannot convert to variant: " + value.getClass().getName());
+  }
+
+  private static VariantValue numberToVariantValue(Number value) {
+    if (value instanceof Integer) {
+      return Variants.of((Integer) value);
+    }
+    if (value instanceof Long) {
+      return Variants.of((Long) value);
+    }
+    if (value instanceof Float) {
+      return Variants.of((Float) value);
+    }
+    if (value instanceof Double) {
+      return Variants.of((Double) value);
+    }
+    if (value instanceof Byte) {
+      return Variants.of((Byte) value);
+    }
+    if (value instanceof Short) {
+      return Variants.of((Short) value);
+    }
+    Number num = value;
+    if (num.doubleValue() == num.longValue()) {
+      return Variants.of(num.longValue());
+    }
+    return Variants.of(num.doubleValue());
+  }
+
   @SuppressWarnings("JavaUtilDate")
   private OffsetDateTime convertOffsetDateTime(Object value) {
     if (value instanceof Number) {

kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/data/TestRecordConverter.java

@@ -74,7 +74,10 @@
 import org.apache.iceberg.types.Types.TimeType;
 import org.apache.iceberg.types.Types.TimestampType;
 import org.apache.iceberg.types.Types.UUIDType;
+import org.apache.iceberg.types.Types.VariantType;
 import org.apache.iceberg.util.UUIDUtil;
+import org.apache.iceberg.variants.PhysicalType;
+import org.apache.iceberg.variants.Variant;
 import org.apache.kafka.connect.data.Decimal;
 import org.apache.kafka.connect.data.Schema;
 import org.apache.kafka.connect.data.SchemaBuilder;
@@ -152,6 +155,9 @@ public class TestRecordConverter {
           NestedField.required(
               100, "stma", MapType.ofRequired(101, 102, StringType.get(), ID_SCHEMA.asStruct())));
 
+  private static final org.apache.iceberg.Schema VARIANT_SCHEMA =
+      new org.apache.iceberg.Schema(NestedField.required(1, "v", VariantType.get()));
+
   private static final Schema CONNECT_SCHEMA =
       SchemaBuilder.struct()
           .field("i", Schema.INT32_SCHEMA)
@@ -881,6 +887,102 @@ public void testEvolveTypeDetectionStructNested() {
     assertThat(updateMap.get("st.ff").type()).isInstanceOf(DoubleType.class);
   }
 
+  @Test
+  public void testConvertVariantFromPrimitiveString() {
+    Table table = mock(Table.class);
+    when(table.schema()).thenReturn(VARIANT_SCHEMA);
+    RecordConverter converter = new RecordConverter(table, config);
+
+    Record record = converter.convert(ImmutableMap.of("v", "hello"));
+    Variant variant = (Variant) record.getField("v");
+
+    assertThat(variant).isNotNull();
+    assertThat(variant.metadata()).isNotNull();
+    assertThat(variant.metadata().dictionarySize()).isEqualTo(0);
+    assertThat(variant.value().type()).isEqualTo(PhysicalType.STRING);
+    assertThat(variant.value().asPrimitive().get()).isEqualTo("hello");
+  }
+
+  @Test
+  public void testConvertVariantFromPrimitiveNumber() {
+    Table table = mock(Table.class);
+    when(table.schema()).thenReturn(VARIANT_SCHEMA);
+    RecordConverter converter = new RecordConverter(table, config);
+
+    Record record = converter.convert(ImmutableMap.of("v", 123));
+    Variant variant = (Variant) record.getField("v");
+
+    assertThat(variant).isNotNull();
+    assertThat(variant.metadata()).isNotNull();
+    assertThat(variant.metadata().dictionarySize()).isEqualTo(0);
+    assertThat(variant.value().type()).isEqualTo(PhysicalType.INT32);
+    assertThat(variant.value().asPrimitive().get()).isEqualTo(123);
+  }
+
+  @Test
+  public void testConvertVariantFromList() {
+    Table table = mock(Table.class);
+    when(table.schema()).thenReturn(VARIANT_SCHEMA);
+    RecordConverter converter = new RecordConverter(table, config);
+
+    Record record = converter.convert(ImmutableMap.of("v", ImmutableList.of("hello", 1)));
+    Variant variant = (Variant) record.getField("v");
+
+    assertThat(variant).isNotNull();
+    assertThat(variant.metadata()).isNotNull();
+    assertThat(variant.metadata().dictionarySize()).isEqualTo(0);
+    assertThat(variant.value().type()).isEqualTo(PhysicalType.ARRAY);
+    assertThat(variant.value().asArray().numElements()).isEqualTo(2);
+    assertThat(variant.value().asArray().get(0).type()).isEqualTo(PhysicalType.STRING);
+    assertThat(variant.value().asArray().get(0).asPrimitive().get()).isEqualTo("hello");
+    assertThat(variant.value().asArray().get(1).type()).isEqualTo(PhysicalType.INT32);
+    assertThat(variant.value().asArray().get(1).asPrimitive().get()).isEqualTo(1);
+  }
+
+  @Test
+  public void testConvertVariantFromMap() {
+    Table table = mock(Table.class);
+    when(table.schema()).thenReturn(VARIANT_SCHEMA);
+    RecordConverter converter = new RecordConverter(table, config);
+
+    Record record = converter.convert(ImmutableMap.of("v", ImmutableMap.of("hello", 1)));
+    Variant variant = (Variant) record.getField("v");
+
+    assertThat(variant).isNotNull();
+    assertThat(variant.metadata()).isNotNull();
+    assertThat(variant.metadata().dictionarySize()).isEqualTo(1);
+    assertThat(variant.metadata().get(0)).isEqualTo("hello");
+    assertThat(variant.value().type()).isEqualTo(PhysicalType.OBJECT);
+    assertThat(variant.value().asObject().numFields()).isEqualTo(1);
+    assertThat(variant.value().asObject().get("hello").type()).isEqualTo(PhysicalType.INT32);
+    assertThat(variant.value().asObject().get("hello").asPrimitive().get()).isEqualTo(1);
+  }
+
+  @Test
+  public void testConvertVariantFromMapNested() {
+    Table table = mock(Table.class);
+    when(table.schema()).thenReturn(VARIANT_SCHEMA);
+    RecordConverter converter = new RecordConverter(table, config);
+
+    Record record =
+        converter.convert(
+            ImmutableMap.of("v", ImmutableMap.of("hello", ImmutableMap.of("world", 1))));
+    Variant variant = (Variant) record.getField("v");
+
+    assertThat(variant).isNotNull();
+    assertThat(variant.value().type()).isEqualTo(PhysicalType.OBJECT);
+    assertThat(variant.metadata()).isNotNull();
+    assertThat(variant.metadata().dictionarySize()).isEqualTo(2);
+    assertThat(variant.metadata().get(0)).isEqualTo("hello");
+    assertThat(variant.value().asObject().numFields()).isEqualTo(1);
+    assertThat(variant.value().asObject().get("hello").type()).isEqualTo(PhysicalType.OBJECT);
+    assertThat(variant.value().asObject().get("hello").asObject().numFields()).isEqualTo(1);
+    assertThat(variant.value().asObject().get("hello").asObject().get("world").type())
+        .isEqualTo(PhysicalType.INT32);
+    assertThat(variant.value().asObject().get("hello").asObject().get("world").asPrimitive().get())
+        .isEqualTo(1);
+  }
+
   public static Map<String, Object> createMapData() {
     return ImmutableMap.<String, Object>builder()
         .put("i", 1)