Flink: Dynamic Iceberg Sink: Add HashKeyGenerator / RowDataEvolver / TableUpdateOperator #13277

mxm · 2025-06-08T12:53:41Z

This change adds the following components for the Flink Dynamic Iceberg Sink:

HashKeyGenerator

A hash key generator which will be used in DynamicIcebergSink class (next PR) to implement one of Iceberg's DistributionModes (NONE, HASH, RANGE).

The HashKeyGenerator is responsible for creating the appropriate hash key for Flink's keyBy operation. The hash key is generated depending on the user-provided DynamicRecord and the table metadata. Under the hood, we maintain a set of Flink KeySelectors which implement the appropriate Iceberg DistributionMode. For every table, we randomly select a consistent subset of writer subtasks which receive data via their associated keys, depending on the chosen DistributionMode.

Caching ensures that a new key selector is also created when the table metadata (e.g. schema, spec) or the user-provided metadata changes (e.g. distribution mode, write parallelism).

RowDataEvolver

RowDataEvolver is responsible to change the input RowData to make it compatible with the target schema. This is done when:

The input schema has fewer fields than the target schema.
The table types are wider than the input type.
The field order differs for source and target schema.

The resolution is as follows:

In the first case, we would add a null values for the missing field (if the field is optional). In the second case, we would convert the data for the input field to a wider type, e.g. int (input type) => long (table type). In the third case, we would rearrange the input data to match the target table.

DynamicUpdateOperator

A dedicated operator to updating the schema / spec for the table associated with a DynamicRecord.

…TableUpdateOperator This change adds the following components for the Flink Dynamic Iceberg Sink: *** HashKeyGenerator A hash key generator which will be used in DynamicIcebergSink class (next PR) to implement one of Iceberg's DistributionModes (NONE, HASH, RANGE). The HashKeyGenerator is responsible for creating the appropriate hash key for Flink's keyBy operation. The hash key is generated depending on the user-provided DynamicRecord and the table metadata. Under the hood, we maintain a set of Flink {@link KeySelector}s which implement the appropriate Iceberg {@link DistributionMode}. For every table, we randomly select a consistent subset of writer subtasks which receive data via their associated keys, depending on the chosen DistributionMode. Caching ensures that a new key selector is also created when the table metadata (e.g. schema, spec) or the user-provided metadata changes (e.g. distribution mode, write parallelism). *** RowDataEvolver RowDataEvolver is responsible to change the input RowData to make it compatible with the target schema. This is done when: 1. The input schema has fewer fields than the target schema. 2. The table types are wider than the input type. 3. The field order differs for source and target schema. The resolution is as follows: In the first case, we would add a null values for the missing field (if the field is optional). In the second case, we would convert the data for the input field to a wider type, e.g. int (input type) => long (table type). In the third case, we would rearrange the input data to match the target table. *** DynamicUpdateOperator A dedicated operator to updating the schema / spec for the table associated with a DynamicRecord.

flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/NonThrowingKeySelector.java

flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicSinkUtil.java

pvary · 2025-06-10T08:07:56Z

flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicSinkUtil.java

+
+  private DynamicSinkUtil() {}
+
+  static List<Integer> getEqualityFieldIds(List<String> equalityFields, Schema schema) {


It is a bit strange to me to ignore the empty equalityFields.
Maybe a javadoc to highlight this?

What do you mean by ignoring empty equalityFields? There may be none defined.

If I understand correctly, if the equalityFields is empty, then we fall back to the Schema defined stuff.
I feel that if (equalityFields==null) then Schema is natural, but handling empty list as not set is a bit strange.
Minimally we need to document it

Can we have at least a javadoc for this?

pvary · 2025-06-10T08:09:29Z

....0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicTableUpdateOperator.java

+    Catalog catalog = catalogLoader.loadCatalog();
+    this.updater =
+        new TableUpdater(
+            new TableMetadataCache(catalog, cacheMaximumSize, cacheRefreshMs), catalog);


Could we share the TableMetadataCache in a static way, so it is shared between the operator instances?

Yes, possible. Let me look into it.

I would like to defer this change to a follow-up. There are some challenges with regard to concurrent get / put operations on the cache.

....0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicTableUpdateOperator.java

flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/HashKeyGenerator.java

pvary · 2025-06-10T08:17:00Z

flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/HashKeyGenerator.java

+          }
+        }
+
+      case RANGE:


Will the RANGE distribution mode work with the dyntable sink?
Range needed statistics collection and complicated infrastructure to work. It relied on Partitioners to direct the records to the correct writer.

For now we fall back to HASH. Adding support for RANGE in the Dynamic Sink will be a follow-up.

Sure - leave a comment/TODO

flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/HashKeyGenerator.java

pvary · 2025-06-10T08:21:45Z

flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/HashKeyGenerator.java

+    private final Integer specId;
+    private final Schema schema;
+    private final PartitionSpec spec;
+    private final List<String> equalityFields;


Should this be a Set?

Perhaps. Need to double-check.

So far, we are always using List for equality fields (IcebergSink V1 / V2). It doesn't seem though that this is required. Updated everywhere. See 0412697.

There is only a single place where this touches a public API.
Maybe for backward compatibility we could use Collection there. Or we just deprecate the old methods, add a new one (only ones which are currently used)
Seems like a good change to me, but I would ask @stevenzwu. He might know more why the equalityFields are stored in a List instead of a Set.

flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/RowDataEvolver.java

