zhengxiang378928908-code commented on code in PR #18602:
URL: https://github.com/apache/iceberg/pull/18602#discussion_r4380955231
##########
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:
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.
198bab601 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.
##########
flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java:
##########
@@ -291,6 +292,157 @@ private static DataFormatConverters.RowConverter
converter(Schema schema) {
resolvedSchema.getColumnDataTypes().toArray(DataType[]::new));
}
+ @Test
+ void testHashDistributionWithAlternatingSchemaVariantsImmediateUpdate()
throws Exception {
+ runAlternatingSchemaVariants("alternating_variants_immediate", true);
+ }
+
+ @Test
+ void testHashDistributionWithAlternatingSchemaVariants() throws Exception {
+ runAlternatingSchemaVariants("alternating_variants", false);
+ }
+
+ /**
+ * 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.
+ */
+ private void runAlternatingSchemaVariants(String tableName, boolean
immediateUpdate)
+ throws Exception {
+ 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();
+
+ List<DynamicIcebergDataImpl> rows = Lists.newArrayList();
+ 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);
+ }
+
+ env =
+ StreamExecutionEnvironment.getExecutionEnvironment(
+ MiniFlinkClusterExtension.DISABLE_CLASSLOADER_CHECK_CONFIG);
+ executeDynamicSink(rows, env, immediateUpdate, 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).hasSize(6);
+ }
+
+ /**
+ * 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();
+
+ List<DynamicIcebergDataImpl> rows = Lists.newArrayList();
+ 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);
+ }
+
+ env =
+ StreamExecutionEnvironment.getExecutionEnvironment(
+ MiniFlinkClusterExtension.DISABLE_CLASSLOADER_CHECK_CONFIG);
+ executeDynamicSink(rows, 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).hasSize(4);
+ }
+
+ /**
+ * 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();
+
+ List<DynamicIcebergDataImpl> rows = Lists.newArrayList();
+ 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);
+ }
Review Comment:
Done, two explicit records.
--
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]
