zhengxiang378928908-code commented on code in PR #18602:
URL: https://github.com/apache/iceberg/pull/18602#discussion_r4395909492
##########
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:
Kept it as a regression test for the main path with reordered columns; happy
to drop it if you prefer.
--
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]
