Skip to content

Flink: Backport: handle simultaneous schema evolution and data conversion (#17024) #17191

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
huaxingao merged 1 commit into from
Jul 14, 2026
Merged

Flink: Backport: handle simultaneous schema evolution and data conversion (#17024) #17191

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
57 changes: 35 additions & 22 deletions ...1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/CompareSchemasVisitor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<Integer> {
private final Schema tableSchema;
private boolean caseSensitive;
Expand Down
6 changes: 5 additions & 1 deletion .../v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/EvolveSchemaVisitor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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());

Expand Down
52 changes: 40 additions & 12 deletions ...20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestEvolveSchemaVisitor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
52 changes: 42 additions & 10 deletions ...v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestRowDataConverter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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 {

Expand Down Expand Up @@ -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
Expand All @@ -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<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)));
}

@Test
void testStructAddOptionalFields() {
DataGenerator generator = new DataGenerators.StructOfPrimitive();
Expand Down
88 changes: 88 additions & 0 deletions flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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();
Expand Down
Loading
Loading