diff --git a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/CompareSchemasVisitor.java b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/CompareSchemasVisitor.java index cb4fc62c01b8..6c1b2e5673e9 100644 --- a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/CompareSchemasVisitor.java +++ b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/CompareSchemasVisitor.java @@ -166,41 +166,54 @@ public Result map( } @Override - @SuppressWarnings("checkstyle:CyclomaticComplexity") public Result primitive(Type.PrimitiveType primitive, Integer tableSchemaId) { if (tableSchemaId == null) { return Result.SCHEMA_UPDATE_NEEDED; } Type tableSchemaType = tableSchema.findField(tableSchemaId).type(); - if (!tableSchemaType.isPrimitiveType()) { - return Result.SCHEMA_UPDATE_NEEDED; - } - - Type.PrimitiveType tableSchemaPrimitiveType = tableSchemaType.asPrimitiveType(); - if (primitive.equals(tableSchemaPrimitiveType)) { + if (primitive.equals(tableSchemaType)) { 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, tableSchemaType)) { 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 dataType, Type tableType) { + if (!dataType.isPrimitiveType() || !tableType.isPrimitiveType()) { + return false; + } + + 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; + } + + if (dataType.typeId() == Type.TypeID.DECIMAL && tableType.typeId() == Type.TypeID.DECIMAL) { + Types.DecimalType dataDecimalType = (Types.DecimalType) dataType; + Types.DecimalType tableDecimalType = (Types.DecimalType) tableType; + return dataDecimalType.scale() == tableDecimalType.scale() + && dataDecimalType.precision() < tableDecimalType.precision(); + } + + return false; + } + static class PartnerIdByNameAccessors implements PartnerAccessors { private final Schema tableSchema; private boolean caseSensitive; diff --git a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/EvolveSchemaVisitor.java b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/EvolveSchemaVisitor.java index d9747d201e3d..a5d6c8776a44 100644 --- a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/EvolveSchemaVisitor.java +++ b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/EvolveSchemaVisitor.java @@ -211,8 +211,12 @@ private void updateColumn(Types.NestedField existingField, Types.NestedField tar String existingColumnName = this.existingSchema.findColumnName(existingField.fieldId()); boolean needsOptionalUpdate = targetField.isOptional() && existingField.isRequired(); + boolean handledByDataConversion = + CompareSchemasVisitor.isDataConversionPossible(targetField.type(), existingField.type()); boolean needsTypeUpdate = - targetField.type().isPrimitiveType() && !targetField.type().equals(existingField.type()); + targetField.type().isPrimitiveType() + && !targetField.type().equals(existingField.type()) + && !handledByDataConversion; boolean needsDocUpdate = targetField.doc() != null && !targetField.doc().equals(existingField.doc()); diff --git a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestEvolveSchemaVisitor.java b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestEvolveSchemaVisitor.java index d2da73c66973..55e52751febb 100644 --- a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestEvolveSchemaVisitor.java +++ b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestEvolveSchemaVisitor.java @@ -479,20 +479,48 @@ public void testTypePromoteFloatToDouble() { } @Test - public void testInvalidTypePromoteDoubleToFloat() { + public void testTypeNarrowDoubleToFloatIsLeftToConversion() { Schema currentSchema = new Schema(required(1, "aCol", DoubleType.get())); Schema targetSchema = new Schema(required(1, "aCol", FloatType.get())); - assertThatThrownBy( - () -> - EvolveSchemaVisitor.visit( - TABLE, - loadUpdateApi(currentSchema), - currentSchema, - targetSchema, - true, - PRESERVE_COLUMNS)) - .hasMessage("Cannot change column type: aCol: double -> float") - .isInstanceOf(IllegalArgumentException.class); + + UpdateSchema updateApi = loadUpdateApi(currentSchema); + EvolveSchemaVisitor.visit( + TABLE, updateApi, currentSchema, targetSchema, CASE_SENSITIVE, PRESERVE_COLUMNS); + Schema applied = updateApi.apply(); + + assertThat(applied.findField("aCol").type()).isEqualTo(DoubleType.get()); + } + + @Test + public void testAddColumnWithResidualNarrowingDoesNotFail() { + Schema currentSchema = new Schema(required(1, "id", LongType.get())); + Schema targetSchema = + new Schema(required(1, "id", IntegerType.get()), optional(2, "age", IntegerType.get())); + + UpdateSchema updateApi = loadUpdateApi(currentSchema); + EvolveSchemaVisitor.visit( + TABLE, updateApi, currentSchema, targetSchema, CASE_SENSITIVE, PRESERVE_COLUMNS); + Schema applied = updateApi.apply(); + + assertThat(applied.findField("id").type()).isEqualTo(LongType.get()); + assertThat(applied.findField("age")).isNotNull(); + assertThat(applied.findField("age").type()).isEqualTo(IntegerType.get()); + } + + @Test + public void testAddColumnWithResidualDateToTimestampDoesNotFail() { + Schema currentSchema = new Schema(required(1, "ts", Types.TimestampType.withoutZone())); + Schema targetSchema = + new Schema(required(1, "ts", Types.DateType.get()), optional(2, "age", IntegerType.get())); + + UpdateSchema updateApi = loadUpdateApi(currentSchema); + EvolveSchemaVisitor.visit( + TABLE, updateApi, currentSchema, targetSchema, CASE_SENSITIVE, PRESERVE_COLUMNS); + Schema applied = updateApi.apply(); + + assertThat(applied.findField("ts").type()).isEqualTo(Types.TimestampType.withoutZone()); + assertThat(applied.findField("age")).isNotNull(); + assertThat(applied.findField("age").type()).isEqualTo(IntegerType.get()); } @Test diff --git a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestRowDataConverter.java b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestRowDataConverter.java index c4a86bb79e4a..7c179f793af7 100644 --- a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestRowDataConverter.java +++ b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestRowDataConverter.java @@ -24,6 +24,8 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy; import java.math.BigDecimal; +import java.time.LocalDate; +import java.util.stream.Stream; import org.apache.flink.table.data.DecimalData; import org.apache.flink.table.data.GenericArrayData; import org.apache.flink.table.data.GenericMapData; @@ -37,11 +39,12 @@ import org.apache.iceberg.flink.FlinkSchemaUtil; import org.apache.iceberg.flink.SimpleDataUtil; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; +import org.apache.iceberg.types.Type; import org.apache.iceberg.types.Types; -import org.joda.time.DateTime; -import org.joda.time.DateTimeZone; -import org.joda.time.Days; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; class TestRowDataConverter { @@ -110,18 +113,18 @@ void testFloatToDouble() { @Test void testDateToTimestamp() { - Schema schemaWithFloat = + Schema schemaWithDate = new Schema(Types.NestedField.optional(1, "date2timestamp", Types.DateType.get())); - Schema schemaWithDouble = + Schema schemaWithTimestamp = 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(); + LocalDate date = LocalDate.of(2022, 1, 10); - assertThat(convert(GenericRowData.of(days), schemaWithFloat, schemaWithDouble)) - .isEqualTo(GenericRowData.of(TimestampData.fromEpochMillis(time.getMillis()))); + assertThat( + convert( + GenericRowData.of((int) date.toEpochDay()), schemaWithDate, schemaWithTimestamp)) + .isEqualTo(GenericRowData.of(TimestampData.fromLocalDateTime(date.atStartOfDay()))); } @Test @@ -139,6 +142,35 @@ void testIncreasePrecision() { .isEqualTo(GenericRowData.of(DecimalData.fromBigDecimal(new BigDecimal("-1.50"), 10, 2))); } + @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)); + + assertThat(CompareSchemasVisitor.isDataConversionPossible(dataType, tableType)).isTrue(); + assertThat(convert(GenericRowData.of(input), dataSchema, tableSchema)) + .isEqualTo(GenericRowData.of(expected)); + } + + private static Stream 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))); + } + @Test void testStructAddOptionalFields() { DataGenerator generator = new DataGenerators.StructOfPrimitive(); diff --git a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java index bdc825b44f2a..48c0ffda6454 100644 --- a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java +++ b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java @@ -21,8 +21,12 @@ import static org.assertj.core.api.Assertions.assertThat; import java.nio.file.Path; +import java.time.LocalDate; import java.util.Map; import org.apache.flink.api.java.tuple.Tuple2; +import org.apache.flink.table.data.GenericRowData; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.data.StringData; import org.apache.iceberg.PartitionSpec; import org.apache.iceberg.Schema; import org.apache.iceberg.SnapshotRef; @@ -174,6 +178,90 @@ void testInvalidateOldCacheEntryOnUpdate() { .isTrue(); } + @Test + void testAddColumnWithDataConverterNarrowing() { + Schema tableSchema = + new Schema( + Types.NestedField.optional(1, "id", Types.LongType.get()), + Types.NestedField.optional(2, "data", Types.StringType.get())); + 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())); + + TableMetadataCache.ResolvedSchemaInfo result = + updateWithAddedColumnAndConversion(tableSchema, rowSchema); + + RowData converted = + (RowData) + result + .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 testAddColumnWithDateToTimestampConversion() { + Schema tableSchema = + new Schema( + Types.NestedField.optional(1, "ts", Types.TimestampType.withoutZone()), + Types.NestedField.optional(2, "data", Types.StringType.get())); + 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())); + + TableMetadataCache.ResolvedSchemaInfo result = + updateWithAddedColumnAndConversion(tableSchema, rowSchema); + + LocalDate date = LocalDate.of(2026, 6, 30); + RowData converted = + (RowData) + result + .recordConverter() + .convert( + GenericRowData.of( + (int) date.toEpochDay(), + StringData.fromString("a"), + StringData.fromString("x"))); + assertThat(converted.getTimestamp(0, 6).toLocalDateTime()).isEqualTo(date.atStartOfDay()); + assertThat(converted.getString(1)).hasToString("a"); + assertThat(converted.getString(2)).hasToString("x"); + } + + private TableMetadataCache.ResolvedSchemaInfo updateWithAddedColumnAndConversion( + Schema tableSchema, Schema rowSchema) { + Catalog catalog = CATALOG_EXTENSION.catalog(); + TableIdentifier tableIdentifier = TableIdentifier.parse("default.myTable"); + 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); + + TableMetadataCache.ResolvedSchemaInfo result = + tableUpdater.update( + tableIdentifier, + SnapshotRef.MAIN_BRANCH, + rowSchema, + PartitionSpec.unpartitioned(), + TableCreator.DEFAULT) + .f0; + + Schema evolved = catalog.loadTable(tableIdentifier).schema(); + Types.NestedField convertedColumn = tableSchema.columns().get(0); + assertThat(evolved.findField(convertedColumn.name()).type()).isEqualTo(convertedColumn.type()); + assertThat(evolved.findField("extra")).isNotNull(); + assertThat(result.compareResult()) + .isEqualTo(CompareSchemasVisitor.Result.DATA_CONVERSION_NEEDED); + return result; + } + @Test void testLastResultInvalidation() { Catalog catalog = CATALOG_EXTENSION.catalog(); diff --git a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/CompareSchemasVisitor.java b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/CompareSchemasVisitor.java index cb4fc62c01b8..6c1b2e5673e9 100644 --- a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/CompareSchemasVisitor.java +++ b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/CompareSchemasVisitor.java @@ -166,41 +166,54 @@ public Result map( } @Override - @SuppressWarnings("checkstyle:CyclomaticComplexity") public Result primitive(Type.PrimitiveType primitive, Integer tableSchemaId) { if (tableSchemaId == null) { return Result.SCHEMA_UPDATE_NEEDED; } Type tableSchemaType = tableSchema.findField(tableSchemaId).type(); - if (!tableSchemaType.isPrimitiveType()) { - return Result.SCHEMA_UPDATE_NEEDED; - } - - Type.PrimitiveType tableSchemaPrimitiveType = tableSchemaType.asPrimitiveType(); - if (primitive.equals(tableSchemaPrimitiveType)) { + if (primitive.equals(tableSchemaType)) { 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, tableSchemaType)) { 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 dataType, Type tableType) { + if (!dataType.isPrimitiveType() || !tableType.isPrimitiveType()) { + return false; + } + + 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; + } + + if (dataType.typeId() == Type.TypeID.DECIMAL && tableType.typeId() == Type.TypeID.DECIMAL) { + Types.DecimalType dataDecimalType = (Types.DecimalType) dataType; + Types.DecimalType tableDecimalType = (Types.DecimalType) tableType; + return dataDecimalType.scale() == tableDecimalType.scale() + && dataDecimalType.precision() < tableDecimalType.precision(); + } + + return false; + } + static class PartnerIdByNameAccessors implements PartnerAccessors { private final Schema tableSchema; private boolean caseSensitive; diff --git a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/EvolveSchemaVisitor.java b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/EvolveSchemaVisitor.java index d9747d201e3d..a5d6c8776a44 100644 --- a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/EvolveSchemaVisitor.java +++ b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/EvolveSchemaVisitor.java @@ -211,8 +211,12 @@ private void updateColumn(Types.NestedField existingField, Types.NestedField tar String existingColumnName = this.existingSchema.findColumnName(existingField.fieldId()); boolean needsOptionalUpdate = targetField.isOptional() && existingField.isRequired(); + boolean handledByDataConversion = + CompareSchemasVisitor.isDataConversionPossible(targetField.type(), existingField.type()); boolean needsTypeUpdate = - targetField.type().isPrimitiveType() && !targetField.type().equals(existingField.type()); + targetField.type().isPrimitiveType() + && !targetField.type().equals(existingField.type()) + && !handledByDataConversion; boolean needsDocUpdate = targetField.doc() != null && !targetField.doc().equals(existingField.doc()); diff --git a/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestEvolveSchemaVisitor.java b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestEvolveSchemaVisitor.java index d2da73c66973..55e52751febb 100644 --- a/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestEvolveSchemaVisitor.java +++ b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestEvolveSchemaVisitor.java @@ -479,20 +479,48 @@ public void testTypePromoteFloatToDouble() { } @Test - public void testInvalidTypePromoteDoubleToFloat() { + public void testTypeNarrowDoubleToFloatIsLeftToConversion() { Schema currentSchema = new Schema(required(1, "aCol", DoubleType.get())); Schema targetSchema = new Schema(required(1, "aCol", FloatType.get())); - assertThatThrownBy( - () -> - EvolveSchemaVisitor.visit( - TABLE, - loadUpdateApi(currentSchema), - currentSchema, - targetSchema, - true, - PRESERVE_COLUMNS)) - .hasMessage("Cannot change column type: aCol: double -> float") - .isInstanceOf(IllegalArgumentException.class); + + UpdateSchema updateApi = loadUpdateApi(currentSchema); + EvolveSchemaVisitor.visit( + TABLE, updateApi, currentSchema, targetSchema, CASE_SENSITIVE, PRESERVE_COLUMNS); + Schema applied = updateApi.apply(); + + assertThat(applied.findField("aCol").type()).isEqualTo(DoubleType.get()); + } + + @Test + public void testAddColumnWithResidualNarrowingDoesNotFail() { + Schema currentSchema = new Schema(required(1, "id", LongType.get())); + Schema targetSchema = + new Schema(required(1, "id", IntegerType.get()), optional(2, "age", IntegerType.get())); + + UpdateSchema updateApi = loadUpdateApi(currentSchema); + EvolveSchemaVisitor.visit( + TABLE, updateApi, currentSchema, targetSchema, CASE_SENSITIVE, PRESERVE_COLUMNS); + Schema applied = updateApi.apply(); + + assertThat(applied.findField("id").type()).isEqualTo(LongType.get()); + assertThat(applied.findField("age")).isNotNull(); + assertThat(applied.findField("age").type()).isEqualTo(IntegerType.get()); + } + + @Test + public void testAddColumnWithResidualDateToTimestampDoesNotFail() { + Schema currentSchema = new Schema(required(1, "ts", Types.TimestampType.withoutZone())); + Schema targetSchema = + new Schema(required(1, "ts", Types.DateType.get()), optional(2, "age", IntegerType.get())); + + UpdateSchema updateApi = loadUpdateApi(currentSchema); + EvolveSchemaVisitor.visit( + TABLE, updateApi, currentSchema, targetSchema, CASE_SENSITIVE, PRESERVE_COLUMNS); + Schema applied = updateApi.apply(); + + assertThat(applied.findField("ts").type()).isEqualTo(Types.TimestampType.withoutZone()); + assertThat(applied.findField("age")).isNotNull(); + assertThat(applied.findField("age").type()).isEqualTo(IntegerType.get()); } @Test diff --git a/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestRowDataConverter.java b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestRowDataConverter.java index c4a86bb79e4a..7c179f793af7 100644 --- a/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestRowDataConverter.java +++ b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestRowDataConverter.java @@ -24,6 +24,8 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy; import java.math.BigDecimal; +import java.time.LocalDate; +import java.util.stream.Stream; import org.apache.flink.table.data.DecimalData; import org.apache.flink.table.data.GenericArrayData; import org.apache.flink.table.data.GenericMapData; @@ -37,11 +39,12 @@ import org.apache.iceberg.flink.FlinkSchemaUtil; import org.apache.iceberg.flink.SimpleDataUtil; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; +import org.apache.iceberg.types.Type; import org.apache.iceberg.types.Types; -import org.joda.time.DateTime; -import org.joda.time.DateTimeZone; -import org.joda.time.Days; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; class TestRowDataConverter { @@ -110,18 +113,18 @@ void testFloatToDouble() { @Test void testDateToTimestamp() { - Schema schemaWithFloat = + Schema schemaWithDate = new Schema(Types.NestedField.optional(1, "date2timestamp", Types.DateType.get())); - Schema schemaWithDouble = + Schema schemaWithTimestamp = 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(); + LocalDate date = LocalDate.of(2022, 1, 10); - assertThat(convert(GenericRowData.of(days), schemaWithFloat, schemaWithDouble)) - .isEqualTo(GenericRowData.of(TimestampData.fromEpochMillis(time.getMillis()))); + assertThat( + convert( + GenericRowData.of((int) date.toEpochDay()), schemaWithDate, schemaWithTimestamp)) + .isEqualTo(GenericRowData.of(TimestampData.fromLocalDateTime(date.atStartOfDay()))); } @Test @@ -139,6 +142,35 @@ void testIncreasePrecision() { .isEqualTo(GenericRowData.of(DecimalData.fromBigDecimal(new BigDecimal("-1.50"), 10, 2))); } + @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)); + + assertThat(CompareSchemasVisitor.isDataConversionPossible(dataType, tableType)).isTrue(); + assertThat(convert(GenericRowData.of(input), dataSchema, tableSchema)) + .isEqualTo(GenericRowData.of(expected)); + } + + private static Stream 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))); + } + @Test void testStructAddOptionalFields() { DataGenerator generator = new DataGenerators.StructOfPrimitive(); diff --git a/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java index bdc825b44f2a..48c0ffda6454 100644 --- a/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java +++ b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java @@ -21,8 +21,12 @@ import static org.assertj.core.api.Assertions.assertThat; import java.nio.file.Path; +import java.time.LocalDate; import java.util.Map; import org.apache.flink.api.java.tuple.Tuple2; +import org.apache.flink.table.data.GenericRowData; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.data.StringData; import org.apache.iceberg.PartitionSpec; import org.apache.iceberg.Schema; import org.apache.iceberg.SnapshotRef; @@ -174,6 +178,90 @@ void testInvalidateOldCacheEntryOnUpdate() { .isTrue(); } + @Test + void testAddColumnWithDataConverterNarrowing() { + Schema tableSchema = + new Schema( + Types.NestedField.optional(1, "id", Types.LongType.get()), + Types.NestedField.optional(2, "data", Types.StringType.get())); + 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())); + + TableMetadataCache.ResolvedSchemaInfo result = + updateWithAddedColumnAndConversion(tableSchema, rowSchema); + + RowData converted = + (RowData) + result + .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 testAddColumnWithDateToTimestampConversion() { + Schema tableSchema = + new Schema( + Types.NestedField.optional(1, "ts", Types.TimestampType.withoutZone()), + Types.NestedField.optional(2, "data", Types.StringType.get())); + 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())); + + TableMetadataCache.ResolvedSchemaInfo result = + updateWithAddedColumnAndConversion(tableSchema, rowSchema); + + LocalDate date = LocalDate.of(2026, 6, 30); + RowData converted = + (RowData) + result + .recordConverter() + .convert( + GenericRowData.of( + (int) date.toEpochDay(), + StringData.fromString("a"), + StringData.fromString("x"))); + assertThat(converted.getTimestamp(0, 6).toLocalDateTime()).isEqualTo(date.atStartOfDay()); + assertThat(converted.getString(1)).hasToString("a"); + assertThat(converted.getString(2)).hasToString("x"); + } + + private TableMetadataCache.ResolvedSchemaInfo updateWithAddedColumnAndConversion( + Schema tableSchema, Schema rowSchema) { + Catalog catalog = CATALOG_EXTENSION.catalog(); + TableIdentifier tableIdentifier = TableIdentifier.parse("default.myTable"); + 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); + + TableMetadataCache.ResolvedSchemaInfo result = + tableUpdater.update( + tableIdentifier, + SnapshotRef.MAIN_BRANCH, + rowSchema, + PartitionSpec.unpartitioned(), + TableCreator.DEFAULT) + .f0; + + Schema evolved = catalog.loadTable(tableIdentifier).schema(); + Types.NestedField convertedColumn = tableSchema.columns().get(0); + assertThat(evolved.findField(convertedColumn.name()).type()).isEqualTo(convertedColumn.type()); + assertThat(evolved.findField("extra")).isNotNull(); + assertThat(result.compareResult()) + .isEqualTo(CompareSchemasVisitor.Result.DATA_CONVERSION_NEEDED); + return result; + } + @Test void testLastResultInvalidation() { Catalog catalog = CATALOG_EXTENSION.catalog();