Flink: Fix dynamic sink HASH keying of records sent to the table updater by zhengxiang378928908-code · Pull Request #18602 · apache/iceberg

zhengxiang378928908-code · 2026-10-21T17:22:01Z

With immediateUpdate disabled, a record that needs a table update gets its writer key from the table spec (or table schema id) but its raw row, which follows the record's own schema. When field IDs or column order differ, the partition source is read from the wrong position and the job fails with a ClassCastException. The key now uses the record's own schema, spec and row, which always describe the same layout (simplified as suggested in review). Unlike #17685 this leaves the main path alone: its row is already in the table layout, so keying it with the record schema breaks reordered columns (testHashDistributionWithReorderedColumns).

Fixes #17651.

Test plan: three new TestDynamicIcebergSink cases and one TestHashKeyGenerator case. testHashDistributionWithAlternatingSchemaVariants and testHashDistributionToNewBranchWithDataConversion fail on main with the reported ClassCastException and pass with this change; testHashDistributionWithReorderedColumns passes on both and guards the main path. testCacheHitForEqualSchemaInstances checks that equal schemas passed as new instances share one key selector. The 322 tests in org.apache.iceberg.flink.sink.dynamic on flink/v2.3 run with no failures (2 skipped); spotlessCheck, checkstyleMain and checkstyleTest pass. Backports to v1.20/v2.2 can follow once this is agreed.

AI Disclosure

Model: Claude Opus 5.5
Platform/Tool: Claude Code
Human Oversight: reviewed. I went through the fix and the tests and checked the root cause against the failing tests on main.
Generated parts: the reproduction tests, the fix in DynamicRecordProcessor and this description were drafted with AI and run locally. Uncertainty: the stack trace in Flink: Dynamic Sink — ClassCastException in HashKeyGenerator partition keying when records for one table alternate between writer schema variants #17651 (1.10.1) points at emit(), not the updater path fixed here, so the reporter's exact setup (in particular immediateUpdate) is not confirmed.

With immediateUpdate disabled, a record that needs a table update was keyed with the table spec or the table schema id but with its raw row, which is laid out by the record's own schema. When the field IDs or column order of the record differ from the table, the partition source is read from the wrong position and the job fails with a ClassCastException (apache#17651). Key such records in the table's layout only when both the table schema and spec are known, converting the row first, and otherwise with the record's own schema, spec and row. Generated-by: Claude Code (Claude Opus 5.5) Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

pvary · 2026-10-22T11:26:57Z

-                    : data.schema(),
-                foundSpec != null ? foundSpec : data.spec(),
-                data.rowData());
+            keyByTable


What if we just leave out the table schema and friends?

int writerKey = hashKeyGenerator.generateKey(data, null, null, null);

Good point, done in 2650060. The table schema and spec are only both known here when just the branch is missing, so the extra case wasn't worth a second conversion of the row. The 322 tests in org.apache.iceberg.flink.sink.dynamic still run with no failures (2 skipped).

One small difference, for the record: when a record goes to a new branch and needs data conversion, the update path now keys it with the record's types, while emit() keys later records with the table's types. With equality-field distribution, an int key column written to a long table column hashes differently for negative values (Integer.hashCode vs Long.hashCode), so in upsert mode rows with the same key values may go to different writers, depending on the write parallelism. The same can already happen when a schema or spec update is needed. I haven't reproduced this end to end and didn't keep a special case for it.

As suggested in review, key records sent to the table updater with the record's own schema, spec and row in all cases. The table schema and spec are known here only when just the branch is missing, so the separate table-layout case is not worth a second conversion of the row. Generated-by: Claude Code (Claude Opus 5.5) Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01BynXTKFbZXCrfmXQq5xnyi

mxm

Thanks @zhengxiang378928908-code! Some minor comments, but all in all looks good.

mxm · 2026-10-26T09:13:15Z

