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