This is an automated email from the ASF dual-hosted git repository.
mxm pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new 48330b8dac Flink: Backport case-insensitive DataConverter fix to 1.20,
2.1, and 2.2 (#18295)
48330b8dac is described below
commit 48330b8dacab6662242d252b39c8444190979bb2
Author: Mithun M <[email protected]>
AuthorDate: Tue Sep 29 12:47:43 2026 +0530
Flink: Backport case-insensitive DataConverter fix to 1.20, 2.1, and 2.2
(#18295)
---
.../iceberg/flink/sink/dynamic/DataConverter.java | 58 ++++++++++++++++-----
.../flink/sink/dynamic/TableMetadataCache.java | 4 +-
.../iceberg/flink/sink/dynamic/TableUpdater.java | 4 +-
.../flink/sink/dynamic/TestDynamicIcebergSink.java | 60 ++++++++++++++++++++++
.../flink/sink/dynamic/TestRowDataConverter.java | 9 +++-
.../iceberg/flink/sink/dynamic/DataConverter.java | 58 ++++++++++++++++-----
.../flink/sink/dynamic/TableMetadataCache.java | 4 +-
.../iceberg/flink/sink/dynamic/TableUpdater.java | 4 +-
.../flink/sink/dynamic/TestDynamicIcebergSink.java | 60 ++++++++++++++++++++++
.../flink/sink/dynamic/TestRowDataConverter.java | 9 +++-
.../iceberg/flink/sink/dynamic/DataConverter.java | 58 ++++++++++++++++-----
.../flink/sink/dynamic/TableMetadataCache.java | 4 +-
.../iceberg/flink/sink/dynamic/TableUpdater.java | 4 +-
.../flink/sink/dynamic/TestDynamicIcebergSink.java | 60 ++++++++++++++++++++++
.../flink/sink/dynamic/TestRowDataConverter.java | 9 +++-
15 files changed, 354 insertions(+), 51 deletions(-)
diff --git
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DataConverter.java
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DataConverter.java
index 0234c01d64..b3fe35476d 100644
---
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DataConverter.java
+++
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DataConverter.java
@@ -64,11 +64,12 @@ interface DataConverter {
return object -> object;
}
- static DataConverter getNullable(LogicalType sourceType, LogicalType
targetType) {
- return nullable(get(sourceType, targetType));
+ static DataConverter getNullable(
+ LogicalType sourceType, LogicalType targetType, boolean caseSensitive) {
+ return nullable(get(sourceType, targetType, caseSensitive));
}
- static DataConverter get(LogicalType sourceType, LogicalType targetType) {
+ static DataConverter get(LogicalType sourceType, LogicalType targetType,
boolean caseSensitive) {
switch (targetType.getTypeRoot()) {
case BOOLEAN:
case INTEGER:
@@ -118,11 +119,11 @@ interface DataConverter {
}
};
case ROW:
- return new RowDataConverter((RowType) sourceType, (RowType)
targetType);
+ return new RowDataConverter((RowType) sourceType, (RowType)
targetType, caseSensitive);
case ARRAY:
- return new ArrayConverter((ArrayType) sourceType, (ArrayType)
targetType);
+ return new ArrayConverter((ArrayType) sourceType, (ArrayType)
targetType, caseSensitive);
case MAP:
- return new MapConverter((MapType) sourceType, (MapType) targetType);
+ return new MapConverter((MapType) sourceType, (MapType) targetType,
caseSensitive);
default:
throw new UnsupportedOperationException("Not a supported type: " +
targetType);
}
@@ -136,7 +137,7 @@ interface DataConverter {
private final RowData.FieldGetter[] fieldGetters;
private final DataConverter[] dataConverters;
- RowDataConverter(RowType sourceType, RowType targetType) {
+ RowDataConverter(RowType sourceType, RowType targetType, boolean
caseSensitive) {
this.fieldGetters = new
RowData.FieldGetter[targetType.getFields().size()];
this.dataConverters = new DataConverter[targetType.getFields().size()];
@@ -144,7 +145,7 @@ interface DataConverter {
RowData.FieldGetter fieldGetter;
DataConverter dataConverter;
RowType.RowField targetField = targetType.getFields().get(i);
- int sourceFieldIndex = sourceType.getFieldIndex(targetField.getName());
+ int sourceFieldIndex = findFieldIndex(sourceType,
targetField.getName(), caseSensitive);
if (sourceFieldIndex == -1) {
if (targetField.getType().isNullable()) {
fieldGetter = row -> null;
@@ -158,7 +159,9 @@ interface DataConverter {
} else {
RowType.RowField sourceField =
sourceType.getFields().get(sourceFieldIndex);
fieldGetter = RowData.createFieldGetter(sourceField.getType(),
sourceFieldIndex);
- dataConverter = DataConverter.getNullable(sourceField.getType(),
targetField.getType());
+ dataConverter =
+ DataConverter.getNullable(
+ sourceField.getType(), targetField.getType(), caseSensitive);
}
this.fieldGetters[i] = fieldGetter;
@@ -183,10 +186,11 @@ interface DataConverter {
private final ArrayData.ElementGetter elementGetter;
private final DataConverter elementConverter;
- ArrayConverter(ArrayType sourceType, ArrayType targetType) {
+ ArrayConverter(ArrayType sourceType, ArrayType targetType, boolean
caseSensitive) {
this.elementGetter =
ArrayData.createElementGetter(sourceType.getElementType());
this.elementConverter =
- DataConverter.getNullable(sourceType.getElementType(),
targetType.getElementType());
+ DataConverter.getNullable(
+ sourceType.getElementType(), targetType.getElementType(),
caseSensitive);
}
@Override
@@ -208,13 +212,15 @@ interface DataConverter {
private final DataConverter keyConverter;
private final DataConverter valueConverter;
- MapConverter(MapType sourceType, MapType targetType) {
+ MapConverter(MapType sourceType, MapType targetType, boolean
caseSensitive) {
this.keyGetter = ArrayData.createElementGetter(sourceType.getKeyType());
this.valueGetter =
ArrayData.createElementGetter(sourceType.getValueType());
this.keyConverter =
- DataConverter.getNullable(sourceType.getKeyType(),
targetType.getKeyType());
+ DataConverter.getNullable(
+ sourceType.getKeyType(), targetType.getKeyType(), caseSensitive);
this.valueConverter =
- DataConverter.getNullable(sourceType.getValueType(),
targetType.getValueType());
+ DataConverter.getNullable(
+ sourceType.getValueType(), targetType.getValueType(),
caseSensitive);
}
@Override
@@ -232,4 +238,28 @@ interface DataConverter {
return new GenericMapData(convertedMap);
}
}
+
+ private static int findFieldIndex(RowType sourceType, String targetName,
boolean caseSensitive) {
+ if (caseSensitive) {
+ return sourceType.getFieldIndex(targetName);
+ }
+
+ int matchingIndex = -1;
+ for (int i = 0; i < sourceType.getFieldCount(); i++) {
+ if
(sourceType.getFields().get(i).getName().equalsIgnoreCase(targetName)) {
+ if (matchingIndex != -1) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Ambiguous case-insensitive source field match for '%s':
'%s' and '%s' in %s",
+ targetName,
+ sourceType.getFields().get(matchingIndex).getName(),
+ sourceType.getFields().get(i).getName(),
+ sourceType));
+ }
+ matchingIndex = i;
+ }
+ }
+
+ return matchingIndex;
+ }
}
diff --git
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableMetadataCache.java
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableMetadataCache.java
index fdefc01402..dd8f73c166 100644
---
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableMetadataCache.java
+++
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableMetadataCache.java
@@ -184,7 +184,9 @@ class TableMetadataCache {
compatible,
CompareSchemasVisitor.Result.DATA_CONVERSION_NEEDED,
DataConverter.get(
- FlinkSchemaUtil.convert(input),
FlinkSchemaUtil.convert(compatible)));
+ FlinkSchemaUtil.convert(input),
+ FlinkSchemaUtil.convert(compatible),
+ caseSensitive));
cached.inputSchemas.put(input, newResult);
return newResult;
} else if (cached != null && cached.tableExists) {
diff --git
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java
index b0bdad8ed1..26eb34db3c 100644
---
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java
+++
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java
@@ -142,7 +142,9 @@ class TableUpdater {
tableSchema,
result,
DataConverter.get(
- FlinkSchemaUtil.convert(schema),
FlinkSchemaUtil.convert(tableSchema)));
+ FlinkSchemaUtil.convert(schema),
+ FlinkSchemaUtil.convert(tableSchema),
+ caseSensitive));
case SCHEMA_UPDATE_NEEDED:
LOG.info(
"Triggering schema update for table {} {} to {}", identifier,
tableSchema, schema);
diff --git
a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java
index 9e05930b88..861da01032 100644
---
a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java
+++
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java
@@ -55,7 +55,9 @@ import
org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.graph.StreamNode;
import org.apache.flink.table.catalog.ResolvedSchema;
+import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
import org.apache.flink.table.data.util.DataFormatConverters;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.logical.RowType;
@@ -1890,4 +1892,62 @@ class TestDynamicIcebergSink extends
TestFlinkIcebergSinkBase {
RandomGenericData.generate(schema, 1, seedOverride), schema)
.get(0);
}
+
+ @Test
+ void testCaseInsensitiveDataConversionDoesNotDropValues() throws Exception {
+ Schema tableSchema =
+ 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()));
+
+ TableIdentifier identifier = TableIdentifier.of(DATABASE, "t1");
+ CATALOG_EXTENSION.catalog().createTable(identifier, tableSchema,
PartitionSpec.unpartitioned());
+
+ DynamicIcebergSink.forInput(env.fromData(1, 2, 3))
+ .generator(new CaseMismatchGenerator())
+ .catalogLoader(CATALOG_EXTENSION.catalogLoader())
+ .writeParallelism(1)
+ .immediateTableUpdate(true)
+ .caseSensitive(false)
+ .append();
+
+ env.execute("case-insensitive data conversion");
+
+ Table table = CATALOG_EXTENSION.catalog().loadTable(identifier);
+ List<Record> records =
Lists.newArrayList(IcebergGenerics.read(table).build());
+
+ assertThat(records).hasSize(3);
+ assertThat(records)
+ .allSatisfy(
+ record -> {
+ Integer id = (Integer) record.getField("id");
+ assertThat(id).isIn(1, 2, 3);
+ assertThat(record.getField("data")).isEqualTo("value-" + id);
+ assertThat(record.getField("extra")).isNull();
+ });
+ assertThat(records).extracting(r ->
r.getField("id")).containsExactlyInAnyOrder(1, 2, 3);
+ }
+
+ private static class CaseMismatchGenerator implements
DynamicRecordGenerator<Integer> {
+ @Override
+ public void generate(Integer value, Collector<DynamicRecord> out) {
+ Schema inputSchema =
+ new Schema(
+ Types.NestedField.optional(1, "Id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "Data", Types.StringType.get()));
+ GenericRowData row = new GenericRowData(2);
+ row.setField(0, value);
+ row.setField(1, StringData.fromString("value-" + value));
+ out.collect(
+ new DynamicRecord(
+ TableIdentifier.of(DATABASE, "t1"),
+ "main",
+ inputSchema,
+ row,
+ PartitionSpec.unpartitioned(),
+ DistributionMode.NONE,
+ 1));
+ }
+ }
}
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 cd3971ea73..19e08652fb 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
@@ -295,9 +295,16 @@ class TestRowDataConverter {
}
private static RowData convert(RowData sourceData, Schema sourceSchema,
Schema targetSchema) {
+ return convert(sourceData, sourceSchema, targetSchema, true);
+ }
+
+ private static RowData convert(
+ RowData sourceData, Schema sourceSchema, Schema targetSchema, boolean
caseSensitive) {
return (RowData)
DataConverter.get(
- FlinkSchemaUtil.convert(sourceSchema),
FlinkSchemaUtil.convert(targetSchema))
+ FlinkSchemaUtil.convert(sourceSchema),
+ FlinkSchemaUtil.convert(targetSchema),
+ caseSensitive)
.convert(sourceData);
}
}
diff --git
a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DataConverter.java
b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DataConverter.java
index 0234c01d64..b3fe35476d 100644
---
a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DataConverter.java
+++
b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DataConverter.java
@@ -64,11 +64,12 @@ interface DataConverter {
return object -> object;
}
- static DataConverter getNullable(LogicalType sourceType, LogicalType
targetType) {
- return nullable(get(sourceType, targetType));
+ static DataConverter getNullable(
+ LogicalType sourceType, LogicalType targetType, boolean caseSensitive) {
+ return nullable(get(sourceType, targetType, caseSensitive));
}
- static DataConverter get(LogicalType sourceType, LogicalType targetType) {
+ static DataConverter get(LogicalType sourceType, LogicalType targetType,
boolean caseSensitive) {
switch (targetType.getTypeRoot()) {
case BOOLEAN:
case INTEGER:
@@ -118,11 +119,11 @@ interface DataConverter {
}
};
case ROW:
- return new RowDataConverter((RowType) sourceType, (RowType)
targetType);
+ return new RowDataConverter((RowType) sourceType, (RowType)
targetType, caseSensitive);
case ARRAY:
- return new ArrayConverter((ArrayType) sourceType, (ArrayType)
targetType);
+ return new ArrayConverter((ArrayType) sourceType, (ArrayType)
targetType, caseSensitive);
case MAP:
- return new MapConverter((MapType) sourceType, (MapType) targetType);
+ return new MapConverter((MapType) sourceType, (MapType) targetType,
caseSensitive);
default:
throw new UnsupportedOperationException("Not a supported type: " +
targetType);
}
@@ -136,7 +137,7 @@ interface DataConverter {
private final RowData.FieldGetter[] fieldGetters;
private final DataConverter[] dataConverters;
- RowDataConverter(RowType sourceType, RowType targetType) {
+ RowDataConverter(RowType sourceType, RowType targetType, boolean
caseSensitive) {
this.fieldGetters = new
RowData.FieldGetter[targetType.getFields().size()];
this.dataConverters = new DataConverter[targetType.getFields().size()];
@@ -144,7 +145,7 @@ interface DataConverter {
RowData.FieldGetter fieldGetter;
DataConverter dataConverter;
RowType.RowField targetField = targetType.getFields().get(i);
- int sourceFieldIndex = sourceType.getFieldIndex(targetField.getName());
+ int sourceFieldIndex = findFieldIndex(sourceType,
targetField.getName(), caseSensitive);
if (sourceFieldIndex == -1) {
if (targetField.getType().isNullable()) {
fieldGetter = row -> null;
@@ -158,7 +159,9 @@ interface DataConverter {
} else {
RowType.RowField sourceField =
sourceType.getFields().get(sourceFieldIndex);
fieldGetter = RowData.createFieldGetter(sourceField.getType(),
sourceFieldIndex);
- dataConverter = DataConverter.getNullable(sourceField.getType(),
targetField.getType());
+ dataConverter =
+ DataConverter.getNullable(
+ sourceField.getType(), targetField.getType(), caseSensitive);
}
this.fieldGetters[i] = fieldGetter;
@@ -183,10 +186,11 @@ interface DataConverter {
private final ArrayData.ElementGetter elementGetter;
private final DataConverter elementConverter;
- ArrayConverter(ArrayType sourceType, ArrayType targetType) {
+ ArrayConverter(ArrayType sourceType, ArrayType targetType, boolean
caseSensitive) {
this.elementGetter =
ArrayData.createElementGetter(sourceType.getElementType());
this.elementConverter =
- DataConverter.getNullable(sourceType.getElementType(),
targetType.getElementType());
+ DataConverter.getNullable(
+ sourceType.getElementType(), targetType.getElementType(),
caseSensitive);
}
@Override
@@ -208,13 +212,15 @@ interface DataConverter {
private final DataConverter keyConverter;
private final DataConverter valueConverter;
- MapConverter(MapType sourceType, MapType targetType) {
+ MapConverter(MapType sourceType, MapType targetType, boolean
caseSensitive) {
this.keyGetter = ArrayData.createElementGetter(sourceType.getKeyType());
this.valueGetter =
ArrayData.createElementGetter(sourceType.getValueType());
this.keyConverter =
- DataConverter.getNullable(sourceType.getKeyType(),
targetType.getKeyType());
+ DataConverter.getNullable(
+ sourceType.getKeyType(), targetType.getKeyType(), caseSensitive);
this.valueConverter =
- DataConverter.getNullable(sourceType.getValueType(),
targetType.getValueType());
+ DataConverter.getNullable(
+ sourceType.getValueType(), targetType.getValueType(),
caseSensitive);
}
@Override
@@ -232,4 +238,28 @@ interface DataConverter {
return new GenericMapData(convertedMap);
}
}
+
+ private static int findFieldIndex(RowType sourceType, String targetName,
boolean caseSensitive) {
+ if (caseSensitive) {
+ return sourceType.getFieldIndex(targetName);
+ }
+
+ int matchingIndex = -1;
+ for (int i = 0; i < sourceType.getFieldCount(); i++) {
+ if
(sourceType.getFields().get(i).getName().equalsIgnoreCase(targetName)) {
+ if (matchingIndex != -1) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Ambiguous case-insensitive source field match for '%s':
'%s' and '%s' in %s",
+ targetName,
+ sourceType.getFields().get(matchingIndex).getName(),
+ sourceType.getFields().get(i).getName(),
+ sourceType));
+ }
+ matchingIndex = i;
+ }
+ }
+
+ return matchingIndex;
+ }
}
diff --git
a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableMetadataCache.java
b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableMetadataCache.java
index fdefc01402..dd8f73c166 100644
---
a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableMetadataCache.java
+++
b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableMetadataCache.java
@@ -184,7 +184,9 @@ class TableMetadataCache {
compatible,
CompareSchemasVisitor.Result.DATA_CONVERSION_NEEDED,
DataConverter.get(
- FlinkSchemaUtil.convert(input),
FlinkSchemaUtil.convert(compatible)));
+ FlinkSchemaUtil.convert(input),
+ FlinkSchemaUtil.convert(compatible),
+ caseSensitive));
cached.inputSchemas.put(input, newResult);
return newResult;
} else if (cached != null && cached.tableExists) {
diff --git
a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java
b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java
index b0bdad8ed1..26eb34db3c 100644
---
a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java
+++
b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java
@@ -142,7 +142,9 @@ class TableUpdater {
tableSchema,
result,
DataConverter.get(
- FlinkSchemaUtil.convert(schema),
FlinkSchemaUtil.convert(tableSchema)));
+ FlinkSchemaUtil.convert(schema),
+ FlinkSchemaUtil.convert(tableSchema),
+ caseSensitive));
case SCHEMA_UPDATE_NEEDED:
LOG.info(
"Triggering schema update for table {} {} to {}", identifier,
tableSchema, schema);
diff --git
a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java
b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java
index 7c4669ff98..a426529d32 100644
---
a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java
+++
b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java
@@ -56,7 +56,9 @@ import
org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.graph.StreamNode;
import org.apache.flink.table.catalog.ResolvedSchema;
+import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
import org.apache.flink.table.data.util.DataFormatConverters;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.logical.RowType;
@@ -1902,4 +1904,62 @@ class TestDynamicIcebergSink extends
TestFlinkIcebergSinkBase {
RandomGenericData.generate(schema, 1, seedOverride), schema)
.get(0);
}
+
+ @Test
+ void testCaseInsensitiveDataConversionDoesNotDropValues() throws Exception {
+ Schema tableSchema =
+ 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()));
+
+ TableIdentifier identifier = TableIdentifier.of(DATABASE, "t1");
+ CATALOG_EXTENSION.catalog().createTable(identifier, tableSchema,
PartitionSpec.unpartitioned());
+
+ DynamicIcebergSink.forInput(env.fromData(1, 2, 3))
+ .generator(new CaseMismatchGenerator())
+ .catalogLoader(CATALOG_EXTENSION.catalogLoader())
+ .writeParallelism(1)
+ .immediateTableUpdate(true)
+ .caseSensitive(false)
+ .append();
+
+ env.execute("case-insensitive data conversion");
+
+ Table table = CATALOG_EXTENSION.catalog().loadTable(identifier);
+ List<Record> records =
Lists.newArrayList(IcebergGenerics.read(table).build());
+
+ assertThat(records).hasSize(3);
+ assertThat(records)
+ .allSatisfy(
+ record -> {
+ Integer id = (Integer) record.getField("id");
+ assertThat(id).isIn(1, 2, 3);
+ assertThat(record.getField("data")).isEqualTo("value-" + id);
+ assertThat(record.getField("extra")).isNull();
+ });
+ assertThat(records).extracting(r ->
r.getField("id")).containsExactlyInAnyOrder(1, 2, 3);
+ }
+
+ private static class CaseMismatchGenerator implements
DynamicRecordGenerator<Integer> {
+ @Override
+ public void generate(Integer value, Collector<DynamicRecord> out) {
+ Schema inputSchema =
+ new Schema(
+ Types.NestedField.optional(1, "Id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "Data", Types.StringType.get()));
+ GenericRowData row = new GenericRowData(2);
+ row.setField(0, value);
+ row.setField(1, StringData.fromString("value-" + value));
+ out.collect(
+ new DynamicRecord(
+ TableIdentifier.of(DATABASE, "t1"),
+ "main",
+ inputSchema,
+ row,
+ PartitionSpec.unpartitioned(),
+ DistributionMode.NONE,
+ 1));
+ }
+ }
}
diff --git
a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestRowDataConverter.java
b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestRowDataConverter.java
index cd3971ea73..19e08652fb 100644
---
a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestRowDataConverter.java
+++
b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestRowDataConverter.java
@@ -295,9 +295,16 @@ class TestRowDataConverter {
}
private static RowData convert(RowData sourceData, Schema sourceSchema,
Schema targetSchema) {
+ return convert(sourceData, sourceSchema, targetSchema, true);
+ }
+
+ private static RowData convert(
+ RowData sourceData, Schema sourceSchema, Schema targetSchema, boolean
caseSensitive) {
return (RowData)
DataConverter.get(
- FlinkSchemaUtil.convert(sourceSchema),
FlinkSchemaUtil.convert(targetSchema))
+ FlinkSchemaUtil.convert(sourceSchema),
+ FlinkSchemaUtil.convert(targetSchema),
+ caseSensitive)
.convert(sourceData);
}
}
diff --git
a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DataConverter.java
b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DataConverter.java
index 0234c01d64..b3fe35476d 100644
---
a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DataConverter.java
+++
b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DataConverter.java
@@ -64,11 +64,12 @@ interface DataConverter {
return object -> object;
}
- static DataConverter getNullable(LogicalType sourceType, LogicalType
targetType) {
- return nullable(get(sourceType, targetType));
+ static DataConverter getNullable(
+ LogicalType sourceType, LogicalType targetType, boolean caseSensitive) {
+ return nullable(get(sourceType, targetType, caseSensitive));
}
- static DataConverter get(LogicalType sourceType, LogicalType targetType) {
+ static DataConverter get(LogicalType sourceType, LogicalType targetType,
boolean caseSensitive) {
switch (targetType.getTypeRoot()) {
case BOOLEAN:
case INTEGER:
@@ -118,11 +119,11 @@ interface DataConverter {
}
};
case ROW:
- return new RowDataConverter((RowType) sourceType, (RowType)
targetType);
+ return new RowDataConverter((RowType) sourceType, (RowType)
targetType, caseSensitive);
case ARRAY:
- return new ArrayConverter((ArrayType) sourceType, (ArrayType)
targetType);
+ return new ArrayConverter((ArrayType) sourceType, (ArrayType)
targetType, caseSensitive);
case MAP:
- return new MapConverter((MapType) sourceType, (MapType) targetType);
+ return new MapConverter((MapType) sourceType, (MapType) targetType,
caseSensitive);
default:
throw new UnsupportedOperationException("Not a supported type: " +
targetType);
}
@@ -136,7 +137,7 @@ interface DataConverter {
private final RowData.FieldGetter[] fieldGetters;
private final DataConverter[] dataConverters;
- RowDataConverter(RowType sourceType, RowType targetType) {
+ RowDataConverter(RowType sourceType, RowType targetType, boolean
caseSensitive) {
this.fieldGetters = new
RowData.FieldGetter[targetType.getFields().size()];
this.dataConverters = new DataConverter[targetType.getFields().size()];
@@ -144,7 +145,7 @@ interface DataConverter {
RowData.FieldGetter fieldGetter;
DataConverter dataConverter;
RowType.RowField targetField = targetType.getFields().get(i);
- int sourceFieldIndex = sourceType.getFieldIndex(targetField.getName());
+ int sourceFieldIndex = findFieldIndex(sourceType,
targetField.getName(), caseSensitive);
if (sourceFieldIndex == -1) {
if (targetField.getType().isNullable()) {
fieldGetter = row -> null;
@@ -158,7 +159,9 @@ interface DataConverter {
} else {
RowType.RowField sourceField =
sourceType.getFields().get(sourceFieldIndex);
fieldGetter = RowData.createFieldGetter(sourceField.getType(),
sourceFieldIndex);
- dataConverter = DataConverter.getNullable(sourceField.getType(),
targetField.getType());
+ dataConverter =
+ DataConverter.getNullable(
+ sourceField.getType(), targetField.getType(), caseSensitive);
}
this.fieldGetters[i] = fieldGetter;
@@ -183,10 +186,11 @@ interface DataConverter {
private final ArrayData.ElementGetter elementGetter;
private final DataConverter elementConverter;
- ArrayConverter(ArrayType sourceType, ArrayType targetType) {
+ ArrayConverter(ArrayType sourceType, ArrayType targetType, boolean
caseSensitive) {
this.elementGetter =
ArrayData.createElementGetter(sourceType.getElementType());
this.elementConverter =
- DataConverter.getNullable(sourceType.getElementType(),
targetType.getElementType());
+ DataConverter.getNullable(
+ sourceType.getElementType(), targetType.getElementType(),
caseSensitive);
}
@Override
@@ -208,13 +212,15 @@ interface DataConverter {
private final DataConverter keyConverter;
private final DataConverter valueConverter;
- MapConverter(MapType sourceType, MapType targetType) {
+ MapConverter(MapType sourceType, MapType targetType, boolean
caseSensitive) {
this.keyGetter = ArrayData.createElementGetter(sourceType.getKeyType());
this.valueGetter =
ArrayData.createElementGetter(sourceType.getValueType());
this.keyConverter =
- DataConverter.getNullable(sourceType.getKeyType(),
targetType.getKeyType());
+ DataConverter.getNullable(
+ sourceType.getKeyType(), targetType.getKeyType(), caseSensitive);
this.valueConverter =
- DataConverter.getNullable(sourceType.getValueType(),
targetType.getValueType());
+ DataConverter.getNullable(
+ sourceType.getValueType(), targetType.getValueType(),
caseSensitive);
}
@Override
@@ -232,4 +238,28 @@ interface DataConverter {
return new GenericMapData(convertedMap);
}
}
+
+ private static int findFieldIndex(RowType sourceType, String targetName,
boolean caseSensitive) {
+ if (caseSensitive) {
+ return sourceType.getFieldIndex(targetName);
+ }
+
+ int matchingIndex = -1;
+ for (int i = 0; i < sourceType.getFieldCount(); i++) {
+ if
(sourceType.getFields().get(i).getName().equalsIgnoreCase(targetName)) {
+ if (matchingIndex != -1) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Ambiguous case-insensitive source field match for '%s':
'%s' and '%s' in %s",
+ targetName,
+ sourceType.getFields().get(matchingIndex).getName(),
+ sourceType.getFields().get(i).getName(),
+ sourceType));
+ }
+ matchingIndex = i;
+ }
+ }
+
+ return matchingIndex;
+ }
}
diff --git
a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableMetadataCache.java
b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableMetadataCache.java
index fdefc01402..dd8f73c166 100644
---
a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableMetadataCache.java
+++
b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableMetadataCache.java
@@ -184,7 +184,9 @@ class TableMetadataCache {
compatible,
CompareSchemasVisitor.Result.DATA_CONVERSION_NEEDED,
DataConverter.get(
- FlinkSchemaUtil.convert(input),
FlinkSchemaUtil.convert(compatible)));
+ FlinkSchemaUtil.convert(input),
+ FlinkSchemaUtil.convert(compatible),
+ caseSensitive));
cached.inputSchemas.put(input, newResult);
return newResult;
} else if (cached != null && cached.tableExists) {
diff --git
a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java
b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java
index b0bdad8ed1..26eb34db3c 100644
---
a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java
+++
b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java
@@ -142,7 +142,9 @@ class TableUpdater {
tableSchema,
result,
DataConverter.get(
- FlinkSchemaUtil.convert(schema),
FlinkSchemaUtil.convert(tableSchema)));
+ FlinkSchemaUtil.convert(schema),
+ FlinkSchemaUtil.convert(tableSchema),
+ caseSensitive));
case SCHEMA_UPDATE_NEEDED:
LOG.info(
"Triggering schema update for table {} {} to {}", identifier,
tableSchema, schema);
diff --git
a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java
b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java
index 7c4669ff98..a426529d32 100644
---
a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java
+++
b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java
@@ -56,7 +56,9 @@ import
org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.graph.StreamNode;
import org.apache.flink.table.catalog.ResolvedSchema;
+import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
import org.apache.flink.table.data.util.DataFormatConverters;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.logical.RowType;
@@ -1902,4 +1904,62 @@ class TestDynamicIcebergSink extends
TestFlinkIcebergSinkBase {
RandomGenericData.generate(schema, 1, seedOverride), schema)
.get(0);
}
+
+ @Test
+ void testCaseInsensitiveDataConversionDoesNotDropValues() throws Exception {
+ Schema tableSchema =
+ 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()));
+
+ TableIdentifier identifier = TableIdentifier.of(DATABASE, "t1");
+ CATALOG_EXTENSION.catalog().createTable(identifier, tableSchema,
PartitionSpec.unpartitioned());
+
+ DynamicIcebergSink.forInput(env.fromData(1, 2, 3))
+ .generator(new CaseMismatchGenerator())
+ .catalogLoader(CATALOG_EXTENSION.catalogLoader())
+ .writeParallelism(1)
+ .immediateTableUpdate(true)
+ .caseSensitive(false)
+ .append();
+
+ env.execute("case-insensitive data conversion");
+
+ Table table = CATALOG_EXTENSION.catalog().loadTable(identifier);
+ List<Record> records =
Lists.newArrayList(IcebergGenerics.read(table).build());
+
+ assertThat(records).hasSize(3);
+ assertThat(records)
+ .allSatisfy(
+ record -> {
+ Integer id = (Integer) record.getField("id");
+ assertThat(id).isIn(1, 2, 3);
+ assertThat(record.getField("data")).isEqualTo("value-" + id);
+ assertThat(record.getField("extra")).isNull();
+ });
+ assertThat(records).extracting(r ->
r.getField("id")).containsExactlyInAnyOrder(1, 2, 3);
+ }
+
+ private static class CaseMismatchGenerator implements
DynamicRecordGenerator<Integer> {
+ @Override
+ public void generate(Integer value, Collector<DynamicRecord> out) {
+ Schema inputSchema =
+ new Schema(
+ Types.NestedField.optional(1, "Id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "Data", Types.StringType.get()));
+ GenericRowData row = new GenericRowData(2);
+ row.setField(0, value);
+ row.setField(1, StringData.fromString("value-" + value));
+ out.collect(
+ new DynamicRecord(
+ TableIdentifier.of(DATABASE, "t1"),
+ "main",
+ inputSchema,
+ row,
+ PartitionSpec.unpartitioned(),
+ DistributionMode.NONE,
+ 1));
+ }
+ }
}
diff --git
a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestRowDataConverter.java
b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestRowDataConverter.java
index cd3971ea73..19e08652fb 100644
---
a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestRowDataConverter.java
+++
b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestRowDataConverter.java
@@ -295,9 +295,16 @@ class TestRowDataConverter {
}
private static RowData convert(RowData sourceData, Schema sourceSchema,
Schema targetSchema) {
+ return convert(sourceData, sourceSchema, targetSchema, true);
+ }
+
+ private static RowData convert(
+ RowData sourceData, Schema sourceSchema, Schema targetSchema, boolean
caseSensitive) {
return (RowData)
DataConverter.get(
- FlinkSchemaUtil.convert(sourceSchema),
FlinkSchemaUtil.convert(targetSchema))
+ FlinkSchemaUtil.convert(sourceSchema),
+ FlinkSchemaUtil.convert(targetSchema),
+ caseSensitive)
.convert(sourceData);
}
}
