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: [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]
