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