zhengxiang378928908-code commented on code in PR #18602:
URL: https://github.com/apache/iceberg/pull/18602#discussion_r4380953139
##########
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);
+ }
Review Comment:
Removed. The remaining test runs with `immediateUpdate = false`, which is
the path this PR changes.
##########
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);
+ }
Review Comment:
Done in 198bab601: the test now writes four explicit records, alternating A,
B, A, B. Without the fix it still fails with the `ClassCastException` from
#17651.
##########
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);
+ }
Review Comment:
Done, two explicit records. It still fails without the fix.
##########
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);
Review Comment:
Done. All three tests now check every written row by `id` and `event_date`
(and `extra` in the alternating test).
--
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]
