Kafka Connect: Support VARIANT when record convert

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

@@ -22,23 +22,28 @@
 import java.io.IOException;
 import java.io.UncheckedIOException;
 import java.math.BigDecimal;
+import java.math.BigInteger;
 import java.math.RoundingMode;
 import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
+import java.time.Instant;
 import java.time.LocalDate;
 import java.time.LocalDateTime;
 import java.time.LocalTime;
 import java.time.OffsetDateTime;
 import java.time.ZoneOffset;
+import java.time.ZonedDateTime;
 import java.time.format.DateTimeFormatter;
 import java.time.format.DateTimeFormatterBuilder;
 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;
 import java.util.Map;
+import java.util.Set;
 import java.util.UUID;
 import java.util.stream.Collectors;
 import org.apache.iceberg.FileFormat;
@@ -53,6 +58,7 @@
 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.Maps;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
 import org.apache.iceberg.types.Type;
 import org.apache.iceberg.types.Type.PrimitiveType;
 import org.apache.iceberg.types.Types.DecimalType;
@@ -64,6 +70,13 @@
 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.Field;
 import org.apache.kafka.connect.data.Struct;
 import org.apache.kafka.connect.errors.ConnectException;
 
@@ -142,6 +155,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 +479,171 @@ protected Temporal convertTimestampValue(Object value, TimestampType type) {
     return convertLocalDateTime(value);
   }
 
+  protected Variant convertVariantValue(Object value) {
+    if (value instanceof ByteBuffer) {
+      return Variant.from((ByteBuffer) value);
+    }
+
+    Set<String> fieldNames = Sets.newHashSet();
+    collectFieldNames(value, fieldNames);
+    List<String> allFieldNames = fieldNames.stream().sorted().collect(Collectors.toList());
+    VariantMetadata metadata = Variants.metadata(allFieldNames);
+    VariantValue variantValue = objectToVariantValue(value, metadata);
+    return Variant.of(metadata, variantValue);
+  }
+
+  /**
+   * Collects all field names (map keys) from the entire object tree into the given set. Used to
+   * build a single VariantMetadata for the whole Variant (required for nested maps).
+   */
+  private static void collectFieldNames(Object value, Set<String> names) {
+    if (value == null) {
+      return;
+    }
+    if (value instanceof Collection) {
+      for (Object element : (Collection<?>) value) {
+        collectFieldNames(element, names);
+      }
+      return;
+    }
+    if (value instanceof Map) {
+      Map<?, ?> map = (Map<?, ?>) value;
+      for (Map.Entry<?, ?> entry : map.entrySet()) {
+        Object key = entry.getKey();
+        if (key != null && key instanceof String) {
+          names.add((String) key);
+          collectFieldNames(entry.getValue(), names);
+        }
+      }
+      return;
+    }
+    if (value instanceof Struct) {
+      Struct struct = (Struct) value;
+      for (Field field : struct.schema().fields()) {
+        names.add(field.name());
+        collectFieldNames(struct.get(field), names);
+      }
+    }
+  }
+
+  /**
+   * 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();
+    }
+    VariantValue primitive = primitiveToVariantValue(value);
+    if (primitive != null) {
+      return primitive;
+    }
+    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(
+          (key, val) -> {
+            if (key != null && key instanceof String) {
+              object.put((String) key, objectToVariantValue(val, metadata));
+            }
+          });
+      return object;
+    }
+    if (value instanceof Struct) {
+      Struct struct = (Struct) value;
+      ShreddedObject object = Variants.object(metadata);
+      for (Field field : struct.schema().fields()) {
+        object.put(field.name(), objectToVariantValue(struct.get(field), metadata));
+      }
+      return object;
+    }
+    throw new IllegalArgumentException("Cannot convert to variant: " + value.getClass().getName());
+  }
+
+  /**
+   * Converts a primitive or primitive-like value to VariantValue; returns null if not supported.
+   */
+  private static VariantValue primitiveToVariantValue(Object value) {
+    if (value instanceof Boolean) {
+      return Variants.of((Boolean) value);
+    }
+    if (value instanceof Instant) {
+      return Variants.ofTimestamptz(DateTimeUtil.microsFromInstant((Instant) value));
+    }
+    if (value instanceof OffsetDateTime) {
+      return Variants.ofTimestamptz(DateTimeUtil.microsFromTimestamptz((OffsetDateTime) value));
+    }
+    if (value instanceof ZonedDateTime) {
+      return Variants.ofTimestamptz(
+          DateTimeUtil.microsFromTimestamptz(((ZonedDateTime) value).toOffsetDateTime()));
+    }
+    if (value instanceof LocalDateTime) {
+      return Variants.ofTimestampntz(DateTimeUtil.microsFromTimestamp((LocalDateTime) value));
+    }
+    if (value instanceof LocalDate) {
+      return Variants.ofDate(DateTimeUtil.daysFromDate((LocalDate) value));
+    }
+    if (value instanceof LocalTime) {
+      return Variants.ofTime(DateTimeUtil.microsFromTime((LocalTime) value));
+    }
+    if (value instanceof Date) {
+      int days = (int) (((Date) value).getTime() / 1000 / 60 / 60 / 24);
+      return Variants.ofDate(days);
+    }
+    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 UUID) {
+      return Variants.ofUUID((UUID) value);
+    }
+    return null;
+  }
+
+  private static VariantValue numberToVariantValue(Number value) {
+    if (value instanceof BigDecimal) {
+      return Variants.of((BigDecimal) value);
+    }
+    if (value instanceof BigInteger) {
+      return Variants.of(new BigDecimal((BigInteger) 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);
+    }
+    throw new IllegalArgumentException(
+        "Cannot convert Number to variant (unknown type): " + value.getClass().getName());
+  }
+
   @SuppressWarnings("JavaUtilDate")
   private OffsetDateTime convertOffsetDateTime(Object value) {
     if (value instanceof Number) {