This is an automated email from the ASF dual-hosted git repository.
mxm pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new b61ef96a34 Flink: Fix dynamic sink HASH keying of records sent to the
table updater (#18602)
b61ef96a34 is described below
commit b61ef96a34bd1054fca4a705ee73c7567f84781b
Author: zhengxiang378928908-code <[email protected]>
AuthorDate: Wed Oct 28 18:34:53 2026 +0800
Flink: Fix dynamic sink HASH keying of records sent to the table updater
(#18602)
---
.../flink/sink/dynamic/DynamicRecordProcessor.java | 13 +-
.../flink/sink/dynamic/HashKeyGenerator.java | 10 +-
.../flink/sink/dynamic/TestDynamicIcebergSink.java | 151 +++++++++++++++++++++
.../flink/sink/dynamic/TestHashKeyGenerator.java | 46 +++++++
4 files changed, 207 insertions(+), 13 deletions(-)
diff --git
a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordProcessor.java
b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordProcessor.java
index ba2d816ae9..e0ed8b8a26 100644
---
a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordProcessor.java
+++
b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordProcessor.java
@@ -159,15 +159,10 @@ class DynamicRecordProcessor<T> extends
ProcessFunction<T, DynamicRecordInternal
newData.f1,
isForward);
} else {
- // Shuffled records with immediateUpdate=false go to the update side
output
- int writerKey =
- hashKeyGenerator.generateKey(
- data,
- foundSchema.resolvedTableSchema() != null
- ? foundSchema.resolvedTableSchema()
- : data.schema(),
- foundSpec != null ? foundSpec : data.spec(),
- data.rowData());
+ // 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);
context.output(
updateStream,
new DynamicRecordInternal(
diff --git
a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/HashKeyGenerator.java
b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/HashKeyGenerator.java
index d884696414..c60dd9b0ce 100644
---
a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/HashKeyGenerator.java
+++
b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/HashKeyGenerator.java
@@ -38,6 +38,7 @@ import org.apache.iceberg.flink.sink.PartitionKeySelector;
import org.apache.iceberg.relocated.com.google.common.base.MoreObjects;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.types.Types;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -66,7 +67,7 @@ class HashKeyGenerator {
this.keySelectorCache = new LRUCache<>(maxCacheSize);
}
- int generateKey(DynamicRecord dynamicRecord) throws Exception {
+ int generateKey(DynamicRecord dynamicRecord) {
return generateKey(dynamicRecord, null, null, null);
}
@@ -317,14 +318,15 @@ class HashKeyGenerator {
/**
* Cache key for the {@link KeySelector}. Only contains the {@link Schema}
and the {@link
- * PartitionSpec} if their ids are not provided.
+ * PartitionSpec} if their ids are not provided. The schema is compared by
its struct, because
+ * {@link Schema} has no value equality and callers may pass a new instance
for every record.
*/
static class SelectorKey {
private final String tableName;
private final String branch;
private final Integer schemaId;
private final Integer specId;
- private final Schema schema;
+ private final Types.StructType schema;
private final PartitionSpec spec;
private final Set<String> equalityFields;
private final DistributionMode distributionMode;
@@ -344,7 +346,7 @@ class HashKeyGenerator {
this.branch = branch;
this.schemaId = tableSchemaId;
this.specId = tableSpecId;
- this.schema = tableSchemaId == null ? schema : null;
+ this.schema = tableSchemaId == null ? schema.asStruct() : null;
this.spec = tableSpecId == null ? spec : null;
this.equalityFields = equalityFields;
this.distributionMode = distributionMode;
diff --git
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java
index 9e2f00bca0..26565089bd 100644
---
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java
+++
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java
@@ -21,11 +21,13 @@ package org.apache.iceberg.flink.sink.dynamic;
import static org.apache.iceberg.flink.TestFixtures.DATABASE;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.fail;
+import static org.assertj.core.api.Assertions.tuple;
import java.io.IOException;
import java.io.Serializable;
import java.nio.ByteBuffer;
import java.time.Duration;
+import java.time.LocalDate;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
@@ -291,6 +293,155 @@ class TestDynamicIcebergSink extends
TestFlinkIcebergSinkBase {
resolvedSchema.getColumnDataTypes().toArray(DataType[]::new));
}
+ /**
+ * Two writer-schema variants of one table with independently assigned field
IDs: event_date has
+ * id 2 / position 1 in variant A and id 3 / position 2 in variant B. Each
variant brings its own
+ * identity(event_date) spec. Records alternate between the variants with
HASH distribution.
+ */
+ @Test
+ void testHashDistributionWithAlternatingSchemaVariants() throws Exception {
+ String tableName = "alternating_variants";
+ Schema variantA =
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.StringType.get()),
+ Types.NestedField.optional(2, "event_date", Types.DateType.get()));
+ Schema variantB =
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.StringType.get()),
+ Types.NestedField.optional(2, "extra", Types.StringType.get()),
+ Types.NestedField.optional(3, "event_date", Types.DateType.get()));
+ PartitionSpec specA =
PartitionSpec.builderFor(variantA).identity("event_date").build();
+ PartitionSpec specB =
PartitionSpec.builderFor(variantB).identity("event_date").build();
+
+ DynamicIcebergDataImpl a1 =
+ new DynamicIcebergDataImpl(variantA, tableName,
SnapshotRef.MAIN_BRANCH, specA);
+ a1.rowProvided = Row.of("a1", LocalDate.of(2026, 10, 1));
+ DynamicIcebergDataImpl b1 =
+ new DynamicIcebergDataImpl(variantB, tableName,
SnapshotRef.MAIN_BRANCH, specB);
+ b1.rowProvided = Row.of("b1", "x1", LocalDate.of(2026, 10, 2));
+ DynamicIcebergDataImpl a2 =
+ new DynamicIcebergDataImpl(variantA, tableName,
SnapshotRef.MAIN_BRANCH, specA);
+ a2.rowProvided = Row.of("a2", LocalDate.of(2026, 10, 3));
+ DynamicIcebergDataImpl b2 =
+ new DynamicIcebergDataImpl(variantB, tableName,
SnapshotRef.MAIN_BRANCH, specB);
+ b2.rowProvided = Row.of("b2", "x2", LocalDate.of(2026, 10, 4));
+
+ env =
+ StreamExecutionEnvironment.getExecutionEnvironment(
+ MiniFlinkClusterExtension.DISABLE_CLASSLOADER_CHECK_CONFIG);
+ executeDynamicSink(Lists.newArrayList(a1, b1, a2, b2), env, false, 1,
null);
+
+ Table table =
CATALOG_EXTENSION.catalog().loadTable(TableIdentifier.of(DATABASE, tableName));
+ List<Record> actual;
+ try (CloseableIterable<Record> records =
IcebergGenerics.read(table).build()) {
+ actual = Lists.newArrayList(records);
+ }
+ assertThat(actual)
+ .extracting(r -> r.getField("id"), r -> r.getField("extra"), r ->
r.getField("event_date"))
+ .containsExactlyInAnyOrder(
+ tuple("a1", null, LocalDate.of(2026, 10, 1)),
+ tuple("b1", "x1", LocalDate.of(2026, 10, 2)),
+ tuple("a2", null, LocalDate.of(2026, 10, 3)),
+ tuple("b2", "x2", LocalDate.of(2026, 10, 4)));
+ }
+
+ /**
+ * The table exists with (id, event_date); records list the columns in the
other order, so they
+ * need a data conversion, and go to a branch that does not exist yet, so
they take the update
+ * path although the table schema and spec are known.
+ */
+ @Test
+ void testHashDistributionToNewBranchWithDataConversion() throws Exception {
+ String tableName = "hash_new_branch_conversion";
+ Schema tableSchema =
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.StringType.get()),
+ Types.NestedField.optional(2, "event_date", Types.DateType.get()));
+ PartitionSpec tableSpec =
PartitionSpec.builderFor(tableSchema).identity("event_date").build();
+ CATALOG_EXTENSION
+ .catalog()
+ .createTable(TableIdentifier.of(DATABASE, tableName), tableSchema,
tableSpec);
+
+ Schema reordered =
+ new Schema(
+ Types.NestedField.optional(1, "event_date", Types.DateType.get()),
+ Types.NestedField.optional(2, "id", Types.StringType.get()));
+ PartitionSpec reorderedSpec =
+ PartitionSpec.builderFor(reordered).identity("event_date").build();
+
+ DynamicIcebergDataImpl r1 =
+ new DynamicIcebergDataImpl(reordered, tableName, "b1", reorderedSpec);
+ r1.rowProvided = Row.of(LocalDate.of(2026, 10, 1), "r1");
+ DynamicIcebergDataImpl r2 =
+ new DynamicIcebergDataImpl(reordered, tableName, "b1", reorderedSpec);
+ r2.rowProvided = Row.of(LocalDate.of(2026, 10, 2), "r2");
+
+ env =
+ StreamExecutionEnvironment.getExecutionEnvironment(
+ MiniFlinkClusterExtension.DISABLE_CLASSLOADER_CHECK_CONFIG);
+ executeDynamicSink(Lists.newArrayList(r1, r2), env, false, 1, null);
+
+ Table table =
CATALOG_EXTENSION.catalog().loadTable(TableIdentifier.of(DATABASE, tableName));
+ List<Record> actual;
+ try (CloseableIterable<Record> records =
+
IcebergGenerics.read(table).useSnapshot(table.refs().get("b1").snapshotId()).build())
{
+ actual = Lists.newArrayList(records);
+ }
+ assertThat(actual)
+ .extracting(r -> r.getField("id"), r -> r.getField("event_date"))
+ .containsExactlyInAnyOrder(
+ tuple("r1", LocalDate.of(2026, 10, 1)), tuple("r2",
LocalDate.of(2026, 10, 2)));
+ }
+
+ /**
+ * The table exists with (id, event_date) and a main branch, and records
list the columns in the
+ * other order. No table update is needed, so the row is converted to the
table layout before it
+ * is keyed.
+ */
+ @Test
+ void testHashDistributionWithReorderedColumns() throws Exception {
+ String tableName = "hash_reordered_columns";
+ Schema tableSchema =
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.StringType.get()),
+ Types.NestedField.optional(2, "event_date", Types.DateType.get()));
+ PartitionSpec tableSpec =
PartitionSpec.builderFor(tableSchema).identity("event_date").build();
+ CATALOG_EXTENSION
+ .catalog()
+ .createTable(TableIdentifier.of(DATABASE, tableName), tableSchema,
tableSpec)
+ .newFastAppend()
+ .commit();
+
+ Schema reordered =
+ new Schema(
+ Types.NestedField.optional(1, "event_date", Types.DateType.get()),
+ Types.NestedField.optional(2, "id", Types.StringType.get()));
+ PartitionSpec reorderedSpec =
+ PartitionSpec.builderFor(reordered).identity("event_date").build();
+
+ DynamicIcebergDataImpl r1 =
+ new DynamicIcebergDataImpl(reordered, tableName,
SnapshotRef.MAIN_BRANCH, reorderedSpec);
+ r1.rowProvided = Row.of(LocalDate.of(2026, 10, 1), "r1");
+ DynamicIcebergDataImpl r2 =
+ new DynamicIcebergDataImpl(reordered, tableName,
SnapshotRef.MAIN_BRANCH, reorderedSpec);
+ r2.rowProvided = Row.of(LocalDate.of(2026, 10, 2), "r2");
+
+ env =
+ StreamExecutionEnvironment.getExecutionEnvironment(
+ MiniFlinkClusterExtension.DISABLE_CLASSLOADER_CHECK_CONFIG);
+ executeDynamicSink(Lists.newArrayList(r1, r2), env, false, 1, null);
+
+ Table table =
CATALOG_EXTENSION.catalog().loadTable(TableIdentifier.of(DATABASE, tableName));
+ List<Record> actual;
+ try (CloseableIterable<Record> records =
IcebergGenerics.read(table).build()) {
+ actual = Lists.newArrayList(records);
+ }
+ assertThat(actual)
+ .extracting(r -> r.getField("id"), r -> r.getField("event_date"))
+ .containsExactlyInAnyOrder(
+ tuple("r1", LocalDate.of(2026, 10, 1)), tuple("r2",
LocalDate.of(2026, 10, 2)));
+ }
+
@Test
void testWrite() throws Exception {
List<DynamicIcebergDataImpl> rows =
diff --git
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestHashKeyGenerator.java
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestHashKeyGenerator.java
index 0040b367e9..837de70d3d 100644
---
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestHashKeyGenerator.java
+++
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestHashKeyGenerator.java
@@ -686,6 +686,52 @@ class TestHashKeyGenerator {
assertThat(writeKey1).isEqualTo(writeKey3);
}
+ @Test
+ void testCacheHitForEqualSchemaInstances() throws Exception {
+ int maxWriteParallelism = 8;
+ HashKeyGenerator generator = new HashKeyGenerator(10, maxWriteParallelism);
+ Map<HashKeyGenerator.SelectorKey, KeySelector<RowData, Integer>>
keySelectorCache =
+ generator.getKeySelectorCache();
+
+ PartitionSpec unpartitioned = PartitionSpec.unpartitioned();
+ DynamicRecord record1 =
+ new DynamicRecord(
+ TABLE_IDENTIFIER,
+ BRANCH,
+ new Schema(SCHEMA.columns()),
+ GenericRowData.of(1, StringData.fromString("foo")),
+ unpartitioned,
+ DistributionMode.NONE,
+ 2);
+ DynamicRecord record2 =
+ new DynamicRecord(
+ TABLE_IDENTIFIER,
+ BRANCH,
+ new Schema(SCHEMA.columns()),
+ GenericRowData.of(1, StringData.fromString("foo")),
+ unpartitioned,
+ DistributionMode.NONE,
+ 2);
+
+ int writeKey1 = generator.generateKey(record1);
+ int writeKey2 = generator.generateKey(record2);
+ // The second record reuses the key selector, so the round robin moves on
+ assertThat(keySelectorCache).hasSize(1);
+ assertThat(writeKey2).isNotEqualTo(writeKey1);
+
+ DynamicRecord record3 =
+ new DynamicRecord(
+ TABLE_IDENTIFIER,
+ BRANCH,
+ new Schema(Types.NestedField.required(1, "id",
Types.IntegerType.get())),
+ GenericRowData.of(1),
+ unpartitioned,
+ DistributionMode.NONE,
+ 2);
+ generator.generateKey(record3);
+ assertThat(keySelectorCache).hasSize(2);
+ }
+
@Test
void testCacheMissOnWriteParallelismChange() throws Exception {
int maxWriteParallelism = 8;
