You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "sqd (via GitHub)" <gi...@apache.org> on 2026/06/30 21:44:50 UTC
[PR] Flink: handle simultaneous schema evolution and data conversion [iceberg]
sqd opened a new pull request, #17024:
URL: https://github.com/apache/iceberg/pull/17024
Currently, the dynamic sink has a bug in that if an incoming record requires both schema evolution (because of additional column) and data conversion (because of type widening), the logics will try to apply the schema evolution first (correct behavior), but when it's doing so, it is not aware of the type widening data conversion that it'll do in the next step. As a result, it tries to evolve the table column to the narrow type and crashes. Concrete example: table schema [id: long], incoming record [id: INT, value: string].
This change fixes the bug, by checking whether a data conversion can be performed in the next step, and skip schema evolution on that column.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: handle simultaneous schema evolution and data conversion [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on PR #17024:
URL: https://github.com/apache/iceberg/pull/17024#issuecomment-4957592667
Merged to main.
Thanks @sqd for the PR and @mxm for the review!
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: handle simultaneous schema evolution and data conversion [iceberg]
Posted by "sqd (via GitHub)" <gi...@apache.org>.
sqd commented on code in PR #17024:
URL: https://github.com/apache/iceberg/pull/17024#discussion_r3520938863
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/EvolveSchemaVisitor.java:
##########
@@ -211,8 +212,15 @@ private void updateColumn(Types.NestedField existingField, Types.NestedField tar
String existingColumnName = this.existingSchema.findColumnName(existingField.fieldId());
boolean needsOptionalUpdate = targetField.isOptional() && existingField.isRequired();
+ boolean narrowingHandledByConversion =
+ existingField.type().isPrimitiveType()
+ && targetField.type().isPrimitiveType()
+ && TypeUtil.isPromotionAllowed(
Review Comment:
Any insight on why we have the mismatch between spec and CompareSchemasVisitor in the first place?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: handle simultaneous schema evolution and data conversion [iceberg]
Posted by "sqd (via GitHub)" <gi...@apache.org>.
sqd commented on code in PR #17024:
URL: https://github.com/apache/iceberg/pull/17024#discussion_r3545331317
##########
flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java:
##########
@@ -174,6 +178,102 @@ void testInvalidateOldCacheEntryOnUpdate() {
.isTrue();
}
+ @Test
+ void testAddColumnWithResidualNarrowingEvolvesThenConverts() {
Review Comment:
done
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/EvolveSchemaVisitor.java:
##########
@@ -211,8 +211,15 @@ private void updateColumn(Types.NestedField existingField, Types.NestedField tar
String existingColumnName = this.existingSchema.findColumnName(existingField.fieldId());
boolean needsOptionalUpdate = targetField.isOptional() && existingField.isRequired();
+ boolean handledByDataConversion =
+ existingField.type().isPrimitiveType()
+ && targetField.type().isPrimitiveType()
+ && CompareSchemasVisitor.isDataConversionPossible(
+ targetField.type().asPrimitiveType(), existingField.type().asPrimitiveType());
Review Comment:
done
##########
flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java:
##########
@@ -174,6 +178,102 @@ void testInvalidateOldCacheEntryOnUpdate() {
.isTrue();
}
+ @Test
+ void testAddColumnWithResidualNarrowingEvolvesThenConverts() {
+ Catalog catalog = CATALOG_EXTENSION.catalog();
+ TableIdentifier tableIdentifier = TableIdentifier.parse("default.myTable");
+ Schema tableSchema =
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "data", Types.StringType.get()));
+ catalog.createTable(tableIdentifier, tableSchema);
+
+ TableMetadataCache cache =
+ new TableMetadataCache(catalog, 10, Long.MAX_VALUE, 10, CASE_SENSITIVE, PRESERVE_COLUMNS);
+ TableUpdater tableUpdater = new TableUpdater(cache, catalog, CASE_SENSITIVE, PRESERVE_COLUMNS);
+
+ Schema rowSchema =
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "data", Types.StringType.get()),
+ Types.NestedField.optional(3, "extra", Types.StringType.get()));
+
+ Tuple2<TableMetadataCache.ResolvedSchemaInfo, PartitionSpec> result =
+ tableUpdater.update(
+ tableIdentifier,
+ SnapshotRef.MAIN_BRANCH,
+ rowSchema,
+ PartitionSpec.unpartitioned(),
+ TableCreator.DEFAULT);
+
+ Schema evolved = catalog.loadTable(tableIdentifier).schema();
+ assertThat(evolved.findField("id").type()).isEqualTo(Types.LongType.get());
+ assertThat(evolved.findField("extra")).isNotNull();
+ assertThat(result.f0.compareResult())
+ .isEqualTo(CompareSchemasVisitor.Result.DATA_CONVERSION_NEEDED);
+
+ RowData converted =
+ (RowData)
+ result
+ .f0
+ .recordConverter()
+ .convert(
+ GenericRowData.of(5, StringData.fromString("a"), StringData.fromString("x")));
+ assertThat(converted.getLong(0)).isEqualTo(5L);
+ assertThat(converted.getString(1)).hasToString("a");
+ assertThat(converted.getString(2)).hasToString("x");
+ }
+
+ @Test
+ void testAddColumnWithResidualDateToTimestampEvolvesThenConverts() {
+ Catalog catalog = CATALOG_EXTENSION.catalog();
+ TableIdentifier tableIdentifier = TableIdentifier.parse("default.myTable");
+ Schema tableSchema =
+ new Schema(
+ Types.NestedField.optional(1, "ts", Types.TimestampType.withoutZone()),
+ Types.NestedField.optional(2, "data", Types.StringType.get()));
+ catalog.createTable(tableIdentifier, tableSchema);
+
+ TableMetadataCache cache =
+ new TableMetadataCache(catalog, 10, Long.MAX_VALUE, 10, CASE_SENSITIVE, PRESERVE_COLUMNS);
+ TableUpdater tableUpdater = new TableUpdater(cache, catalog, CASE_SENSITIVE, PRESERVE_COLUMNS);
+
+ Schema rowSchema =
+ new Schema(
+ Types.NestedField.optional(1, "ts", Types.DateType.get()),
+ Types.NestedField.optional(2, "data", Types.StringType.get()),
+ Types.NestedField.optional(3, "extra", Types.StringType.get()));
+
+ Tuple2<TableMetadataCache.ResolvedSchemaInfo, PartitionSpec> result =
+ tableUpdater.update(
+ tableIdentifier,
+ SnapshotRef.MAIN_BRANCH,
+ rowSchema,
+ PartitionSpec.unpartitioned(),
+ TableCreator.DEFAULT);
Review Comment:
done
##########
flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java:
##########
@@ -174,6 +178,102 @@ void testInvalidateOldCacheEntryOnUpdate() {
.isTrue();
}
+ @Test
+ void testAddColumnWithResidualNarrowingEvolvesThenConverts() {
+ Catalog catalog = CATALOG_EXTENSION.catalog();
+ TableIdentifier tableIdentifier = TableIdentifier.parse("default.myTable");
+ Schema tableSchema =
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "data", Types.StringType.get()));
+ catalog.createTable(tableIdentifier, tableSchema);
+
+ TableMetadataCache cache =
+ new TableMetadataCache(catalog, 10, Long.MAX_VALUE, 10, CASE_SENSITIVE, PRESERVE_COLUMNS);
+ TableUpdater tableUpdater = new TableUpdater(cache, catalog, CASE_SENSITIVE, PRESERVE_COLUMNS);
+
+ Schema rowSchema =
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "data", Types.StringType.get()),
+ Types.NestedField.optional(3, "extra", Types.StringType.get()));
+
+ Tuple2<TableMetadataCache.ResolvedSchemaInfo, PartitionSpec> result =
+ tableUpdater.update(
+ tableIdentifier,
+ SnapshotRef.MAIN_BRANCH,
+ rowSchema,
+ PartitionSpec.unpartitioned(),
+ TableCreator.DEFAULT);
+
+ Schema evolved = catalog.loadTable(tableIdentifier).schema();
+ assertThat(evolved.findField("id").type()).isEqualTo(Types.LongType.get());
+ assertThat(evolved.findField("extra")).isNotNull();
+ assertThat(result.f0.compareResult())
+ .isEqualTo(CompareSchemasVisitor.Result.DATA_CONVERSION_NEEDED);
+
+ RowData converted =
+ (RowData)
+ result
+ .f0
+ .recordConverter()
+ .convert(
+ GenericRowData.of(5, StringData.fromString("a"), StringData.fromString("x")));
+ assertThat(converted.getLong(0)).isEqualTo(5L);
+ assertThat(converted.getString(1)).hasToString("a");
+ assertThat(converted.getString(2)).hasToString("x");
+ }
+
+ @Test
+ void testAddColumnWithResidualDateToTimestampEvolvesThenConverts() {
Review Comment:
done
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/CompareSchemasVisitor.java:
##########
@@ -180,27 +179,43 @@ public Result primitive(Type.PrimitiveType primitive, Integer tableSchemaId) {
Type.PrimitiveType tableSchemaPrimitiveType = tableSchemaType.asPrimitiveType();
if (primitive.equals(tableSchemaPrimitiveType)) {
return Result.SAME;
- } else if (primitive.equals(Types.IntegerType.get())
- && tableSchemaPrimitiveType.equals(Types.LongType.get())) {
- return Result.DATA_CONVERSION_NEEDED;
- } else if (primitive.equals(Types.FloatType.get())
- && tableSchemaPrimitiveType.equals(Types.DoubleType.get())) {
+ } else if (isDataConversionPossible(primitive, tableSchemaPrimitiveType)) {
return Result.DATA_CONVERSION_NEEDED;
- } else if (primitive.equals(Types.DateType.get())
- && tableSchemaPrimitiveType.equals(Types.TimestampType.withoutZone())) {
- return Result.DATA_CONVERSION_NEEDED;
- } else if (primitive.typeId() == Type.TypeID.DECIMAL
- && tableSchemaPrimitiveType.typeId() == Type.TypeID.DECIMAL) {
- Types.DecimalType dataType = (Types.DecimalType) primitive;
- Types.DecimalType tableType = (Types.DecimalType) tableSchemaPrimitiveType;
- return dataType.scale() == tableType.scale() && dataType.precision() < tableType.precision()
- ? Result.DATA_CONVERSION_NEEDED
- : Result.SCHEMA_UPDATE_NEEDED;
} else {
return Result.SCHEMA_UPDATE_NEEDED;
}
}
+ /**
+ * Whether {@link DataConverter} can convert input data of type {@code dataType} into the table's
+ * {@code tableType}. Must stay in sync with the conversions {@link DataConverter#get} performs.
+ */
Review Comment:
I parameterized DataConverter's tests and add an assertion for isDataConversionPossible==true in it, so the predicate and the actual conversion are now tested together in the same test case.
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/EvolveSchemaVisitor.java:
##########
@@ -211,8 +211,15 @@ private void updateColumn(Types.NestedField existingField, Types.NestedField tar
String existingColumnName = this.existingSchema.findColumnName(existingField.fieldId());
boolean needsOptionalUpdate = targetField.isOptional() && existingField.isRequired();
+ boolean handledByDataConversion =
+ existingField.type().isPrimitiveType()
+ && targetField.type().isPrimitiveType()
+ && CompareSchemasVisitor.isDataConversionPossible(
+ targetField.type().asPrimitiveType(), existingField.type().asPrimitiveType());
boolean needsTypeUpdate =
- targetField.type().isPrimitiveType() && !targetField.type().equals(existingField.type());
+ targetField.type().isPrimitiveType()
+ && !targetField.type().equals(existingField.type())
+ && !handledByDataConversion;
Review Comment:
I don't think we can drop the `targetField.type().isPrimitiveType()` gate. This will let nested types go into the conversion logics, but they should be handled by the visitor's recursion instead. Removing this actually failed a few existing tests.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: handle simultaneous schema evolution and data conversion [iceberg]
Posted by "sqd (via GitHub)" <gi...@apache.org>.
sqd commented on code in PR #17024:
URL: https://github.com/apache/iceberg/pull/17024#discussion_r3537207498
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/CompareSchemasVisitor.java:
##########
@@ -180,27 +179,43 @@ public Result primitive(Type.PrimitiveType primitive, Integer tableSchemaId) {
Type.PrimitiveType tableSchemaPrimitiveType = tableSchemaType.asPrimitiveType();
if (primitive.equals(tableSchemaPrimitiveType)) {
return Result.SAME;
- } else if (primitive.equals(Types.IntegerType.get())
- && tableSchemaPrimitiveType.equals(Types.LongType.get())) {
- return Result.DATA_CONVERSION_NEEDED;
- } else if (primitive.equals(Types.FloatType.get())
- && tableSchemaPrimitiveType.equals(Types.DoubleType.get())) {
+ } else if (isDataConversionPossible(primitive, tableSchemaPrimitiveType)) {
return Result.DATA_CONVERSION_NEEDED;
- } else if (primitive.equals(Types.DateType.get())
- && tableSchemaPrimitiveType.equals(Types.TimestampType.withoutZone())) {
- return Result.DATA_CONVERSION_NEEDED;
- } else if (primitive.typeId() == Type.TypeID.DECIMAL
- && tableSchemaPrimitiveType.typeId() == Type.TypeID.DECIMAL) {
- Types.DecimalType dataType = (Types.DecimalType) primitive;
- Types.DecimalType tableType = (Types.DecimalType) tableSchemaPrimitiveType;
- return dataType.scale() == tableType.scale() && dataType.precision() < tableType.precision()
- ? Result.DATA_CONVERSION_NEEDED
- : Result.SCHEMA_UPDATE_NEEDED;
} else {
return Result.SCHEMA_UPDATE_NEEDED;
}
}
+ /**
+ * Whether {@link DataConverter} can convert input data of type {@code dataType} into the table's
+ * {@code tableType}. Must stay in sync with the conversions {@link DataConverter#get} performs.
+ */
+ @SuppressWarnings("checkstyle:CyclomaticComplexity")
+ static boolean isDataConversionPossible(
+ Type.PrimitiveType dataType, Type.PrimitiveType tableType) {
+ if (dataType.equals(Types.IntegerType.get()) && tableType.equals(Types.LongType.get())) {
+ return true;
+ }
+
+ if (dataType.equals(Types.FloatType.get()) && tableType.equals(Types.DoubleType.get())) {
+ return true;
+ }
+
+ if (dataType.equals(Types.DateType.get())
+ && tableType.equals(Types.TimestampType.withoutZone())) {
+ return true;
+ }
Review Comment:
promotion of timestamp nano was never handled by dynamic sink. This PR just makes the dynamic sink internally consistent. I think adding timestamp nano should be another PR (maybe into the same one we talked about that aligns dynamic sink with spec's promotion rules).
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: handle simultaneous schema evolution and data conversion [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #17024:
URL: https://github.com/apache/iceberg/pull/17024#discussion_r3548941770
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/EvolveSchemaVisitor.java:
##########
@@ -211,8 +211,15 @@ private void updateColumn(Types.NestedField existingField, Types.NestedField tar
String existingColumnName = this.existingSchema.findColumnName(existingField.fieldId());
boolean needsOptionalUpdate = targetField.isOptional() && existingField.isRequired();
+ boolean handledByDataConversion =
+ existingField.type().isPrimitiveType()
+ && targetField.type().isPrimitiveType()
+ && CompareSchemasVisitor.isDataConversionPossible(
+ targetField.type().asPrimitiveType(), existingField.type().asPrimitiveType());
boolean needsTypeUpdate =
- targetField.type().isPrimitiveType() && !targetField.type().equals(existingField.type());
+ targetField.type().isPrimitiveType()
+ && !targetField.type().equals(existingField.type())
+ && !handledByDataConversion;
Review Comment:
Fair. I'm still wondering, if a data conversion is not possible, then the schema update should always fail. Unrelated to this PR though.
##########
flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestRowDataConverter.java:
##########
@@ -86,57 +89,33 @@ void testAddRequiredColumn() {
.hasMessageContaining("is non-nullable but does not exist in source schema");
}
- @Test
- void testIntToLong() {
- Schema schemaWithLong =
- new Schema(
- Types.NestedField.optional(2, "id", Types.LongType.get()),
- Types.NestedField.optional(4, "data", Types.StringType.get()));
-
- assertThat(convert(SimpleDataUtil.createRowData(1, "a"), SimpleDataUtil.SCHEMA, schemaWithLong))
- .isEqualTo(GenericRowData.of(1L, StringData.fromString("a")));
- }
-
- @Test
- void testFloatToDouble() {
- Schema schemaWithFloat =
- new Schema(Types.NestedField.optional(1, "float2double", Types.FloatType.get()));
- Schema schemaWithDouble =
- new Schema(Types.NestedField.optional(2, "float2double", Types.DoubleType.get()));
-
- assertThat(convert(GenericRowData.of(1.5f), schemaWithFloat, schemaWithDouble))
- .isEqualTo(GenericRowData.of(1.5d));
- }
+ @ParameterizedTest
+ @MethodSource("dataConversionCases")
+ void testConversionsDeclaredByCompareSchemasVisitorAreSupported(
+ Type.PrimitiveType dataType, Type.PrimitiveType tableType, Object input, Object expected) {
+ Schema dataSchema = new Schema(optional(1, "field", dataType));
+ Schema tableSchema = new Schema(optional(2, "field", tableType));
- @Test
- void testDateToTimestamp() {
- Schema schemaWithFloat =
- new Schema(Types.NestedField.optional(1, "date2timestamp", Types.DateType.get()));
- Schema schemaWithDouble =
- new Schema(
- Types.NestedField.optional(2, "date2timestamp", Types.TimestampType.withoutZone()));
-
- DateTime time = new DateTime(2022, 1, 10, 0, 0, 0, 0, DateTimeZone.UTC);
- int days =
- Days.daysBetween(new DateTime(1970, 1, 1, 0, 0, 0, 0, DateTimeZone.UTC), time).getDays();
-
- assertThat(convert(GenericRowData.of(days), schemaWithFloat, schemaWithDouble))
- .isEqualTo(GenericRowData.of(TimestampData.fromEpochMillis(time.getMillis())));
+ assertThat(CompareSchemasVisitor.isDataConversionPossible(dataType, tableType)).isTrue();
+ assertThat(convert(GenericRowData.of(input), dataSchema, tableSchema))
+ .isEqualTo(GenericRowData.of(expected));
}
- @Test
- void testIncreasePrecision() {
- Schema before =
- new Schema(Types.NestedField.required(14, "decimal_field", Types.DecimalType.of(9, 2)));
- Schema after =
- new Schema(Types.NestedField.required(14, "decimal_field", Types.DecimalType.of(10, 2)));
-
- assertThat(
- convert(
- GenericRowData.of(DecimalData.fromBigDecimal(new BigDecimal("-1.50"), 9, 2)),
- before,
- after))
- .isEqualTo(GenericRowData.of(DecimalData.fromBigDecimal(new BigDecimal("-1.50"), 10, 2)));
+ private static Stream<Arguments> dataConversionCases() {
+ LocalDate date = LocalDate.of(2022, 1, 10);
+ return Stream.of(
+ Arguments.of(Types.IntegerType.get(), Types.LongType.get(), 1, 1L),
+ Arguments.of(Types.FloatType.get(), Types.DoubleType.get(), 1.5f, 1.5d),
+ Arguments.of(
+ Types.DateType.get(),
+ Types.TimestampType.withoutZone(),
+ (int) date.toEpochDay(),
+ TimestampData.fromLocalDateTime(date.atStartOfDay())),
+ Arguments.of(
+ Types.DecimalType.of(9, 2),
+ Types.DecimalType.of(10, 2),
+ DecimalData.fromBigDecimal(new BigDecimal("-1.50"), 9, 2),
+ DecimalData.fromBigDecimal(new BigDecimal("-1.50"), 10, 2)));
Review Comment:
I prefer spelling these out as test cases, like previously. We can still keep the new `testConversionsDeclaredByCompareSchemasVisitorAreSupported` method.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: handle simultaneous schema evolution and data conversion [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #17024:
URL: https://github.com/apache/iceberg/pull/17024#discussion_r3535230626
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/CompareSchemasVisitor.java:
##########
@@ -180,27 +179,43 @@ public Result primitive(Type.PrimitiveType primitive, Integer tableSchemaId) {
Type.PrimitiveType tableSchemaPrimitiveType = tableSchemaType.asPrimitiveType();
if (primitive.equals(tableSchemaPrimitiveType)) {
return Result.SAME;
- } else if (primitive.equals(Types.IntegerType.get())
- && tableSchemaPrimitiveType.equals(Types.LongType.get())) {
- return Result.DATA_CONVERSION_NEEDED;
- } else if (primitive.equals(Types.FloatType.get())
- && tableSchemaPrimitiveType.equals(Types.DoubleType.get())) {
+ } else if (isDataConversionPossible(primitive, tableSchemaPrimitiveType)) {
return Result.DATA_CONVERSION_NEEDED;
- } else if (primitive.equals(Types.DateType.get())
- && tableSchemaPrimitiveType.equals(Types.TimestampType.withoutZone())) {
- return Result.DATA_CONVERSION_NEEDED;
- } else if (primitive.typeId() == Type.TypeID.DECIMAL
- && tableSchemaPrimitiveType.typeId() == Type.TypeID.DECIMAL) {
- Types.DecimalType dataType = (Types.DecimalType) primitive;
- Types.DecimalType tableType = (Types.DecimalType) tableSchemaPrimitiveType;
- return dataType.scale() == tableType.scale() && dataType.precision() < tableType.precision()
- ? Result.DATA_CONVERSION_NEEDED
- : Result.SCHEMA_UPDATE_NEEDED;
} else {
return Result.SCHEMA_UPDATE_NEEDED;
}
}
+ /**
+ * Whether {@link DataConverter} can convert input data of type {@code dataType} into the table's
+ * {@code tableType}. Must stay in sync with the conversions {@link DataConverter#get} performs.
+ */
+ @SuppressWarnings("checkstyle:CyclomaticComplexity")
+ static boolean isDataConversionPossible(
+ Type.PrimitiveType dataType, Type.PrimitiveType tableType) {
+ if (dataType.equals(Types.IntegerType.get()) && tableType.equals(Types.LongType.get())) {
+ return true;
+ }
+
+ if (dataType.equals(Types.FloatType.get()) && tableType.equals(Types.DoubleType.get())) {
+ return true;
+ }
+
+ if (dataType.equals(Types.DateType.get())
+ && tableType.equals(Types.TimestampType.withoutZone())) {
+ return true;
+ }
Review Comment:
What about timestamp nano?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: handle simultaneous schema evolution and data conversion [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #17024:
URL: https://github.com/apache/iceberg/pull/17024#discussion_r3537542200
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/CompareSchemasVisitor.java:
##########
@@ -180,27 +179,43 @@ public Result primitive(Type.PrimitiveType primitive, Integer tableSchemaId) {
Type.PrimitiveType tableSchemaPrimitiveType = tableSchemaType.asPrimitiveType();
if (primitive.equals(tableSchemaPrimitiveType)) {
return Result.SAME;
- } else if (primitive.equals(Types.IntegerType.get())
- && tableSchemaPrimitiveType.equals(Types.LongType.get())) {
- return Result.DATA_CONVERSION_NEEDED;
- } else if (primitive.equals(Types.FloatType.get())
- && tableSchemaPrimitiveType.equals(Types.DoubleType.get())) {
+ } else if (isDataConversionPossible(primitive, tableSchemaPrimitiveType)) {
return Result.DATA_CONVERSION_NEEDED;
- } else if (primitive.equals(Types.DateType.get())
- && tableSchemaPrimitiveType.equals(Types.TimestampType.withoutZone())) {
- return Result.DATA_CONVERSION_NEEDED;
- } else if (primitive.typeId() == Type.TypeID.DECIMAL
- && tableSchemaPrimitiveType.typeId() == Type.TypeID.DECIMAL) {
- Types.DecimalType dataType = (Types.DecimalType) primitive;
- Types.DecimalType tableType = (Types.DecimalType) tableSchemaPrimitiveType;
- return dataType.scale() == tableType.scale() && dataType.precision() < tableType.precision()
- ? Result.DATA_CONVERSION_NEEDED
- : Result.SCHEMA_UPDATE_NEEDED;
} else {
return Result.SCHEMA_UPDATE_NEEDED;
}
}
+ /**
+ * Whether {@link DataConverter} can convert input data of type {@code dataType} into the table's
+ * {@code tableType}. Must stay in sync with the conversions {@link DataConverter#get} performs.
+ */
+ @SuppressWarnings("checkstyle:CyclomaticComplexity")
+ static boolean isDataConversionPossible(
+ Type.PrimitiveType dataType, Type.PrimitiveType tableType) {
+ if (dataType.equals(Types.IntegerType.get()) && tableType.equals(Types.LongType.get())) {
+ return true;
+ }
+
+ if (dataType.equals(Types.FloatType.get()) && tableType.equals(Types.DoubleType.get())) {
+ return true;
+ }
+
+ if (dataType.equals(Types.DateType.get())
+ && tableType.equals(Types.TimestampType.withoutZone())) {
+ return true;
+ }
Review Comment:
I'm ok fixing/testing the dynsink with nano in a different PR.
Spec propotion rules are a bigger task, as they might have requirements outside Flink
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/CompareSchemasVisitor.java:
##########
@@ -180,27 +179,43 @@ public Result primitive(Type.PrimitiveType primitive, Integer tableSchemaId) {
Type.PrimitiveType tableSchemaPrimitiveType = tableSchemaType.asPrimitiveType();
if (primitive.equals(tableSchemaPrimitiveType)) {
return Result.SAME;
- } else if (primitive.equals(Types.IntegerType.get())
- && tableSchemaPrimitiveType.equals(Types.LongType.get())) {
- return Result.DATA_CONVERSION_NEEDED;
- } else if (primitive.equals(Types.FloatType.get())
- && tableSchemaPrimitiveType.equals(Types.DoubleType.get())) {
+ } else if (isDataConversionPossible(primitive, tableSchemaPrimitiveType)) {
return Result.DATA_CONVERSION_NEEDED;
- } else if (primitive.equals(Types.DateType.get())
- && tableSchemaPrimitiveType.equals(Types.TimestampType.withoutZone())) {
- return Result.DATA_CONVERSION_NEEDED;
- } else if (primitive.typeId() == Type.TypeID.DECIMAL
- && tableSchemaPrimitiveType.typeId() == Type.TypeID.DECIMAL) {
- Types.DecimalType dataType = (Types.DecimalType) primitive;
- Types.DecimalType tableType = (Types.DecimalType) tableSchemaPrimitiveType;
- return dataType.scale() == tableType.scale() && dataType.precision() < tableType.precision()
- ? Result.DATA_CONVERSION_NEEDED
- : Result.SCHEMA_UPDATE_NEEDED;
} else {
return Result.SCHEMA_UPDATE_NEEDED;
}
}
+ /**
+ * Whether {@link DataConverter} can convert input data of type {@code dataType} into the table's
+ * {@code tableType}. Must stay in sync with the conversions {@link DataConverter#get} performs.
+ */
+ @SuppressWarnings("checkstyle:CyclomaticComplexity")
+ static boolean isDataConversionPossible(
+ Type.PrimitiveType dataType, Type.PrimitiveType tableType) {
+ if (dataType.equals(Types.IntegerType.get()) && tableType.equals(Types.LongType.get())) {
+ return true;
+ }
+
+ if (dataType.equals(Types.FloatType.get()) && tableType.equals(Types.DoubleType.get())) {
+ return true;
+ }
+
+ if (dataType.equals(Types.DateType.get())
+ && tableType.equals(Types.TimestampType.withoutZone())) {
+ return true;
+ }
Review Comment:
I'm ok fixing/testing the dynsink with nano in a different PR.
Spec promotion rules are a bigger task, as they might have requirements outside Flink
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: handle simultaneous schema evolution and data conversion [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on PR #17024:
URL: https://github.com/apache/iceberg/pull/17024#issuecomment-4905260755
@mxm, @swapna267: Do you want to review?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: handle simultaneous schema evolution and data conversion [iceberg]
Posted by "wombatu-kun (via GitHub)" <gi...@apache.org>.
wombatu-kun commented on code in PR #17024:
URL: https://github.com/apache/iceberg/pull/17024#discussion_r3510173697
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/EvolveSchemaVisitor.java:
##########
@@ -211,8 +212,15 @@ private void updateColumn(Types.NestedField existingField, Types.NestedField tar
String existingColumnName = this.existingSchema.findColumnName(existingField.fieldId());
boolean needsOptionalUpdate = targetField.isOptional() && existingField.isRequired();
+ boolean narrowingHandledByConversion =
+ existingField.type().isPrimitiveType()
+ && targetField.type().isPrimitiveType()
+ && TypeUtil.isPromotionAllowed(
Review Comment:
`TypeUtil.isPromotionAllowed` does not cover date -> timestamp_without_zone, but `CompareSchemasVisitor.primitive` treats it as `DATA_CONVERSION_NEEDED` and `DataConverter.get` converts it. So a record that both adds a column (forcing SCHEMA_UPDATE_NEEDED) and carries a date for a timestamp table column still reaches `api.updateColumn(col, date)` here and throws `Cannot change column type: ...: timestamp -> date` - the same failure this fix removes for int->long and float->double. Mirror the conversion set `CompareSchemasVisitor` recognizes (add date -> timestamp), ideally by extracting a single shared predicate both visitors call so the two cannot drift, and add a test for the add-column + date->timestamp case.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: handle simultaneous schema evolution and data conversion [iceberg]
Posted by "sqd (via GitHub)" <gi...@apache.org>.
sqd commented on code in PR #17024:
URL: https://github.com/apache/iceberg/pull/17024#discussion_r3510230597
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/EvolveSchemaVisitor.java:
##########
@@ -211,8 +212,15 @@ private void updateColumn(Types.NestedField existingField, Types.NestedField tar
String existingColumnName = this.existingSchema.findColumnName(existingField.fieldId());
boolean needsOptionalUpdate = targetField.isOptional() && existingField.isRequired();
+ boolean narrowingHandledByConversion =
+ existingField.type().isPrimitiveType()
+ && targetField.type().isPrimitiveType()
+ && TypeUtil.isPromotionAllowed(
Review Comment:
Sounds like this is an issue with `TypeUtil.isPromotionAllowed` missing the timestamp case?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: handle simultaneous schema evolution and data conversion [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #17024:
URL: https://github.com/apache/iceberg/pull/17024#discussion_r3529253081
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/EvolveSchemaVisitor.java:
##########
@@ -211,8 +212,15 @@ private void updateColumn(Types.NestedField existingField, Types.NestedField tar
String existingColumnName = this.existingSchema.findColumnName(existingField.fieldId());
boolean needsOptionalUpdate = targetField.isOptional() && existingField.isRequired();
+ boolean narrowingHandledByConversion =
+ existingField.type().isPrimitiveType()
+ && targetField.type().isPrimitiveType()
+ && TypeUtil.isPromotionAllowed(
Review Comment:
IIRC the Date -> timestamp_without_zone promotion is added later to the spec, but the implementation might not be there yet.
We might want to fix the functionality around the full Iceberg implementation, or remove the functionality from the DynamicSink too.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: handle simultaneous schema evolution and data conversion [iceberg]
Posted by "sqd (via GitHub)" <gi...@apache.org>.
sqd commented on code in PR #17024:
URL: https://github.com/apache/iceberg/pull/17024#discussion_r3561749721
##########
flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestRowDataConverter.java:
##########
@@ -86,57 +89,33 @@ void testAddRequiredColumn() {
.hasMessageContaining("is non-nullable but does not exist in source schema");
}
- @Test
- void testIntToLong() {
- Schema schemaWithLong =
- new Schema(
- Types.NestedField.optional(2, "id", Types.LongType.get()),
- Types.NestedField.optional(4, "data", Types.StringType.get()));
-
- assertThat(convert(SimpleDataUtil.createRowData(1, "a"), SimpleDataUtil.SCHEMA, schemaWithLong))
- .isEqualTo(GenericRowData.of(1L, StringData.fromString("a")));
- }
-
- @Test
- void testFloatToDouble() {
- Schema schemaWithFloat =
- new Schema(Types.NestedField.optional(1, "float2double", Types.FloatType.get()));
- Schema schemaWithDouble =
- new Schema(Types.NestedField.optional(2, "float2double", Types.DoubleType.get()));
-
- assertThat(convert(GenericRowData.of(1.5f), schemaWithFloat, schemaWithDouble))
- .isEqualTo(GenericRowData.of(1.5d));
- }
+ @ParameterizedTest
+ @MethodSource("dataConversionCases")
+ void testConversionsDeclaredByCompareSchemasVisitorAreSupported(
+ Type.PrimitiveType dataType, Type.PrimitiveType tableType, Object input, Object expected) {
+ Schema dataSchema = new Schema(optional(1, "field", dataType));
+ Schema tableSchema = new Schema(optional(2, "field", tableType));
- @Test
- void testDateToTimestamp() {
- Schema schemaWithFloat =
- new Schema(Types.NestedField.optional(1, "date2timestamp", Types.DateType.get()));
- Schema schemaWithDouble =
- new Schema(
- Types.NestedField.optional(2, "date2timestamp", Types.TimestampType.withoutZone()));
-
- DateTime time = new DateTime(2022, 1, 10, 0, 0, 0, 0, DateTimeZone.UTC);
- int days =
- Days.daysBetween(new DateTime(1970, 1, 1, 0, 0, 0, 0, DateTimeZone.UTC), time).getDays();
-
- assertThat(convert(GenericRowData.of(days), schemaWithFloat, schemaWithDouble))
- .isEqualTo(GenericRowData.of(TimestampData.fromEpochMillis(time.getMillis())));
+ assertThat(CompareSchemasVisitor.isDataConversionPossible(dataType, tableType)).isTrue();
+ assertThat(convert(GenericRowData.of(input), dataSchema, tableSchema))
+ .isEqualTo(GenericRowData.of(expected));
}
- @Test
- void testIncreasePrecision() {
- Schema before =
- new Schema(Types.NestedField.required(14, "decimal_field", Types.DecimalType.of(9, 2)));
- Schema after =
- new Schema(Types.NestedField.required(14, "decimal_field", Types.DecimalType.of(10, 2)));
-
- assertThat(
- convert(
- GenericRowData.of(DecimalData.fromBigDecimal(new BigDecimal("-1.50"), 9, 2)),
- before,
- after))
- .isEqualTo(GenericRowData.of(DecimalData.fromBigDecimal(new BigDecimal("-1.50"), 10, 2)));
+ private static Stream<Arguments> dataConversionCases() {
+ LocalDate date = LocalDate.of(2022, 1, 10);
+ return Stream.of(
+ Arguments.of(Types.IntegerType.get(), Types.LongType.get(), 1, 1L),
+ Arguments.of(Types.FloatType.get(), Types.DoubleType.get(), 1.5f, 1.5d),
+ Arguments.of(
+ Types.DateType.get(),
+ Types.TimestampType.withoutZone(),
+ (int) date.toEpochDay(),
+ TimestampData.fromLocalDateTime(date.atStartOfDay())),
+ Arguments.of(
+ Types.DecimalType.of(9, 2),
+ Types.DecimalType.of(10, 2),
+ DecimalData.fromBigDecimal(new BigDecimal("-1.50"), 9, 2),
+ DecimalData.fromBigDecimal(new BigDecimal("-1.50"), 10, 2)));
Review Comment:
done
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: handle simultaneous schema evolution and data conversion [iceberg]
Posted by "sqd (via GitHub)" <gi...@apache.org>.
sqd commented on code in PR #17024:
URL: https://github.com/apache/iceberg/pull/17024#discussion_r3561830871
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/EvolveSchemaVisitor.java:
##########
@@ -211,8 +211,15 @@ private void updateColumn(Types.NestedField existingField, Types.NestedField tar
String existingColumnName = this.existingSchema.findColumnName(existingField.fieldId());
boolean needsOptionalUpdate = targetField.isOptional() && existingField.isRequired();
+ boolean handledByDataConversion =
+ existingField.type().isPrimitiveType()
+ && targetField.type().isPrimitiveType()
+ && CompareSchemasVisitor.isDataConversionPossible(
+ targetField.type().asPrimitiveType(), existingField.type().asPrimitiveType());
boolean needsTypeUpdate =
- targetField.type().isPrimitiveType() && !targetField.type().equals(existingField.type());
+ targetField.type().isPrimitiveType()
+ && !targetField.type().equals(existingField.type())
+ && !handledByDataConversion;
Review Comment:
I am not sure I understand. For example, if both types are structs, then data conversion is not possible. But we shouldn't fail schema update in this case, because that is handled by the visitor recursing into the struct, not by data conversion.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: handle simultaneous schema evolution and data conversion [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary merged PR #17024:
URL: https://github.com/apache/iceberg/pull/17024
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: handle simultaneous schema evolution and data conversion [iceberg]
Posted by "sqd (via GitHub)" <gi...@apache.org>.
sqd commented on code in PR #17024:
URL: https://github.com/apache/iceberg/pull/17024#discussion_r3529817746
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/EvolveSchemaVisitor.java:
##########
@@ -211,8 +212,15 @@ private void updateColumn(Types.NestedField existingField, Types.NestedField tar
String existingColumnName = this.existingSchema.findColumnName(existingField.fieldId());
boolean needsOptionalUpdate = targetField.isOptional() && existingField.isRequired();
+ boolean narrowingHandledByConversion =
+ existingField.type().isPrimitiveType()
+ && targetField.type().isPrimitiveType()
+ && TypeUtil.isPromotionAllowed(
Review Comment:
I see. I'll add `date -> timestamp` into this PR so we are consistent within the dynamic sink, then in a later change we can make the decision to fix the functionality around the full Iceberg implementation, or remove the functionality from the DynamicSink too. Thank you @pvary !
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: handle simultaneous schema evolution and data conversion [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on PR #17024:
URL: https://github.com/apache/iceberg/pull/17024#issuecomment-4906714005
Ack, I'll take a look.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: handle simultaneous schema evolution and data conversion [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #17024:
URL: https://github.com/apache/iceberg/pull/17024#discussion_r3541139533
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/EvolveSchemaVisitor.java:
##########
@@ -211,8 +211,15 @@ private void updateColumn(Types.NestedField existingField, Types.NestedField tar
String existingColumnName = this.existingSchema.findColumnName(existingField.fieldId());
boolean needsOptionalUpdate = targetField.isOptional() && existingField.isRequired();
+ boolean handledByDataConversion =
+ existingField.type().isPrimitiveType()
+ && targetField.type().isPrimitiveType()
+ && CompareSchemasVisitor.isDataConversionPossible(
+ targetField.type().asPrimitiveType(), existingField.type().asPrimitiveType());
boolean needsTypeUpdate =
- targetField.type().isPrimitiveType() && !targetField.type().equals(existingField.type());
+ targetField.type().isPrimitiveType()
+ && !targetField.type().equals(existingField.type())
+ && !handledByDataConversion;
Review Comment:
We can simplify to:
```suggestion
!handledByDataConversion
&& !targetField.type().equals(existingField.type());
```
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/EvolveSchemaVisitor.java:
##########
@@ -211,8 +211,15 @@ private void updateColumn(Types.NestedField existingField, Types.NestedField tar
String existingColumnName = this.existingSchema.findColumnName(existingField.fieldId());
boolean needsOptionalUpdate = targetField.isOptional() && existingField.isRequired();
+ boolean handledByDataConversion =
+ existingField.type().isPrimitiveType()
+ && targetField.type().isPrimitiveType()
+ && CompareSchemasVisitor.isDataConversionPossible(
+ targetField.type().asPrimitiveType(), existingField.type().asPrimitiveType());
Review Comment:
Can we move the `existingField.type().isPrimitiveType() && targetField.type().isPrimitiveType()` check into the first lines of `isDataConversionPossible`?
##########
flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java:
##########
@@ -174,6 +178,102 @@ void testInvalidateOldCacheEntryOnUpdate() {
.isTrue();
}
+ @Test
+ void testAddColumnWithResidualNarrowingEvolvesThenConverts() {
+ Catalog catalog = CATALOG_EXTENSION.catalog();
+ TableIdentifier tableIdentifier = TableIdentifier.parse("default.myTable");
+ Schema tableSchema =
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "data", Types.StringType.get()));
+ catalog.createTable(tableIdentifier, tableSchema);
+
+ TableMetadataCache cache =
+ new TableMetadataCache(catalog, 10, Long.MAX_VALUE, 10, CASE_SENSITIVE, PRESERVE_COLUMNS);
+ TableUpdater tableUpdater = new TableUpdater(cache, catalog, CASE_SENSITIVE, PRESERVE_COLUMNS);
+
+ Schema rowSchema =
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "data", Types.StringType.get()),
+ Types.NestedField.optional(3, "extra", Types.StringType.get()));
+
+ Tuple2<TableMetadataCache.ResolvedSchemaInfo, PartitionSpec> result =
+ tableUpdater.update(
+ tableIdentifier,
+ SnapshotRef.MAIN_BRANCH,
+ rowSchema,
+ PartitionSpec.unpartitioned(),
+ TableCreator.DEFAULT);
+
+ Schema evolved = catalog.loadTable(tableIdentifier).schema();
+ assertThat(evolved.findField("id").type()).isEqualTo(Types.LongType.get());
+ assertThat(evolved.findField("extra")).isNotNull();
+ assertThat(result.f0.compareResult())
+ .isEqualTo(CompareSchemasVisitor.Result.DATA_CONVERSION_NEEDED);
+
+ RowData converted =
+ (RowData)
+ result
+ .f0
+ .recordConverter()
+ .convert(
+ GenericRowData.of(5, StringData.fromString("a"), StringData.fromString("x")));
+ assertThat(converted.getLong(0)).isEqualTo(5L);
+ assertThat(converted.getString(1)).hasToString("a");
+ assertThat(converted.getString(2)).hasToString("x");
+ }
+
+ @Test
+ void testAddColumnWithResidualDateToTimestampEvolvesThenConverts() {
Review Comment:
```suggestion
void testAddColumnWithDateToTimestampConversion() {
```
##########
flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java:
##########
@@ -174,6 +178,102 @@ void testInvalidateOldCacheEntryOnUpdate() {
.isTrue();
}
+ @Test
+ void testAddColumnWithResidualNarrowingEvolvesThenConverts() {
Review Comment:
```suggestion
void testAddColumnWithDataConverterNarrowing() {
```
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/CompareSchemasVisitor.java:
##########
@@ -180,27 +179,43 @@ public Result primitive(Type.PrimitiveType primitive, Integer tableSchemaId) {
Type.PrimitiveType tableSchemaPrimitiveType = tableSchemaType.asPrimitiveType();
if (primitive.equals(tableSchemaPrimitiveType)) {
return Result.SAME;
- } else if (primitive.equals(Types.IntegerType.get())
- && tableSchemaPrimitiveType.equals(Types.LongType.get())) {
- return Result.DATA_CONVERSION_NEEDED;
- } else if (primitive.equals(Types.FloatType.get())
- && tableSchemaPrimitiveType.equals(Types.DoubleType.get())) {
+ } else if (isDataConversionPossible(primitive, tableSchemaPrimitiveType)) {
return Result.DATA_CONVERSION_NEEDED;
- } else if (primitive.equals(Types.DateType.get())
- && tableSchemaPrimitiveType.equals(Types.TimestampType.withoutZone())) {
- return Result.DATA_CONVERSION_NEEDED;
- } else if (primitive.typeId() == Type.TypeID.DECIMAL
- && tableSchemaPrimitiveType.typeId() == Type.TypeID.DECIMAL) {
- Types.DecimalType dataType = (Types.DecimalType) primitive;
- Types.DecimalType tableType = (Types.DecimalType) tableSchemaPrimitiveType;
- return dataType.scale() == tableType.scale() && dataType.precision() < tableType.precision()
- ? Result.DATA_CONVERSION_NEEDED
- : Result.SCHEMA_UPDATE_NEEDED;
} else {
return Result.SCHEMA_UPDATE_NEEDED;
}
}
+ /**
+ * Whether {@link DataConverter} can convert input data of type {@code dataType} into the table's
+ * {@code tableType}. Must stay in sync with the conversions {@link DataConverter#get} performs.
+ */
Review Comment:
Do we have a test to enforce this?
##########
flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java:
##########
@@ -174,6 +178,102 @@ void testInvalidateOldCacheEntryOnUpdate() {
.isTrue();
}
+ @Test
+ void testAddColumnWithResidualNarrowingEvolvesThenConverts() {
+ Catalog catalog = CATALOG_EXTENSION.catalog();
+ TableIdentifier tableIdentifier = TableIdentifier.parse("default.myTable");
+ Schema tableSchema =
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "data", Types.StringType.get()));
+ catalog.createTable(tableIdentifier, tableSchema);
+
+ TableMetadataCache cache =
+ new TableMetadataCache(catalog, 10, Long.MAX_VALUE, 10, CASE_SENSITIVE, PRESERVE_COLUMNS);
+ TableUpdater tableUpdater = new TableUpdater(cache, catalog, CASE_SENSITIVE, PRESERVE_COLUMNS);
+
+ Schema rowSchema =
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "data", Types.StringType.get()),
+ Types.NestedField.optional(3, "extra", Types.StringType.get()));
+
+ Tuple2<TableMetadataCache.ResolvedSchemaInfo, PartitionSpec> result =
+ tableUpdater.update(
+ tableIdentifier,
+ SnapshotRef.MAIN_BRANCH,
+ rowSchema,
+ PartitionSpec.unpartitioned(),
+ TableCreator.DEFAULT);
+
+ Schema evolved = catalog.loadTable(tableIdentifier).schema();
+ assertThat(evolved.findField("id").type()).isEqualTo(Types.LongType.get());
+ assertThat(evolved.findField("extra")).isNotNull();
+ assertThat(result.f0.compareResult())
+ .isEqualTo(CompareSchemasVisitor.Result.DATA_CONVERSION_NEEDED);
+
+ RowData converted =
+ (RowData)
+ result
+ .f0
+ .recordConverter()
+ .convert(
+ GenericRowData.of(5, StringData.fromString("a"), StringData.fromString("x")));
+ assertThat(converted.getLong(0)).isEqualTo(5L);
+ assertThat(converted.getString(1)).hasToString("a");
+ assertThat(converted.getString(2)).hasToString("x");
+ }
+
+ @Test
+ void testAddColumnWithResidualDateToTimestampEvolvesThenConverts() {
+ Catalog catalog = CATALOG_EXTENSION.catalog();
+ TableIdentifier tableIdentifier = TableIdentifier.parse("default.myTable");
+ Schema tableSchema =
+ new Schema(
+ Types.NestedField.optional(1, "ts", Types.TimestampType.withoutZone()),
+ Types.NestedField.optional(2, "data", Types.StringType.get()));
+ catalog.createTable(tableIdentifier, tableSchema);
+
+ TableMetadataCache cache =
+ new TableMetadataCache(catalog, 10, Long.MAX_VALUE, 10, CASE_SENSITIVE, PRESERVE_COLUMNS);
+ TableUpdater tableUpdater = new TableUpdater(cache, catalog, CASE_SENSITIVE, PRESERVE_COLUMNS);
+
+ Schema rowSchema =
+ new Schema(
+ Types.NestedField.optional(1, "ts", Types.DateType.get()),
+ Types.NestedField.optional(2, "data", Types.StringType.get()),
+ Types.NestedField.optional(3, "extra", Types.StringType.get()));
+
+ Tuple2<TableMetadataCache.ResolvedSchemaInfo, PartitionSpec> result =
+ tableUpdater.update(
+ tableIdentifier,
+ SnapshotRef.MAIN_BRANCH,
+ rowSchema,
+ PartitionSpec.unpartitioned(),
+ TableCreator.DEFAULT);
Review Comment:
Could we move the common test logic for this and the above test to a method?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org