Kafka Connect: Support VARIANT when record convert by seokyun-ha-toss · Pull Request #15283 · apache/iceberg

seokyun-ha-toss · 2026-02-10T09:11:28Z

Summary

Add support for converting arbitrary Java objects (e.g. Map<String, Object>, lists, temporal types(ex: Date, Instant), primitives) into Iceberg Variant type in the Kafka Connect RecordConverter. Nested maps and lists are converted recursively so that structures like {"user": {"name": "alice", "address": {"city": "Seoul", "zip": "12345"}}} are correctly represented as a single Variant.

Motivation

Kafka Connect payloads often come as schema-less or JSON-like maps. To write them into Iceberg tables with a Variant column, the connector must convert these Java objects into the Variant format (metadata + value) and support nested maps/arrays without losing structure or key names.

Behaviour

Input	Result
Primitives (String, int, long, boolean, etc.)	Single metadata (empty) + corresponding Variant primitive.
Flat map e.g. `{"a": 1, "b": "x"}`	One metadata with keys `["a", "b"]`, one ShreddedObject with two fields.
Nested map e.g. `{"user": {"name": "alice", "address": {"city": "Seoul", "zip": "12345"}}}`	One shared metadata for all keys; root and nested objects as ShreddedObjects with consistent field IDs.
Lists	Converted to VariantArray with elements converted recursively.
Already Variant / ByteBuffer	Pass-through or `Variant.from(ByteBuffer)` where appropriate.
Temporal types (`java.time` and related)	Empty metadata at the root; values map to Variant primitives: `Instant` / `OffsetDateTime` / `ZonedDateTime` → timestamptz; `LocalDateTime` → timestampntz; `LocalDate` → date; `LocalTime` → time (micros/day encoding via `DateTimeUtil`).
`java.util.Date` (Connect runtime)	Same empty metadata when valid; requires the Kafka Connect field schema so the logical type name can distinguish Timestamp, Time, and Date (all represented as `java.util.Date` at runtime). Without a matching logical type, conversion throws. Nested values receive schema from the parent map value schema or struct field schema when present.
Kafka Connect `Struct`	Treated like a keyed object: one shared metadata with all field names collected from the struct tree (including nested maps/structs/lists); the struct becomes a ShreddedObject whose keys are connector field names. Each field is converted recursively with `field.schema()` so nested collections, maps, temporal fields, and inner structs keep correct typing.

Relates

Variant Data Type Support #10392

Thanks, Good Day!

++ Thanks for contribute @brandonstanleyappfolio

…ersion methods for nested structures

danielcweeks · 2026-02-10T20:20:42Z

