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,218 @@ 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, null);
+    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, org.apache.kafka.connect.data.Schema schema) {
+    if (value == null) {
+      return Variants.ofNull();
+    }
+    VariantValue primitive = primitiveToVariantValue(value, schema);
+    if (primitive != null) {
+      return primitive;
+    }
+    if (value instanceof Collection) {
+      ValueArray array = Variants.array();
+      org.apache.kafka.connect.data.Schema elementSchema =
+          schema != null ? schema.valueSchema() : null;
+      for (Object element : (Collection<?>) value) {
+        array.add(objectToVariantValue(element, metadata, elementSchema));
+      }
+      return array;
+    }
+    if (value instanceof Map) {
+      return mapToVariantValue(value, metadata, schema);
+    }
+    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, field.schema()));
+      }
+      return object;
+    }
+    throw new IllegalArgumentException("Cannot convert to variant: " + value.getClass().getName());
+  }
+
+  private static VariantValue mapToVariantValue(
+      Object value, VariantMetadata metadata, org.apache.kafka.connect.data.Schema schema) {
+    Map<?, ?> map = (Map<?, ?>) value;
+    ShreddedObject object = Variants.object(metadata);
+    org.apache.kafka.connect.data.Schema mapValueSchema =
+        schema != null ? schema.valueSchema() : null;
+    map.forEach(
+        (key, val) -> {
+          if (key != null && key instanceof String) {
+            object.put((String) key, objectToVariantValue(val, metadata, mapValueSchema));
+          } else {
+            throw new IllegalArgumentException(
+                "Cannot convert map to variant: keys must be non-null strings, was: "
+                    + (key == null ? "null" : key.getClass().getName()));
+          }
+        });
+    return object;
+  }
+
+  /**
+   * Converts a primitive or primitive-like value to VariantValue; returns null if not supported.
+   * The optional schema is used to disambiguate java.util.Date which Kafka Connect uses for Date,
+   * Time, and Timestamp logical types.
+   */
+  private static VariantValue primitiveToVariantValue(
+      Object value, org.apache.kafka.connect.data.Schema schema) {
+    if (value instanceof Boolean) {
+      return Variants.of((Boolean) value);
+    }
+    VariantValue temporal = temporalObjectToVariantValue(value, schema);
+    if (temporal != null) {
+      return temporal;
+    }
+    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;
+  }
+
+  /**
+   * Converts java.time values and java.util.Date (with Connect logical type from the optional
+   * schema) to VariantValue; returns null if the value is not a supported temporal representation.
+   */
+  private static VariantValue temporalObjectToVariantValue(
+      Object value, org.apache.kafka.connect.data.Schema schema) {
+    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) {
+      String logicalName = schema != null ? schema.name() : null;
+      // Connect represents Timestamp, Time, and Date logical types as java.util.Date at runtime;
+      // normalize to Instant once, then interpret using the schema logical type name.
+      Instant connectInstant = ((Date) value).toInstant();
+      if (org.apache.kafka.connect.data.Timestamp.LOGICAL_NAME.equals(logicalName)) {
+        return Variants.ofTimestamptz(DateTimeUtil.microsFromInstant(connectInstant));
+      }
+      if (org.apache.kafka.connect.data.Time.LOGICAL_NAME.equals(logicalName)) {
+        LocalTime utcTime = connectInstant.atZone(ZoneOffset.UTC).toLocalTime();
+        return Variants.ofTime(DateTimeUtil.microsFromTime(utcTime));
+      }
+      if (org.apache.kafka.connect.data.Date.LOGICAL_NAME.equals(logicalName)) {
+        return Variants.ofDate(DateTimeUtil.daysFromInstant(connectInstant));
+      }
+      throw new IllegalArgumentException(
+          "Cannot convert java.util.Date to variant without a recognized logical type schema"
+              + " (expected Timestamp, Time, or Date but got: "
+              + logicalName
+              + ")");
+    }
+    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) {