mxm commented on code in PR #18602:
URL: https://github.com/apache/iceberg/pull/18602#discussion_r4390716543
##########
flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java:
##########
@@ -291,6 +293,155 @@ private static DataFormatConverters.RowConverter
converter(Schema schema) {
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 {
Review Comment:
Not sure we need this test because it doesn't test the update path, but it
doesn't hurt.
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordProcessor.java:
##########
@@ -159,15 +159,10 @@ public void collect(DynamicRecord inputData) {
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, null, null, null);
Review Comment:
This could be replaced with
```suggestion
int writerKey = hashKeyGenerator.generateKey(data);
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