...link/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicTableUpdateOperator.java

pvary · 2025-06-10T11:14:29Z

flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/RowDataTaskWriterFactory.java

@@ -60,7 +60,7 @@ public RowDataTaskWriterFactory(
      long targetFileSizeBytes,
      FileFormat format,
      Map<String, String> writeProperties,
-      List<Integer> equalityFieldIds,
+      Set<Integer> equalityFieldIds,


This is public API, since we don't have annotation on this.

Ok. I've reverted this change in favor of an internal conversion. See 05fd580

flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/HashKeyGenerator.java

…riter

pvary

+1 pending tests

pvary · 2025-06-11T15:53:13Z

Merged to main.
Thanks for the PR @mxm!

mxm · 2025-06-11T16:04:10Z

Thank for reviewing / merging @pvary 🙏

mxm · 2025-06-11T16:13:30Z

I'll prepare the backports.

…volver / TableUpdateOperator to Flink 1.19 / 1.20 (#13303) backports #13277

github-actions bot added the flink label Jun 8, 2025

mxm mentioned this pull request Jun 8, 2025

Flink: Dynamic Iceberg Sink Contribution #12424

Open

pvary reviewed Jun 10, 2025

View reviewed changes

flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/NonThrowingKeySelector.java Outdated Show resolved Hide resolved

pvary reviewed Jun 10, 2025

View reviewed changes

flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicSinkUtil.java Show resolved Hide resolved

pvary reviewed Jun 10, 2025

View reviewed changes

....0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicTableUpdateOperator.java Show resolved Hide resolved

pvary reviewed Jun 10, 2025

View reviewed changes

flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/HashKeyGenerator.java Show resolved Hide resolved

pvary reviewed Jun 10, 2025

View reviewed changes

flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/HashKeyGenerator.java Outdated Show resolved Hide resolved

pvary reviewed Jun 10, 2025

View reviewed changes

flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/HashKeyGenerator.java Outdated Show resolved Hide resolved

pvary reviewed Jun 10, 2025

View reviewed changes

flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/HashKeyGenerator.java Show resolved Hide resolved

pvary reviewed Jun 10, 2025

View reviewed changes

flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/HashKeyGenerator.java Show resolved Hide resolved

pvary reviewed Jun 10, 2025

View reviewed changes

flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/RowDataEvolver.java Show resolved Hide resolved

pvary reviewed Jun 10, 2025

View reviewed changes

...link/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicTableUpdateOperator.java Show resolved Hide resolved

mxm added 9 commits June 10, 2025 11:40

Remove NonThrowingKeySelector

871bb09

Adjust logging

1b326a7

JavaDoc for DynamicTableUpdateOperator

dc410e8

JavaDoc Deterministic hashing

cd2d825

More logging context

36cfeda

Assert subtask id (slotid)

bfdbe5a

Change equalityFields from List to Set

0412697

newline

3d4ccd7

verify duplicate updates

78fe682

pvary reviewed Jun 10, 2025

View reviewed changes

flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/HashKeyGenerator.java Outdated Show resolved Hide resolved

pvary reviewed Jun 10, 2025

View reviewed changes

flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/HashKeyGenerator.java Outdated Show resolved Hide resolved

pvary reviewed Jun 10, 2025

View reviewed changes

flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/HashKeyGenerator.java Show resolved Hide resolved

mxm added 4 commits June 11, 2025 10:18

Revert public API change

05fd580

Fix logging

f9d2b67

Log table name in TargetLimitedKeySelector

bcbaa71

Use Set for equalityFields in BaseDeltaTaskWriter / PartitionedDeltaW…

303d3b4

…riter

pvary approved these changes Jun 11, 2025

View reviewed changes

pvary merged commit 76972ef into apache:main Jun 11, 2025
18 checks passed

mxm added a commit to mxm/iceberg that referenced this pull request Jun 12, 2025

Flink: Backport apache#13277 to Flink 1.19

c951a05

mxm added a commit to mxm/iceberg that referenced this pull request Jun 12, 2025

Flink: Backport apache#13277 to Flink 1.20

3f45b3d

This was referenced Jun 12, 2025

Flink: Backport #13277 to Flink 1.19 / 1.20 #13303

Merged

Flink: Dynamic Iceberg Sink: Add sink / core processing logic / benchmarking #13304

Open

pvary pushed a commit that referenced this pull request Jun 12, 2025

Flink: Backport dynamic Iceberg Sink: Add HashKeyGenerator / RowDataE…

e141ff7

…volver / TableUpdateOperator to Flink 1.19 / 1.20 (#13303) backports #13277


		private DynamicSinkUtil() {}

		static List<Integer> getEqualityFieldIds(List<String> equalityFields, Schema schema) {

Flink: Dynamic Iceberg Sink: Add HashKeyGenerator / RowDataEvolver / TableUpdateOperator #13277

Flink: Dynamic Iceberg Sink: Add HashKeyGenerator / RowDataEvolver / TableUpdateOperator #13277

Conversation

mxm commented Jun 8, 2025

HashKeyGenerator

RowDataEvolver

DynamicUpdateOperator

Uh oh!

Uh oh!

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!

Uh oh!

Uh oh!

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!

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!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

pvary left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

pvary commented Jun 11, 2025

Uh oh!

mxm commented Jun 11, 2025

Uh oh!

mxm commented Jun 11, 2025

Uh oh!

Uh oh!