+  protected Variant convertVariantValue(Object value) {
+    if (value instanceof Variant) {
+      return (Variant) value;
+    } else if (value instanceof ByteBuffer) {


I don't think it is correct to assume this is encoded as a variant. It should be handled as a literal binary.

#15283 (comment)

same as this, handled together

@seokyun-ha-toss I still don't think this has been addressed.

I think it's fine to check and pass through a variant if it is an instance of a variant. I think it's unexpected as I noted in another comment (I suppose this could be done through an SMT and I believe Kafka may be introducing a variant type, but I'm not sure if that's a real scenario).

However, if we see a bytebuffer, I don't see why we would assume it's a variant as opposed to a literal binary. That is what the comment originally was intended to address, but it appears we're still treating it as a variant.

alexkot1394 · 2026-03-11T22:11:02Z

Hi @seokyun-ha-toss,
I've opened a PR that's trying to solve the same issue: #15498
I'm happy to help with your PR to get this change into main. Let me know if you want to collaborate on your PR or where you're at with the review comments above.

seokyun-ha-toss · 2026-03-12T08:12:31Z

Hi @seokyun-ha-toss, I've opened a PR that's trying to solve the same issue: #15498 I'm happy to help with your PR to get this change into main. Let me know if you want to collaborate on your PR or where you're at with the review comments above.

Hi @alexkot1394, thanks for reaching out.

I stepped away for a bit to validate my changes in our production environment.

I took a look at your PR as well. From what I see, it treats Variant as a string primitive, whereas I believe it should be handled recursively to properly construct the Variant structure.

I’d prefer to continue working on this PR and incorporate the review comments as soon as possible.

Thanks again!

…Set for uniqueness

…ordConverter apache#15283 (comment)

… for null, primitive types, lists, maps, and mixed types

…xity

seokyun-ha-toss · 2026-03-13T03:19:47Z

Hello, @danielcweeks, @emkornfield, and @alexkot1394, I'm ready to get a review on this. I've tested this for a month in production, and it works well! I can query the output Iceberg tables with Variant columns using Spark and Snowflake.

Please take a look at the latest updates. Thanks!

brandonstanleyappfolio · 2026-03-17T01:24:17Z

👋 Hi @seokyun-ha-toss, thanks for introducing this change! Is there a reason Struct types aren't included in this PR?

seokyun-ha-toss · 2026-03-17T02:31:04Z

👋 Hi @seokyun-ha-toss, thanks for introducing this change! Is there a reason Struct types aren't included in this PR?

Good catch! I missed the case of org.apache.kafka.connect.data.Struct. I will handle it as well. Thanks!

seokyun-ha-toss · 2026-03-17T10:21:43Z

Hello, @brandonstanleyappfolio. I've added Struct type handling and unit tests in the following two commits:

Thanks for pointing this out!

brandonstanleyappfolio · 2026-03-17T20:06:38Z

@seokyun-ha-toss Thank you - I’ve tested the changes locally, and everything is working as expected!

seokyun-ha-toss · 2026-03-25T05:14:35Z

@danielcweeks @emkornfield Gentle ping — would appreciate your review when you have a moment. Thank you!

…nverter https://docs.oracle.com/en/java/javase/17/docs/api/java.base/java/math/BigDecimal.html#%3Cinit%3E(java.math.BigInteger)

…and Time logical types Kafka Connect represents three distinct temporal logical types using the same Java class (java.util.Date): - org.apache.kafka.connect.data.Timestamp (milliseconds since epoch) - org.apache.kafka.connect.data.Time (milliseconds since midnight) - org.apache.kafka.connect.data.Date (days since epoch) The existing primitiveToVariantValue method only checked `instanceof Date` with no schema context, converting all java.util.Date values to Iceberg DATE (days since epoch). This silently discarded the time component from Timestamp fields (e.g. 2025-12-09T14:30:45.123Z became 2025-12-09). Fix: thread the Kafka Connect schema through objectToVariantValue and primitiveToVariantValue so the Date branch can inspect the schema's logical type name and convert to the correct Iceberg variant type: - Timestamp (logical name: org.apache.kafka.connect.data.Timestamp) -> Variants.ofTimestamptz (microseconds since epoch) - Time (logical name: org.apache.kafka.connect.data.Time) -> Variants.ofTime (microseconds since midnight) - Date (logical name: org.apache.kafka.connect.data.Date) -> Variants.ofDate (days since epoch) The schema is propagated from Struct fields (field.schema()), Collection elements (schema.valueSchema()), and Map values (schema.valueSchema()). When no schema is available, an IllegalArgumentException is thrown to prevent silent data loss.

Fix `java.util.Date` variant conversion losing precision for Timestamp and Time logical types

seokyun-ha-toss · 2026-04-04T08:14:54Z

Hello, @emkornfield @brandonstanleyappfolio @danielcweeks @alexkot1394, all temporal types are handled in this PR, and I re-arrange the unittest to checking multiple primitive, mixed types and nested values. Please, take a look at 🙏 Many thanks!

seokyun-ha-toss · 2026-04-13T01:43:42Z

Good morning, @emkornfield . I've applied all the feedback and commented:

"throw an exception for non-string/empty keys on Map type."
LocalDateTime is treated as micro-seconds across the repo.

So please take a look again 🙏

danielcweeks · 2026-04-13T15:35:02Z


+  protected Variant convertVariantValue(Object value) {
+    if (value instanceof ByteBuffer) {
+      return Variant.from((ByteBuffer) value);


@seokyun-ha-toss I still don't think this has been addressed.

I think it's fine to check and pass through a variant if it is an instance of a variant. I think it's unexpected as I noted in another comment (I suppose this could be done through an SMT, but I'm not sure if that's a real scenario).

However, if we see a bytebuffer, I don't see why we would assume it's a variant as opposed to a literal binary. That is what the comment originally was intended to address, but it appears we're still treating it as a variant.

danielcweeks · 2026-04-13T15:37:08Z

+    Set<String> fieldNames = Sets.newHashSet();
+    collectFieldNames(value, fieldNames);


Why not have collectFiledNames return the Set<String> as opposed to creating it here and passing it in?

danielcweeks · 2026-04-13T15:54:29Z

+      Map<?, ?> map = (Map<?, ?>) value;
+      for (Map.Entry<?, ?> entry : map.entrySet()) {
+        Object key = entry.getKey();
+        if (key != null && key instanceof String) {


key != null && is unnecessary here

danielcweeks · 2026-04-13T16:03:40Z

+        schema != null ? schema.valueSchema() : null;
+    map.forEach(
+        (key, val) -> {
+          if (key != null && key instanceof String) {


key != null && redundant here as well

danielcweeks · 2026-04-13T16:04:45Z

+      return mapToVariantValue(value, metadata, schema);
+    }
+    if (value instanceof Struct) {
+      Struct struct = (Struct) value;


We're now on Java 17, so this should be a pattern variable.

danielcweeks · 2026-04-13T16:05:01Z

+      return;
+    }
+    if (value instanceof Struct) {
+      Struct struct = (Struct) value;


We're now on Java 17, so this should be a pattern variable.

seokyun-ha-toss added 2 commits February 10, 2026 17:17

feat: Implement support for VARIANT type in RecordConverter with conv…

033f47f

…ersion methods for nested structures

test: implement unit test for RecordConverter for Variant

a53af92

github-actions bot added the KAFKACONNECT label Feb 10, 2026

chore: lint spotlessApply

73ea656

emkornfield reviewed Feb 10, 2026

View reviewed changes

danielcweeks reviewed Feb 10, 2026

View reviewed changes

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

danielcweeks reviewed Feb 10, 2026

View reviewed changes

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

fix: there's no already Variant value, and use Lists.of()

2b3fcb1

danielcweeks mentioned this pull request Mar 11, 2026

Kafka Connect: Add variant type support #15498

Closed

seokyun-ha-toss added 8 commits March 13, 2026 07:24

fix: handle BigDecimal type in numberToVariantValue()

47f6fe5

fix: explicit error when unknown numeric type

e8f57f1

refactor: optimize field name collection in RecordConverter to use a …

b57e0d8

…Set for uniqueness

fix: ensure keys in map are non-null strings before processing in Rec…

0117174

…ordConverter apache#15283 (comment)

refactor: loop entrySet once

be09349

test: enhance unit tests for Variant conversion with additional cases…

75ec7ce

… for null, primitive types, lists, maps, and mixed types

lint: apply gh actions auto-review and lints, split Cyclomatic Comple…

c6e558a

…xity

lint: spotlessApply

787663d

seokyun-ha-toss added 3 commits March 17, 2026 18:53

feat: support kafka Struct type, and align if checking ordering

75dc8a2

test: implt unittest for Variant for Struct

d9c8721

lint: gradlew spotlessApply

e517884

emkornfield reviewed Mar 25, 2026

View reviewed changes

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

brandonstanleyappfolio reviewed Mar 26, 2026

View reviewed changes

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

seokyun-ha-toss added 4 commits March 27, 2026 13:07

feat: add support for BigInteger conversion to BigDecimal in RecordCo…

8bb8740

…nverter https://docs.oracle.com/en/java/javase/17/docs/api/java.base/java/math/BigDecimal.html#%3Cinit%3E(java.math.BigInteger)

feat: support for Date conversion to Variants

aa2bcce

feat: support additional Date, Datetime, time types

5be9fa4

lint: spotlessApply

179b07a

emkornfield reviewed Mar 30, 2026

View reviewed changes

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

brandonstanleyappfolio and others added 11 commits March 31, 2026 18:16

Merge pull request #2 from brandonstanleyappfolio/variant-date-fix

465fa92

Fix `java.util.Date` variant conversion losing precision for Timestamp and Time logical types

fix: explict throw exception when non-string key is used

06b7afe

lint: spotlessApply

a4c12b8

test: remove duplicated

568f7b1

test: integrate test convert variant value from list, map, struct

10bff3d

test: left only one unittest on VariantValueFromStructWith Date family

b8772ba

test: add unittest for variant with timestamp family

a99262e

Merge branch 'apache:main' into support-variant-for-sink-connector

ca47edf

fix: [JavaUtilDate] Date has a bad API that leads to bugs

7530597

fix: resolve Cyclomatic Complexity

a312eb0

emkornfield reviewed Apr 5, 2026

View reviewed changes

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

emkornfield reviewed Apr 5, 2026

View reviewed changes

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

danielcweeks self-requested a review April 6, 2026 00:41

seokyun-ha-toss added 2 commits April 9, 2026 12:48

fix: throw invalid key type

818a515

lint: spotlessApply

b1f379e

danielcweeks reviewed Apr 13, 2026

View reviewed changes

		Set<String> fieldNames = Sets.newHashSet();
		collectFieldNames(value, fieldNames);

Conversation

seokyun-ha-toss commented Feb 10, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary

Motivation

Behaviour

Relates

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

seokyun-ha-toss Feb 12, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

alexkot1394 commented Mar 11, 2026

Uh oh!

seokyun-ha-toss commented Mar 12, 2026

Uh oh!

seokyun-ha-toss commented Mar 13, 2026

Uh oh!

brandonstanleyappfolio commented Mar 17, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

seokyun-ha-toss commented Mar 17, 2026

Uh oh!

seokyun-ha-toss commented Mar 17, 2026

Uh oh!

brandonstanleyappfolio commented Mar 17, 2026

Uh oh!

seokyun-ha-toss commented Mar 25, 2026

Uh oh!

Uh oh!

Uh oh!

Uh oh!

seokyun-ha-toss commented Apr 4, 2026

Uh oh!

Uh oh!

Uh oh!

seokyun-ha-toss commented Apr 13, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

seokyun-ha-toss commented Feb 10, 2026 •

edited

Loading

seokyun-ha-toss Feb 12, 2026 •

edited

Loading

brandonstanleyappfolio commented Mar 17, 2026 •

edited

Loading