+    for (int i = 0; i < 6; i++) {
+      boolean useVariantA = i % 2 == 0;
+      DynamicIcebergDataImpl row =
+          new DynamicIcebergDataImpl(
+              useVariantA ? variantA : variantB,
+              tableName,
+              SnapshotRef.MAIN_BRANCH,
+              useVariantA ? specA : specB);
+      row.rowProvided =
+          useVariantA
+              ? Row.of("a" + i, LocalDate.of(2026, 10, 1 + i))
+              : Row.of("b" + i, "x" + i, LocalDate.of(2026, 10, 1 + i));
+      rows.add(row);
+    }


Do we need all those iterations? Can we spell out the ones we need?

Done in 198bab6: the test now writes four explicit records, alternating A, B, A, B. Without the fix it still fails with the ClassCastException from #17651.

mxm · 2026-10-26T09:14:14Z

+    for (int i = 0; i < 4; i++) {
+      DynamicIcebergDataImpl row =
+          new DynamicIcebergDataImpl(reordered, tableName, "b1", reorderedSpec);
+      row.rowProvided = Row.of(LocalDate.of(2026, 10, 1 + i), "r" + i);
+      rows.add(row);
+    }


Same here, let's be explicit about the records.

Done, two explicit records. It still fails without the fix.

mxm · 2026-10-26T09:14:21Z

+    for (int i = 0; i < 4; i++) {
+      DynamicIcebergDataImpl row =
+          new DynamicIcebergDataImpl(reordered, tableName, SnapshotRef.MAIN_BRANCH, reorderedSpec);
+      row.rowProvided = Row.of(LocalDate.of(2026, 10, 1 + i), "r" + i);
+      rows.add(row);
+    }


Done, two explicit records.

mxm · 2026-10-26T09:16:51Z

+  void testHashDistributionWithAlternatingSchemaVariantsImmediateUpdate() throws Exception {
+    runAlternatingSchemaVariants("alternating_variants_immediate", true);
+  }


Do we need this test? Immediate updates are already well-tested.

Removed. The remaining test runs with immediateUpdate = false, which is the path this PR changes.

mxm · 2026-10-26T09:18:08Z

+    try (CloseableIterable<Record> records = IcebergGenerics.read(table).build()) {
+      actual = Lists.newArrayList(records);
+    }
+    assertThat(actual).hasSize(6);


Once we reduce the number of input records, can we check the actual values?

Done. All three tests now check every written row by id and event_date (and extra in the alternating test).

mxm · 2026-10-26T09:20:54Z

+        // Shuffled records with immediateUpdate=false go to the update side output.
+        // Key them with the record's own schema, spec and row: the raw row is laid out by the
+        // record's schema, so the table's schema or spec cannot be used with it.
+        int writerKey = hashKeyGenerator.generateKey(data, null, null, null);


What if the user always passes a new schema? We key by the schema reference which will create a new key selector for every record. Admittedly, this issue was there before but is now more pronounced.

Good catch. On this path it is actually new with this PR: before, the update path always passed a non-null schema (the resolved table schema when there was one, otherwise data.schema()), so the cache key used that schema's id rather than the object. With null the key holds the record's Schema itself, and Schema has no equals, so every record with a new Schema instance built a new key selector. Besides the cost, a fresh round-robin selector always starts at the same position, so those records would all go to one writer.

198bab6 keys the cache by schema.asStruct() instead, which has value equality. Identifier fields are already part of the key through the resolved equality fields. This only applies when no table schema id is passed, which is the update path; emit() still keys by the table schema id. testCacheHitForEqualSchemaInstances covers it.

- Spell out the test records and check the written values; drop the immediate-update variant of the alternating-schema test. - Compare the record schema by struct in the key selector cache, so a new Schema instance per record no longer creates a new key selector on the update path. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

mxm

LGTM, just a minor NIT.

mxm · 2026-10-27T15:18:19Z

+        // Shuffled records with immediateUpdate=false go to the update side output.
+        // Key them with the record's own schema, spec and row: the raw row is laid out by the
+        // record's schema, so the table's schema or spec cannot be used with it.
+        int writerKey = hashKeyGenerator.generateKey(data, null, null, null);


