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 e689699fa6 Flink: Fix dynamic sink HASH distribution check for nested
equality/partition fields (#18296)
e689699fa6 is described below
commit e689699fa6392a2da9aa10870f0d4de575ae23c1
Author: kumarpritam863 <[email protected]>
AuthorDate: Mon Sep 28 17:23:13 2026 +0530
Flink: Fix dynamic sink HASH distribution check for nested
equality/partition fields (#18296)
---
.../flink/sink/dynamic/HashKeyGenerator.java | 15 ++--
.../flink/sink/dynamic/TestHashKeyGenerator.java | 100 +++++++++++++++++++++
.../flink/sink/dynamic/HashKeyGenerator.java | 15 ++--
.../flink/sink/dynamic/TestHashKeyGenerator.java | 100 +++++++++++++++++++++
.../flink/sink/dynamic/HashKeyGenerator.java | 15 ++--
.../flink/sink/dynamic/TestHashKeyGenerator.java | 100 +++++++++++++++++++++
.../flink/sink/dynamic/HashKeyGenerator.java | 15 ++--
.../flink/sink/dynamic/TestHashKeyGenerator.java | 100 +++++++++++++++++++++
8 files changed, 432 insertions(+), 28 deletions(-)
diff --git
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/HashKeyGenerator.java
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/HashKeyGenerator.java
index 0fd6bed9e1..d884696414 100644
---
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/HashKeyGenerator.java
+++
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/HashKeyGenerator.java
@@ -23,7 +23,6 @@ import static
org.apache.iceberg.TableProperties.WRITE_DISTRIBUTION_MODE;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
-import java.util.stream.Collectors;
import javax.annotation.Nullable;
import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.java.functions.KeySelector;
@@ -39,7 +38,6 @@ 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;
@@ -155,17 +153,20 @@ class HashKeyGenerator {
return equalityFieldKeySelector(
tableName, schema, equalityFields, writeParallelism,
maxWriteParallelism);
} else {
+ // Compare by fully-qualified name: a partition source that is a
nested field has a
+ // simple name (e.g. "name") that differs from the dotted
equality-field name (e.g.
+ // "user.name"). Names are also stable across schema evolution,
whereas field IDs are
+ // only meaningful within the schema that assigned them, and the
spec's schema may not
+ // match the schema used here.
for (PartitionField partitionField : spec.fields()) {
- Types.NestedField sourceField =
schema.findField(partitionField.sourceId());
+ String sourceName =
spec.schema().findColumnName(partitionField.sourceId());
Preconditions.checkState(
- sourceField != null &&
equalityFields.contains(sourceField.name()),
+ sourceName != null && equalityFields.contains(sourceName),
"%s: In 'hash' distribution mode with equality fields set,
partition field '%s' "
+ "should be included in equality fields: '%s'",
tableName,
partitionField,
- schema.columns().stream()
- .filter(c -> equalityFields.contains(c.name()))
- .collect(Collectors.toList()));
+ equalityFields);
}
return partitionKeySelector(
tableName, schema, spec, writeParallelism,
maxWriteParallelism);
diff --git
a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestHashKeyGenerator.java
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestHashKeyGenerator.java
index 7a50ca4356..0040b367e9 100644
---
a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestHashKeyGenerator.java
+++
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestHashKeyGenerator.java
@@ -119,6 +119,87 @@ class TestHashKeyGenerator {
assertThat(getSubTaskId(writeKey4, writeParallelism,
maxWriteParallelism)).isEqualTo(1);
}
+ @Test
+ void testHashDistributionModeWithNestedEqualityAndPartitionField() throws
Exception {
+ // A partition sourced from a nested field that is also an equality field
must be accepted.
+ // The check matches by fully-qualified name ("user.name"), not by the
source field's simple
+ // name ("name") which would not match the dotted equality-field name.
+ int writeParallelism = 3;
+ int maxWriteParallelism = 8;
+ HashKeyGenerator generator = new HashKeyGenerator(16, maxWriteParallelism);
+ Schema nestedSchema =
+ new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.required(
+ 2,
+ "user",
+ Types.StructType.of(
+ Types.NestedField.required(3, "name",
Types.StringType.get()))));
+ PartitionSpec spec =
PartitionSpec.builderFor(nestedSchema).identity("user.name").build();
+ Set<String> equalityFields = Collections.singleton("user.name");
+
+ GenericRowData row1 = GenericRowData.of(1,
GenericRowData.of(StringData.fromString("a")));
+ GenericRowData row2 = GenericRowData.of(2,
GenericRowData.of(StringData.fromString("a")));
+ GenericRowData row3 = GenericRowData.of(3,
GenericRowData.of(StringData.fromString("b")));
+
+ int writeKey1 =
+ getWriteKey(
+ generator,
+ nestedSchema,
+ spec,
+ DistributionMode.HASH,
+ writeParallelism,
+ equalityFields,
+ row1);
+ int writeKey2 =
+ getWriteKey(
+ generator,
+ nestedSchema,
+ spec,
+ DistributionMode.HASH,
+ writeParallelism,
+ equalityFields,
+ row2);
+ int writeKey3 =
+ getWriteKey(
+ generator,
+ nestedSchema,
+ spec,
+ DistributionMode.HASH,
+ writeParallelism,
+ equalityFields,
+ row3);
+
+ // Rows sharing the nested partition value land on the same writer; a
different value does not.
+ assertThat(writeKey1).isEqualTo(writeKey2);
+ assertThat(writeKey3).isNotEqualTo(writeKey1);
+ }
+
+ @Test
+ void testHashDistributionModePartitionFieldMustBeEqualityField() throws
Exception {
+ // A partition source that is not an equality field is rejected: equality
deletes are
+ // partition-scoped, so partitioning by a non-key column would corrupt
upsert dedup.
+ int writeParallelism = 3;
+ int maxWriteParallelism = 8;
+ HashKeyGenerator generator = new HashKeyGenerator(16, maxWriteParallelism);
+ PartitionSpec spec =
PartitionSpec.builderFor(SCHEMA).identity("data").build();
+ Set<String> equalityFields = Collections.singleton("id");
+ GenericRowData row = GenericRowData.of(1, StringData.fromString("a"));
+
+ assertThatThrownBy(
+ () ->
+ getWriteKey(
+ generator,
+ SCHEMA,
+ spec,
+ DistributionMode.HASH,
+ writeParallelism,
+ equalityFields,
+ row))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("should be included in equality fields");
+ }
+
@Test
void testEqualityKeys() throws Exception {
int writeParallelism = 2;
@@ -673,6 +754,25 @@ class TestHashKeyGenerator {
assertThat(keySelectorCache).hasSize(2);
}
+ private static int getWriteKey(
+ HashKeyGenerator generator,
+ Schema schema,
+ PartitionSpec spec,
+ DistributionMode mode,
+ int writeParallelism,
+ Set<String> equalityFields,
+ GenericRowData row)
+ throws Exception {
+ DynamicRecord inputRecord =
+ new DynamicRecord(TABLE_IDENTIFIER, BRANCH, schema, row, spec, mode,
writeParallelism);
+ inputRecord.setEqualityFields(equalityFields);
+
+ DynamicRecordWithConfig dynamicRecordWithConfig =
+ new DynamicRecordWithConfig(
+ new FlinkWriteConf(Collections.emptyMap(), new Configuration()));
+ return generator.generateKey(dynamicRecordWithConfig.wrap(inputRecord));
+ }
+
private static int getWriteKey(
HashKeyGenerator generator,
PartitionSpec spec,
diff --git
a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/HashKeyGenerator.java
b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/HashKeyGenerator.java
index 0fd6bed9e1..d884696414 100644
---
a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/HashKeyGenerator.java
+++
b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/HashKeyGenerator.java
@@ -23,7 +23,6 @@ import static
org.apache.iceberg.TableProperties.WRITE_DISTRIBUTION_MODE;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
-import java.util.stream.Collectors;
import javax.annotation.Nullable;
import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.java.functions.KeySelector;
@@ -39,7 +38,6 @@ 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;
@@ -155,17 +153,20 @@ class HashKeyGenerator {
return equalityFieldKeySelector(
tableName, schema, equalityFields, writeParallelism,
maxWriteParallelism);
} else {
+ // Compare by fully-qualified name: a partition source that is a
nested field has a
+ // simple name (e.g. "name") that differs from the dotted
equality-field name (e.g.
+ // "user.name"). Names are also stable across schema evolution,
whereas field IDs are
+ // only meaningful within the schema that assigned them, and the
spec's schema may not
+ // match the schema used here.
for (PartitionField partitionField : spec.fields()) {
- Types.NestedField sourceField =
schema.findField(partitionField.sourceId());
+ String sourceName =
spec.schema().findColumnName(partitionField.sourceId());
Preconditions.checkState(
- sourceField != null &&
equalityFields.contains(sourceField.name()),
+ sourceName != null && equalityFields.contains(sourceName),
"%s: In 'hash' distribution mode with equality fields set,
partition field '%s' "
+ "should be included in equality fields: '%s'",
tableName,
partitionField,
- schema.columns().stream()
- .filter(c -> equalityFields.contains(c.name()))
- .collect(Collectors.toList()));
+ equalityFields);
}
return partitionKeySelector(
tableName, schema, spec, writeParallelism,
maxWriteParallelism);
diff --git
a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestHashKeyGenerator.java
b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestHashKeyGenerator.java
index 7a50ca4356..0040b367e9 100644
---
a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestHashKeyGenerator.java
+++
b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestHashKeyGenerator.java
@@ -119,6 +119,87 @@ class TestHashKeyGenerator {
assertThat(getSubTaskId(writeKey4, writeParallelism,
maxWriteParallelism)).isEqualTo(1);
}
+ @Test
+ void testHashDistributionModeWithNestedEqualityAndPartitionField() throws
Exception {
+ // A partition sourced from a nested field that is also an equality field
must be accepted.
+ // The check matches by fully-qualified name ("user.name"), not by the
source field's simple
+ // name ("name") which would not match the dotted equality-field name.
+ int writeParallelism = 3;
+ int maxWriteParallelism = 8;
+ HashKeyGenerator generator = new HashKeyGenerator(16, maxWriteParallelism);
+ Schema nestedSchema =
+ new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.required(
+ 2,
+ "user",
+ Types.StructType.of(
+ Types.NestedField.required(3, "name",
Types.StringType.get()))));
+ PartitionSpec spec =
PartitionSpec.builderFor(nestedSchema).identity("user.name").build();
+ Set<String> equalityFields = Collections.singleton("user.name");
+
+ GenericRowData row1 = GenericRowData.of(1,
GenericRowData.of(StringData.fromString("a")));
+ GenericRowData row2 = GenericRowData.of(2,
GenericRowData.of(StringData.fromString("a")));
+ GenericRowData row3 = GenericRowData.of(3,
GenericRowData.of(StringData.fromString("b")));
+
+ int writeKey1 =
+ getWriteKey(
+ generator,
+ nestedSchema,
+ spec,
+ DistributionMode.HASH,
+ writeParallelism,
+ equalityFields,
+ row1);
+ int writeKey2 =
+ getWriteKey(
+ generator,
+ nestedSchema,
+ spec,
+ DistributionMode.HASH,
+ writeParallelism,
+ equalityFields,
+ row2);
+ int writeKey3 =
+ getWriteKey(
+ generator,
+ nestedSchema,
+ spec,
+ DistributionMode.HASH,
+ writeParallelism,
+ equalityFields,
+ row3);
+
+ // Rows sharing the nested partition value land on the same writer; a
different value does not.
+ assertThat(writeKey1).isEqualTo(writeKey2);
+ assertThat(writeKey3).isNotEqualTo(writeKey1);
+ }
+
+ @Test
+ void testHashDistributionModePartitionFieldMustBeEqualityField() throws
Exception {
+ // A partition source that is not an equality field is rejected: equality
deletes are
+ // partition-scoped, so partitioning by a non-key column would corrupt
upsert dedup.
+ int writeParallelism = 3;
+ int maxWriteParallelism = 8;
+ HashKeyGenerator generator = new HashKeyGenerator(16, maxWriteParallelism);
+ PartitionSpec spec =
PartitionSpec.builderFor(SCHEMA).identity("data").build();
+ Set<String> equalityFields = Collections.singleton("id");
+ GenericRowData row = GenericRowData.of(1, StringData.fromString("a"));
+
+ assertThatThrownBy(
+ () ->
+ getWriteKey(
+ generator,
+ SCHEMA,
+ spec,
+ DistributionMode.HASH,
+ writeParallelism,
+ equalityFields,
+ row))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("should be included in equality fields");
+ }
+
@Test
void testEqualityKeys() throws Exception {
int writeParallelism = 2;
@@ -673,6 +754,25 @@ class TestHashKeyGenerator {
assertThat(keySelectorCache).hasSize(2);
}
+ private static int getWriteKey(
+ HashKeyGenerator generator,
+ Schema schema,
+ PartitionSpec spec,
+ DistributionMode mode,
+ int writeParallelism,
+ Set<String> equalityFields,
+ GenericRowData row)
+ throws Exception {
+ DynamicRecord inputRecord =
+ new DynamicRecord(TABLE_IDENTIFIER, BRANCH, schema, row, spec, mode,
writeParallelism);
+ inputRecord.setEqualityFields(equalityFields);
+
+ DynamicRecordWithConfig dynamicRecordWithConfig =
+ new DynamicRecordWithConfig(
+ new FlinkWriteConf(Collections.emptyMap(), new Configuration()));
+ return generator.generateKey(dynamicRecordWithConfig.wrap(inputRecord));
+ }
+
private static int getWriteKey(
HashKeyGenerator generator,
PartitionSpec spec,
diff --git
a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/HashKeyGenerator.java
b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/HashKeyGenerator.java
index 0fd6bed9e1..d884696414 100644
---
a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/HashKeyGenerator.java
+++
b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/HashKeyGenerator.java
@@ -23,7 +23,6 @@ import static
org.apache.iceberg.TableProperties.WRITE_DISTRIBUTION_MODE;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
-import java.util.stream.Collectors;
import javax.annotation.Nullable;
import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.java.functions.KeySelector;
@@ -39,7 +38,6 @@ 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;
@@ -155,17 +153,20 @@ class HashKeyGenerator {
return equalityFieldKeySelector(
tableName, schema, equalityFields, writeParallelism,
maxWriteParallelism);
} else {
+ // Compare by fully-qualified name: a partition source that is a
nested field has a
+ // simple name (e.g. "name") that differs from the dotted
equality-field name (e.g.
+ // "user.name"). Names are also stable across schema evolution,
whereas field IDs are
+ // only meaningful within the schema that assigned them, and the
spec's schema may not
+ // match the schema used here.
for (PartitionField partitionField : spec.fields()) {
- Types.NestedField sourceField =
schema.findField(partitionField.sourceId());
+ String sourceName =
spec.schema().findColumnName(partitionField.sourceId());
Preconditions.checkState(
- sourceField != null &&
equalityFields.contains(sourceField.name()),
+ sourceName != null && equalityFields.contains(sourceName),
"%s: In 'hash' distribution mode with equality fields set,
partition field '%s' "
+ "should be included in equality fields: '%s'",
tableName,
partitionField,
- schema.columns().stream()
- .filter(c -> equalityFields.contains(c.name()))
- .collect(Collectors.toList()));
+ equalityFields);
}
return partitionKeySelector(
tableName, schema, spec, writeParallelism,
maxWriteParallelism);
diff --git
a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestHashKeyGenerator.java
b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestHashKeyGenerator.java
index 7a50ca4356..0040b367e9 100644
---
a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestHashKeyGenerator.java
+++
b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestHashKeyGenerator.java
@@ -119,6 +119,87 @@ class TestHashKeyGenerator {
assertThat(getSubTaskId(writeKey4, writeParallelism,
maxWriteParallelism)).isEqualTo(1);
}
+ @Test
+ void testHashDistributionModeWithNestedEqualityAndPartitionField() throws
Exception {
+ // A partition sourced from a nested field that is also an equality field
must be accepted.
+ // The check matches by fully-qualified name ("user.name"), not by the
source field's simple
+ // name ("name") which would not match the dotted equality-field name.
+ int writeParallelism = 3;
+ int maxWriteParallelism = 8;
+ HashKeyGenerator generator = new HashKeyGenerator(16, maxWriteParallelism);
+ Schema nestedSchema =
+ new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.required(
+ 2,
+ "user",
+ Types.StructType.of(
+ Types.NestedField.required(3, "name",
Types.StringType.get()))));
+ PartitionSpec spec =
PartitionSpec.builderFor(nestedSchema).identity("user.name").build();
+ Set<String> equalityFields = Collections.singleton("user.name");
+
+ GenericRowData row1 = GenericRowData.of(1,
GenericRowData.of(StringData.fromString("a")));
+ GenericRowData row2 = GenericRowData.of(2,
GenericRowData.of(StringData.fromString("a")));
+ GenericRowData row3 = GenericRowData.of(3,
GenericRowData.of(StringData.fromString("b")));
+
+ int writeKey1 =
+ getWriteKey(
+ generator,
+ nestedSchema,
+ spec,
+ DistributionMode.HASH,
+ writeParallelism,
+ equalityFields,
+ row1);
+ int writeKey2 =
+ getWriteKey(
+ generator,
+ nestedSchema,
+ spec,
+ DistributionMode.HASH,
+ writeParallelism,
+ equalityFields,
+ row2);
+ int writeKey3 =
+ getWriteKey(
+ generator,
+ nestedSchema,
+ spec,
+ DistributionMode.HASH,
+ writeParallelism,
+ equalityFields,
+ row3);
+
+ // Rows sharing the nested partition value land on the same writer; a
different value does not.
+ assertThat(writeKey1).isEqualTo(writeKey2);
+ assertThat(writeKey3).isNotEqualTo(writeKey1);
+ }
+
+ @Test
+ void testHashDistributionModePartitionFieldMustBeEqualityField() throws
Exception {
+ // A partition source that is not an equality field is rejected: equality
deletes are
+ // partition-scoped, so partitioning by a non-key column would corrupt
upsert dedup.
+ int writeParallelism = 3;
+ int maxWriteParallelism = 8;
+ HashKeyGenerator generator = new HashKeyGenerator(16, maxWriteParallelism);
+ PartitionSpec spec =
PartitionSpec.builderFor(SCHEMA).identity("data").build();
+ Set<String> equalityFields = Collections.singleton("id");
+ GenericRowData row = GenericRowData.of(1, StringData.fromString("a"));
+
+ assertThatThrownBy(
+ () ->
+ getWriteKey(
+ generator,
+ SCHEMA,
+ spec,
+ DistributionMode.HASH,
+ writeParallelism,
+ equalityFields,
+ row))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("should be included in equality fields");
+ }
+
@Test
void testEqualityKeys() throws Exception {
int writeParallelism = 2;
@@ -673,6 +754,25 @@ class TestHashKeyGenerator {
assertThat(keySelectorCache).hasSize(2);
}
+ private static int getWriteKey(
+ HashKeyGenerator generator,
+ Schema schema,
+ PartitionSpec spec,
+ DistributionMode mode,
+ int writeParallelism,
+ Set<String> equalityFields,
+ GenericRowData row)
+ throws Exception {
+ DynamicRecord inputRecord =
+ new DynamicRecord(TABLE_IDENTIFIER, BRANCH, schema, row, spec, mode,
writeParallelism);
+ inputRecord.setEqualityFields(equalityFields);
+
+ DynamicRecordWithConfig dynamicRecordWithConfig =
+ new DynamicRecordWithConfig(
+ new FlinkWriteConf(Collections.emptyMap(), new Configuration()));
+ return generator.generateKey(dynamicRecordWithConfig.wrap(inputRecord));
+ }
+
private static int getWriteKey(
HashKeyGenerator generator,
PartitionSpec spec,
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 0fd6bed9e1..d884696414 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
@@ -23,7 +23,6 @@ import static
org.apache.iceberg.TableProperties.WRITE_DISTRIBUTION_MODE;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
-import java.util.stream.Collectors;
import javax.annotation.Nullable;
import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.java.functions.KeySelector;
@@ -39,7 +38,6 @@ 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;
@@ -155,17 +153,20 @@ class HashKeyGenerator {
return equalityFieldKeySelector(
tableName, schema, equalityFields, writeParallelism,
maxWriteParallelism);
} else {
+ // Compare by fully-qualified name: a partition source that is a
nested field has a
+ // simple name (e.g. "name") that differs from the dotted
equality-field name (e.g.
+ // "user.name"). Names are also stable across schema evolution,
whereas field IDs are
+ // only meaningful within the schema that assigned them, and the
spec's schema may not
+ // match the schema used here.
for (PartitionField partitionField : spec.fields()) {
- Types.NestedField sourceField =
schema.findField(partitionField.sourceId());
+ String sourceName =
spec.schema().findColumnName(partitionField.sourceId());
Preconditions.checkState(
- sourceField != null &&
equalityFields.contains(sourceField.name()),
+ sourceName != null && equalityFields.contains(sourceName),
"%s: In 'hash' distribution mode with equality fields set,
partition field '%s' "
+ "should be included in equality fields: '%s'",
tableName,
partitionField,
- schema.columns().stream()
- .filter(c -> equalityFields.contains(c.name()))
- .collect(Collectors.toList()));
+ equalityFields);
}
return partitionKeySelector(
tableName, schema, spec, writeParallelism,
maxWriteParallelism);
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 7a50ca4356..0040b367e9 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
@@ -119,6 +119,87 @@ class TestHashKeyGenerator {
assertThat(getSubTaskId(writeKey4, writeParallelism,
maxWriteParallelism)).isEqualTo(1);
}
+ @Test
+ void testHashDistributionModeWithNestedEqualityAndPartitionField() throws
Exception {
+ // A partition sourced from a nested field that is also an equality field
must be accepted.
+ // The check matches by fully-qualified name ("user.name"), not by the
source field's simple
+ // name ("name") which would not match the dotted equality-field name.
+ int writeParallelism = 3;
+ int maxWriteParallelism = 8;
+ HashKeyGenerator generator = new HashKeyGenerator(16, maxWriteParallelism);
+ Schema nestedSchema =
+ new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.required(
+ 2,
+ "user",
+ Types.StructType.of(
+ Types.NestedField.required(3, "name",
Types.StringType.get()))));
+ PartitionSpec spec =
PartitionSpec.builderFor(nestedSchema).identity("user.name").build();
+ Set<String> equalityFields = Collections.singleton("user.name");
+
+ GenericRowData row1 = GenericRowData.of(1,
GenericRowData.of(StringData.fromString("a")));
+ GenericRowData row2 = GenericRowData.of(2,
GenericRowData.of(StringData.fromString("a")));
+ GenericRowData row3 = GenericRowData.of(3,
GenericRowData.of(StringData.fromString("b")));
+
+ int writeKey1 =
+ getWriteKey(
+ generator,
+ nestedSchema,
+ spec,
+ DistributionMode.HASH,
+ writeParallelism,
+ equalityFields,
+ row1);
+ int writeKey2 =
+ getWriteKey(
+ generator,
+ nestedSchema,
+ spec,
+ DistributionMode.HASH,
+ writeParallelism,
+ equalityFields,
+ row2);
+ int writeKey3 =
+ getWriteKey(
+ generator,
+ nestedSchema,
+ spec,
+ DistributionMode.HASH,
+ writeParallelism,
+ equalityFields,
+ row3);
+
+ // Rows sharing the nested partition value land on the same writer; a
different value does not.
+ assertThat(writeKey1).isEqualTo(writeKey2);
+ assertThat(writeKey3).isNotEqualTo(writeKey1);
+ }
+
+ @Test
+ void testHashDistributionModePartitionFieldMustBeEqualityField() throws
Exception {
+ // A partition source that is not an equality field is rejected: equality
deletes are
+ // partition-scoped, so partitioning by a non-key column would corrupt
upsert dedup.
+ int writeParallelism = 3;
+ int maxWriteParallelism = 8;
+ HashKeyGenerator generator = new HashKeyGenerator(16, maxWriteParallelism);
+ PartitionSpec spec =
PartitionSpec.builderFor(SCHEMA).identity("data").build();
+ Set<String> equalityFields = Collections.singleton("id");
+ GenericRowData row = GenericRowData.of(1, StringData.fromString("a"));
+
+ assertThatThrownBy(
+ () ->
+ getWriteKey(
+ generator,
+ SCHEMA,
+ spec,
+ DistributionMode.HASH,
+ writeParallelism,
+ equalityFields,
+ row))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("should be included in equality fields");
+ }
+
@Test
void testEqualityKeys() throws Exception {
int writeParallelism = 2;
@@ -673,6 +754,25 @@ class TestHashKeyGenerator {
assertThat(keySelectorCache).hasSize(2);
}
+ private static int getWriteKey(
+ HashKeyGenerator generator,
+ Schema schema,
+ PartitionSpec spec,
+ DistributionMode mode,
+ int writeParallelism,
+ Set<String> equalityFields,
+ GenericRowData row)
+ throws Exception {
+ DynamicRecord inputRecord =
+ new DynamicRecord(TABLE_IDENTIFIER, BRANCH, schema, row, spec, mode,
writeParallelism);
+ inputRecord.setEqualityFields(equalityFields);
+
+ DynamicRecordWithConfig dynamicRecordWithConfig =
+ new DynamicRecordWithConfig(
+ new FlinkWriteConf(Collections.emptyMap(), new Configuration()));
+ return generator.generateKey(dynamicRecordWithConfig.wrap(inputRecord));
+ }
+
private static int getWriteKey(
HashKeyGenerator generator,
PartitionSpec spec,
