mxm commented on code in PR #18066:
URL: https://github.com/apache/iceberg/pull/18066#discussion_r4025816060
##########
flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java:
##########
@@ -1902,4 +1904,59 @@ private static Row randomRow(Schema schema, long
seedOverride) {
RandomGenericData.generate(schema, 1, seedOverride), schema)
.get(0);
}
+
+ @Test
+ void testCaseInsensitiveDataConversionDropsValues() throws Exception {
Review Comment:
I think the name should be:
```suggestion
void testCaseInsensitiveDataConversionDoesNotDropValues() throws Exception
{
```
##########
flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java:
##########
@@ -1902,4 +1904,59 @@ private static Row randomRow(Schema schema, long
seedOverride) {
RandomGenericData.generate(schema, 1, seedOverride), schema)
.get(0);
}
+
+ @Test
+ void testCaseInsensitiveDataConversionDropsValues() 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 -> {
+ assertThat(record.getField("id")).isNotNull();
+ assertThat(record.getField("data")).isNotNull();
Review Comment:
Shall we check the values?
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DataConverter.java:
##########
@@ -232,4 +239,24 @@ public MapData convert(Object object) {
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)) {
+ Preconditions.checkArgument(
+ matchingIndex == -1,
+ "Ambiguous case-insensitive source field match for '%s' in %s",
+ targetName,
+ sourceType);
Review Comment:
We can also print the exact names of the duplicate field names based on the
indices. That should help with debugging.
--
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]