This could be replaced with

Suggested change

int writerKey = hashKeyGenerator.generateKey(data, null, null, null);

int writerKey = hashKeyGenerator.generateKey(data);

Done in f0e3b72. The single-argument generateKey declared throws Exception, so I dropped that clause as well; it only delegates to the overload that doesn't throw.

mxm · 2026-10-27T15:35:56Z

+   * is keyed.
+   */
+  @Test
+  void testHashDistributionWithReorderedColumns() throws Exception {


Not sure we need this test because it doesn't test the update path, but it doesn't hurt.

Kept it as a regression test for the main path with reordered columns; happy to drop it if you prefer.

Drop the unneeded throws clause from HashKeyGenerator#generateKey(DynamicRecord), which only delegates to the overload that does not throw. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

openinx · 2026-10-28T02:47:33Z

      this.schemaId = tableSchemaId;
      this.specId = tableSpecId;
-      this.schema = tableSchemaId == null ? schema : null;
+      this.schema = tableSchemaId == null ? schema.asStruct() : null;


(unrelated): this is a little bit strange, why do we need the tableSchemId to decide the schema existence?

It decides how the schema is represented in the cache key. When a resolved table schema is passed, its ID identifies the schema within the table. Otherwise the key holds the input schema's struct.

In emit, the row is first converted to the resolved table schema, so records with different input schemas can share a selector once they map to the same table schema (and the rest of the key — spec, equality fields, distribution mode, parallelism — matches).

The deferred update path calls generateKey(record) with the record's original schema, spec and row, so the input schema has to be in the key. Comparing by asStruct() lets structurally equal schema instances reuse the selector.

The schema id is used when we already have a corresponding table schema (immediate update). The line above is for the non-immediate path where we do not have a table schema yet.

openinx · 2026-10-28T02:57:36Z

+        // Shuffled records with immediateUpdate=false go to the update side output.
+        // Key them with the record's own schema, spec and row: the raw row is laid out by the
+        // record's schema, so the table's schema or spec cannot be used with it.
+        int writerKey = hashKeyGenerator.generateKey(data);


Can you explain a bit more why does this change solve the bug mentioned in #17651?

The change will always use schema and spec from the record and not from table schemas / specs. The issue before was that it could mix record schema / spec with the corresponding table schema / spec, which leads to undefined behavior.

mxm · 2026-10-28T10:35:40Z

Thanks @zhengxiang378928908-code for the PR! Thanks @pvary @openinx for reviewing!

@zhengxiang378928908-code Could you create the backport PR for the other Flink versions?

zhengxiang378928908-code mentioned this pull request Oct 21, 2026

fix: use record schema for partition key selector in HashKeyGenerator #17685

Closed

github-actions Bot added the flink label Oct 21, 2026

pvary reviewed Oct 22, 2026

View reviewed changes

pvary approved these changes Oct 22, 2026

View reviewed changes

mxm reviewed Oct 26, 2026

View reviewed changes

mxm approved these changes Oct 27, 2026

View reviewed changes

Flink: Use the single-argument generateKey on the update path

f0e3b72

Drop the unneeded throws clause from HashKeyGenerator#generateKey(DynamicRecord), which only delegates to the overload that does not throw. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

openinx reviewed Oct 28, 2026

View reviewed changes

mxm merged commit b61ef96 into apache:main Oct 28, 2026
26 checks passed

	int writerKey = hashKeyGenerator.generateKey(data, null, null, null);
	int writerKey = hashKeyGenerator.generateKey(data);

Conversation

zhengxiang378928908-code commented Oct 21, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mxm left a comment

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!

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!

Choose a reason for hiding this comment

Uh oh!

mxm left a comment

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!

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!

Uh oh!

mxm commented Oct 28, 2026

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

zhengxiang378928908-code commented Oct 21, 2026 •

edited

Loading