You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "mxm (via GitHub)" <gi...@apache.org> on 2025/12/01 13:55:22 UTC
[PR] Flink: Dynamic Sink: Add case-insensitive field matching [iceberg]
mxm opened a new pull request, #14729:
URL: https://github.com/apache/iceberg/pull/14729
Adds case-insensitive field name matching support to the Flink Dynamic Iceberg sink. This allows the sink to match schema fields regardless of letter casing, enabling compatibility with data sources that may use different casing conventions.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Add case-insensitive field matching [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #14729:
URL: https://github.com/apache/iceberg/pull/14729#discussion_r2580470831
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/EvolveSchemaVisitor.java:
##########
@@ -89,14 +91,23 @@ public Boolean struct(Types.StructType struct, Integer partnerId, List<Boolean>
Types.StructType partnerStruct = findFieldType(partnerId).asStructType();
String after = null;
for (Types.NestedField targetField : struct.fields()) {
- Types.NestedField nestedField = partnerStruct.field(targetField.name());
+ Types.NestedField nestedField =
+ caseSensitive
+ ? partnerStruct.field(targetField.name())
+ : partnerStruct.caseInsensitiveField(targetField.name());
Review Comment:
Yep, will consolidate.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Add case-insensitive field matching [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14729:
URL: https://github.com/apache/iceberg/pull/14729#discussion_r2580436675
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/EvolveSchemaVisitor.java:
##########
@@ -89,14 +91,23 @@ public Boolean struct(Types.StructType struct, Integer partnerId, List<Boolean>
Types.StructType partnerStruct = findFieldType(partnerId).asStructType();
String after = null;
for (Types.NestedField targetField : struct.fields()) {
- Types.NestedField nestedField = partnerStruct.field(targetField.name());
+ Types.NestedField nestedField =
+ caseSensitive
+ ? partnerStruct.field(targetField.name())
+ : partnerStruct.caseInsensitiveField(targetField.name());
final String columnName;
if (nestedField != null) {
updateColumn(nestedField, targetField);
columnName = this.existingSchema.findColumnName(nestedField.fieldId());
} else {
addColumn(partnerId, targetField);
- columnName = this.targetSchema.findColumnName(targetField.fieldId());
+ // We need to lookup the proper parent name because the letter case might be different on
+ // table vs target parent field (e.g. struct1.FIELD2 vs STRUCT1.STRUCT2). This is actually a
+ // bug in SchemaUpdate#moveAfter(columnName, after) because added fields are case-sensitive,
+ // even if case sensitivity is disabled on the API.
Review Comment:
Shouldn't we fix the bug instead of working around?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Add case-insensitive field matching [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14729:
URL: https://github.com/apache/iceberg/pull/14729#discussion_r2704055453
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableMetadataCache.java:
##########
@@ -52,13 +52,19 @@ class TableMetadataCache {
private final long refreshMs;
private final int inputSchemasPerTableCacheMaximumSize;
private final Map<TableIdentifier, CacheItem> tableCache;
+ private final boolean caseSensitive;
Review Comment:
Skip this for now and cache based on the actual input value
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Add case-insensitive field matching [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14729:
URL: https://github.com/apache/iceberg/pull/14729#discussion_r2580492329
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableMetadataCache.java:
##########
@@ -52,13 +52,19 @@ class TableMetadataCache {
private final long refreshMs;
private final int inputSchemasPerTableCacheMaximumSize;
private final Map<TableIdentifier, CacheItem> tableCache;
+ private final boolean caseSensitive;
Review Comment:
Would it make sense?
I expect that in this case the catalog is also case insensitive
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Add case-insensitive field matching [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #14729: Flink: Dynamic Sink: Add case-insensitive field matching
URL: https://github.com/apache/iceberg/pull/14729
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Add case-insensitive field matching [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #14729:
URL: https://github.com/apache/iceberg/pull/14729#discussion_r2580469570
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableMetadataCache.java:
##########
@@ -52,13 +52,19 @@ class TableMetadataCache {
private final long refreshMs;
private final int inputSchemasPerTableCacheMaximumSize;
private final Map<TableIdentifier, CacheItem> tableCache;
+ private final boolean caseSensitive;
Review Comment:
So far, the case insensitivity only extends to fields. Are you suggesting to also extend it to table names?
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicIcebergSink.java:
##########
@@ -336,6 +337,15 @@ public Builder<T> inputSchemasPerTableCacheMaxSize(int inputSchemasPerTableCache
return this;
}
+ /**
+ * Set whether schema field name matching should be case-sensitive. The default is to match the
+ * field names case-sensitive.
+ */
+ public Builder<T> caseSensitive(boolean newCaseSensitive) {
Review Comment:
Yes, that would be a good addition.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Add case-insensitive field matching [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #14729:
URL: https://github.com/apache/iceberg/pull/14729#discussion_r2705219135
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/EvolveSchemaVisitor.java:
##########
@@ -89,14 +91,23 @@ public Boolean struct(Types.StructType struct, Integer partnerId, List<Boolean>
Types.StructType partnerStruct = findFieldType(partnerId).asStructType();
String after = null;
for (Types.NestedField targetField : struct.fields()) {
- Types.NestedField nestedField = partnerStruct.field(targetField.name());
+ Types.NestedField nestedField =
+ caseSensitive
+ ? partnerStruct.field(targetField.name())
+ : partnerStruct.caseInsensitiveField(targetField.name());
final String columnName;
if (nestedField != null) {
updateColumn(nestedField, targetField);
columnName = this.existingSchema.findColumnName(nestedField.fieldId());
} else {
addColumn(partnerId, targetField);
- columnName = this.targetSchema.findColumnName(targetField.fieldId());
+ // We need to lookup the proper parent name because the letter case might be different on
+ // table vs target parent field (e.g. struct1.FIELD2 vs STRUCT1.STRUCT2). This is actually a
+ // bug in SchemaUpdate#moveAfter(columnName, after) because added fields are case-sensitive,
+ // even if case sensitivity is disabled on the API.
Review Comment:
Ack, fixed.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Add case-insensitive field matching [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #14729:
URL: https://github.com/apache/iceberg/pull/14729#discussion_r2588806377
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicIcebergSink.java:
##########
@@ -336,6 +337,15 @@ public Builder<T> inputSchemasPerTableCacheMaxSize(int inputSchemasPerTableCache
return this;
}
+ /**
+ * Set whether schema field name matching should be case-sensitive. The default is to match the
+ * field names case-sensitive.
+ */
+ public Builder<T> caseSensitive(boolean newCaseSensitive) {
Review Comment:
Then better not to make any assumptions here because we don't control their catalog logic.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Add case-insensitive field matching [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on PR #14729:
URL: https://github.com/apache/iceberg/pull/14729#issuecomment-3755400860
Rebased after these have been merged:
- #14734
- #14728
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Add case-insensitive field matching [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #14729:
URL: https://github.com/apache/iceberg/pull/14729#discussion_r2705034225
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicIcebergSink.java:
##########
@@ -418,11 +429,12 @@ public DataStreamSink<DynamicRecordInternal> append() {
.map(
new DynamicTableUpdateOperator(
catalogLoader,
- dropUnusedColumns,
Review Comment:
Same reason as https://github.com/apache/iceberg/pull/14729#discussion_r2705033372 (consistency).
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Add case-insensitive field matching [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14729:
URL: https://github.com/apache/iceberg/pull/14729#discussion_r2580465101
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicIcebergSink.java:
##########
@@ -336,6 +337,15 @@ public Builder<T> inputSchemasPerTableCacheMaxSize(int inputSchemasPerTableCache
return this;
}
+ /**
+ * Set whether schema field name matching should be case-sensitive. The default is to match the
+ * field names case-sensitive.
+ */
+ public Builder<T> caseSensitive(boolean newCaseSensitive) {
Review Comment:
Shall we use this to caseSensitive table name matching?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Add case-insensitive field matching [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14729:
URL: https://github.com/apache/iceberg/pull/14729#discussion_r2580415727
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/CompareSchemasVisitor.java:
##########
@@ -88,7 +87,11 @@ public Result struct(Types.StructType struct, Integer tableSchemaId, List<Result
}
for (Types.NestedField tableField : tableSchemaType.asStructType().fields()) {
- if (tableField.isRequired() && struct.field(tableField.name()) == null) {
+ if (tableField.isRequired()
+ && (caseSensitive
+ ? struct.field(tableField.name())
+ : struct.caseInsensitiveField(tableField.name()))
+ == null) {
Review Comment:
nit: Maybe create a method like `equalFieldNames` or something better named ๐, and then we hide the ugly details, and don't increase complexity?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Add case-insensitive field matching [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14729:
URL: https://github.com/apache/iceberg/pull/14729#discussion_r2580448142
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableMetadataCache.java:
##########
@@ -52,13 +52,19 @@ class TableMetadataCache {
private final long refreshMs;
private final int inputSchemasPerTableCacheMaximumSize;
private final Map<TableIdentifier, CacheItem> tableCache;
+ private final boolean caseSensitive;
Review Comment:
Shall we use it to change the keys in the cache to be also not `caseSensitive`?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Add case-insensitive field matching [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14729:
URL: https://github.com/apache/iceberg/pull/14729#discussion_r2580487426
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/EvolveSchemaVisitor.java:
##########
@@ -89,14 +91,23 @@ public Boolean struct(Types.StructType struct, Integer partnerId, List<Boolean>
Types.StructType partnerStruct = findFieldType(partnerId).asStructType();
String after = null;
for (Types.NestedField targetField : struct.fields()) {
- Types.NestedField nestedField = partnerStruct.field(targetField.name());
+ Types.NestedField nestedField =
+ caseSensitive
+ ? partnerStruct.field(targetField.name())
+ : partnerStruct.caseInsensitiveField(targetField.name());
final String columnName;
if (nestedField != null) {
updateColumn(nestedField, targetField);
columnName = this.existingSchema.findColumnName(nestedField.fieldId());
} else {
addColumn(partnerId, targetField);
- columnName = this.targetSchema.findColumnName(targetField.fieldId());
+ // We need to lookup the proper parent name because the letter case might be different on
+ // table vs target parent field (e.g. struct1.FIELD2 vs STRUCT1.STRUCT2). This is actually a
+ // bug in SchemaUpdate#moveAfter(columnName, after) because added fields are case-sensitive,
+ // even if case sensitivity is disabled on the API.
Review Comment:
Maybe do it first?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Add case-insensitive field matching [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14729:
URL: https://github.com/apache/iceberg/pull/14729#discussion_r2588375050
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicIcebergSink.java:
##########
@@ -336,6 +337,15 @@ public Builder<T> inputSchemasPerTableCacheMaxSize(int inputSchemasPerTableCache
return this;
}
+ /**
+ * Set whether schema field name matching should be case-sensitive. The default is to match the
+ * field names case-sensitive.
+ */
+ public Builder<T> caseSensitive(boolean newCaseSensitive) {
Review Comment:
If I remember correctly HMS is case insensitive (everything is stored in lower case)
If I remember correctly JDBC is case sensitive (the names used without changes)
So catalogs behave differently wrt case sensitivity
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Add case-insensitive field matching [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14729:
URL: https://github.com/apache/iceberg/pull/14729#discussion_r2704079848
##########
flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestEvolveSchemaVisitor.java:
##########
@@ -682,6 +705,78 @@ private static Schema getNestedSchemaWithOptionalModifier(boolean nestedIsOption
9, "s4", StringType.get()))))))))))))));
}
+ @Test
+ public void testCaseInsensitiveAddField() {
+ Schema existingSchema =
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "name", Types.StringType.get()));
+ Schema targetSchema =
+ new Schema(
+ Types.NestedField.optional(1, "ID", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "NAME", Types.StringType.get()),
+ Types.NestedField.optional(3, "AGE", Types.IntegerType.get()));
+
+ UpdateSchema updateApi = loadUpdateApi(existingSchema);
+ EvolveSchemaVisitor.visit(TABLE, updateApi, existingSchema, targetSchema, false, false);
+ Schema result = updateApi.apply();
+ assertThat(result.columns()).hasSize(3);
+ assertThat(result.findField("AGE")).isNotNull();
+ }
+
+ @Test
+ public void testCaseInsensitiveMakeFieldOptional() {
+ Schema existingSchema =
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.IntegerType.get()),
+ Types.NestedField.required(2, "name", Types.StringType.get()));
+ Schema targetSchema = new Schema(Types.NestedField.optional(1, "ID", Types.IntegerType.get()));
+
+ UpdateSchema updateApi = loadUpdateApi(existingSchema);
+ EvolveSchemaVisitor.visit(TABLE, updateApi, existingSchema, targetSchema, false, false);
+ Schema result = updateApi.apply();
+ assertThat(result.findField("name").isOptional()).isTrue();
+ }
+
+ @Test
+ public void testCaseInsensitiveNestedStructField() {
+ Schema existingSchema =
+ new Schema(
+ optional(1, "struct1", StructType.of(optional(2, "field1", Types.StringType.get()))));
+ Schema targetSchema =
+ new Schema(
+ optional(
+ 1,
+ "STRUCT1",
+ StructType.of(
+ optional(2, "FIELD1", Types.StringType.get()),
+ optional(3, "FIELD2", Types.IntegerType.get()))));
+
+ UpdateSchema updateApi = loadUpdateApi(existingSchema);
+ EvolveSchemaVisitor.visit(TABLE, updateApi, existingSchema, targetSchema, false, false);
+ Schema result = updateApi.apply();
+ Types.StructType struct = result.findField("struct1").type().asStructType();
+ assertThat(struct.fields()).hasSize(2);
+ assertThat(struct.field("FIELD2")).isNotNull();
+ }
+
+ @Test
+ public void testCaseSensitiveDoesNotMatch() {
+ Schema existingSchema =
+ new Schema(Types.NestedField.optional(1, "id", Types.IntegerType.get()));
+ Schema targetSchema =
+ new Schema(
+ Types.NestedField.optional(1, "ID", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "name", Types.StringType.get()));
+
+ UpdateSchema updateApi = loadUpdateApi(existingSchema);
+ EvolveSchemaVisitor.visit(TABLE, updateApi, existingSchema, targetSchema, true, false);
Review Comment:
This is why we try to avoid boolean parameters. It is really hard to understand what's happening here. We usually add a comment `/* caseSensitive */`, or if it often used, you might want to use a constant
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Add case-insensitive field matching [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #14729:
URL: https://github.com/apache/iceberg/pull/14729#issuecomment-3708573615
This pull request has been marked as stale due to 30 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think thatโs incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the dev@iceberg.apache.org list. Thank you for your contributions.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Add case-insensitive field matching [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14729:
URL: https://github.com/apache/iceberg/pull/14729#discussion_r2580442189
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/EvolveSchemaVisitor.java:
##########
@@ -89,14 +91,23 @@ public Boolean struct(Types.StructType struct, Integer partnerId, List<Boolean>
Types.StructType partnerStruct = findFieldType(partnerId).asStructType();
String after = null;
for (Types.NestedField targetField : struct.fields()) {
- Types.NestedField nestedField = partnerStruct.field(targetField.name());
+ Types.NestedField nestedField =
+ caseSensitive
+ ? partnerStruct.field(targetField.name())
+ : partnerStruct.caseInsensitiveField(targetField.name());
Review Comment:
This is again the same check. Maybe an util method somewhere?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Add case-insensitive field matching [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14729:
URL: https://github.com/apache/iceberg/pull/14729#discussion_r2580495474
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicIcebergSink.java:
##########
@@ -336,6 +337,15 @@ public Builder<T> inputSchemasPerTableCacheMaxSize(int inputSchemasPerTableCache
return this;
}
+ /**
+ * Set whether schema field name matching should be case-sensitive. The default is to match the
+ * field names case-sensitive.
+ */
+ public Builder<T> caseSensitive(boolean newCaseSensitive) {
Review Comment:
It really depends on the catalog - it needs to match the case sensitivity of the setting here. Otherwise it would be unfeasible to find the table ๐
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Add case-insensitive field matching [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #14729:
URL: https://github.com/apache/iceberg/pull/14729#discussion_r2580472290
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/EvolveSchemaVisitor.java:
##########
@@ -89,14 +91,23 @@ public Boolean struct(Types.StructType struct, Integer partnerId, List<Boolean>
Types.StructType partnerStruct = findFieldType(partnerId).asStructType();
String after = null;
for (Types.NestedField targetField : struct.fields()) {
- Types.NestedField nestedField = partnerStruct.field(targetField.name());
+ Types.NestedField nestedField =
+ caseSensitive
+ ? partnerStruct.field(targetField.name())
+ : partnerStruct.caseInsensitiveField(targetField.name());
final String columnName;
if (nestedField != null) {
updateColumn(nestedField, targetField);
columnName = this.existingSchema.findColumnName(nestedField.fieldId());
} else {
addColumn(partnerId, targetField);
- columnName = this.targetSchema.findColumnName(targetField.fieldId());
+ // We need to lookup the proper parent name because the letter case might be different on
+ // table vs target parent field (e.g. struct1.FIELD2 vs STRUCT1.STRUCT2). This is actually a
+ // bug in SchemaUpdate#moveAfter(columnName, after) because added fields are case-sensitive,
+ // even if case sensitivity is disabled on the API.
Review Comment:
Absolutely, if feasible in the scope of this PR. Otherwise we need to follow up.
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/CompareSchemasVisitor.java:
##########
@@ -88,7 +87,11 @@ public Result struct(Types.StructType struct, Integer tableSchemaId, List<Result
}
for (Types.NestedField tableField : tableSchemaType.asStructType().fields()) {
- if (tableField.isRequired() && struct.field(tableField.name()) == null) {
+ if (tableField.isRequired()
+ && (caseSensitive
+ ? struct.field(tableField.name())
+ : struct.caseInsensitiveField(tableField.name()))
+ == null) {
Review Comment:
Thanks, we consolidate this!
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Add case-insensitive field matching [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #14729:
URL: https://github.com/apache/iceberg/pull/14729#discussion_r2592246598
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicIcebergSink.java:
##########
@@ -336,6 +337,15 @@ public Builder<T> inputSchemasPerTableCacheMaxSize(int inputSchemasPerTableCache
return this;
}
+ /**
+ * Set whether schema field name matching should be case-sensitive. The default is to match the
+ * field names case-sensitive.
+ */
+ public Builder<T> caseSensitive(boolean newCaseSensitive) {
Review Comment:
I agree. It's kind of unfortunate that we cannot easily extend this feature.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Add case-insensitive field matching [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #14729:
URL: https://github.com/apache/iceberg/pull/14729#discussion_r2592142092
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicIcebergSink.java:
##########
@@ -336,6 +337,15 @@ public Builder<T> inputSchemasPerTableCacheMaxSize(int inputSchemasPerTableCache
return this;
}
+ /**
+ * Set whether schema field name matching should be case-sensitive. The default is to match the
+ * field names case-sensitive.
+ */
+ public Builder<T> caseSensitive(boolean newCaseSensitive) {
Review Comment:
We can help detecting it, but simply using lower case in the cache won't do it. We need a detection mechanism, which I elaborated on in my previous post. The reason is that there might actually be two tables: "TABLE" and "table"; without the detection logic I mentioned, there is no way to tell them apart in case of case insensitivity. We might accidentally commit to "table", when we wanted to commit to "TABLE". Field names don't have that issue because we will fail if there is "field" and "FIELD" with case insensitivity.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Add case-insensitive field matching [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #14729:
URL: https://github.com/apache/iceberg/pull/14729#discussion_r2704764134
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/CompareSchemasVisitor.java:
##########
@@ -43,23 +44,22 @@ public class CompareSchemasVisitor
extends SchemaWithPartnerVisitor<Integer, CompareSchemasVisitor.Result> {
private final Schema tableSchema;
+ private final boolean caseSensitive;
private final boolean dropUnusedColumns;
- private CompareSchemasVisitor(Schema tableSchema, boolean dropUnusedColumns) {
+ private CompareSchemasVisitor(
+ Schema tableSchema, boolean caseSensitive, boolean dropUnusedColumns) {
this.tableSchema = tableSchema;
+ this.caseSensitive = caseSensitive;
this.dropUnusedColumns = dropUnusedColumns;
}
- public static Result visit(Schema dataSchema, Schema tableSchema) {
- return visit(dataSchema, tableSchema, true, false);
- }
Review Comment:
Ack, re-added and deprecated.
##########
flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestEvolveSchemaVisitor.java:
##########
@@ -94,7 +94,7 @@ private static Types.NestedField[] primitiveFields(
public void testAddTopLevelPrimitives() {
Schema targetSchema = new Schema(primitiveFields(0, primitiveTypes()));
UpdateSchema updateApi = loadUpdateApi(new Schema());
- EvolveSchemaVisitor.visit(TABLE, updateApi, new Schema(), targetSchema, PRESERVE_COLUMNS);
+ EvolveSchemaVisitor.visit(TABLE, updateApi, new Schema(), targetSchema, true, PRESERVE_COLUMNS);
Review Comment:
Yes, I added a constant.
##########
flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestEvolveSchemaVisitor.java:
##########
@@ -682,6 +705,78 @@ private static Schema getNestedSchemaWithOptionalModifier(boolean nestedIsOption
9, "s4", StringType.get()))))))))))))));
}
+ @Test
+ public void testCaseInsensitiveAddField() {
+ Schema existingSchema =
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "name", Types.StringType.get()));
+ Schema targetSchema =
+ new Schema(
+ Types.NestedField.optional(1, "ID", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "NAME", Types.StringType.get()),
+ Types.NestedField.optional(3, "AGE", Types.IntegerType.get()));
+
+ UpdateSchema updateApi = loadUpdateApi(existingSchema);
+ EvolveSchemaVisitor.visit(TABLE, updateApi, existingSchema, targetSchema, false, false);
+ Schema result = updateApi.apply();
+ assertThat(result.columns()).hasSize(3);
+ assertThat(result.findField("AGE")).isNotNull();
+ }
+
+ @Test
+ public void testCaseInsensitiveMakeFieldOptional() {
+ Schema existingSchema =
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.IntegerType.get()),
+ Types.NestedField.required(2, "name", Types.StringType.get()));
+ Schema targetSchema = new Schema(Types.NestedField.optional(1, "ID", Types.IntegerType.get()));
+
+ UpdateSchema updateApi = loadUpdateApi(existingSchema);
+ EvolveSchemaVisitor.visit(TABLE, updateApi, existingSchema, targetSchema, false, false);
+ Schema result = updateApi.apply();
+ assertThat(result.findField("name").isOptional()).isTrue();
+ }
+
+ @Test
+ public void testCaseInsensitiveNestedStructField() {
+ Schema existingSchema =
+ new Schema(
+ optional(1, "struct1", StructType.of(optional(2, "field1", Types.StringType.get()))));
+ Schema targetSchema =
+ new Schema(
+ optional(
+ 1,
+ "STRUCT1",
+ StructType.of(
+ optional(2, "FIELD1", Types.StringType.get()),
+ optional(3, "FIELD2", Types.IntegerType.get()))));
+
+ UpdateSchema updateApi = loadUpdateApi(existingSchema);
+ EvolveSchemaVisitor.visit(TABLE, updateApi, existingSchema, targetSchema, false, false);
+ Schema result = updateApi.apply();
+ Types.StructType struct = result.findField("struct1").type().asStructType();
+ assertThat(struct.fields()).hasSize(2);
+ assertThat(struct.field("FIELD2")).isNotNull();
+ }
+
+ @Test
+ public void testCaseSensitiveDoesNotMatch() {
+ Schema existingSchema =
+ new Schema(Types.NestedField.optional(1, "id", Types.IntegerType.get()));
+ Schema targetSchema =
+ new Schema(
+ Types.NestedField.optional(1, "ID", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "name", Types.StringType.get()));
+
+ UpdateSchema updateApi = loadUpdateApi(existingSchema);
+ EvolveSchemaVisitor.visit(TABLE, updateApi, existingSchema, targetSchema, true, false);
Review Comment:
I agree, it can be confusing and error-prone. Fixed.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Add case-insensitive field matching [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary merged PR #14729:
URL: https://github.com/apache/iceberg/pull/14729
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Add case-insensitive field matching [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14729:
URL: https://github.com/apache/iceberg/pull/14729#discussion_r2704053275
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/EvolveSchemaVisitor.java:
##########
@@ -89,14 +91,23 @@ public Boolean struct(Types.StructType struct, Integer partnerId, List<Boolean>
Types.StructType partnerStruct = findFieldType(partnerId).asStructType();
String after = null;
for (Types.NestedField targetField : struct.fields()) {
- Types.NestedField nestedField = partnerStruct.field(targetField.name());
+ Types.NestedField nestedField =
+ caseSensitive
+ ? partnerStruct.field(targetField.name())
+ : partnerStruct.caseInsensitiveField(targetField.name());
final String columnName;
if (nestedField != null) {
updateColumn(nestedField, targetField);
columnName = this.existingSchema.findColumnName(nestedField.fieldId());
} else {
addColumn(partnerId, targetField);
- columnName = this.targetSchema.findColumnName(targetField.fieldId());
+ // We need to lookup the proper parent name because the letter case might be different on
+ // table vs target parent field (e.g. struct1.FIELD2 vs STRUCT1.STRUCT2). This is actually a
+ // bug in SchemaUpdate#moveAfter(columnName, after) because added fields are case-sensitive,
+ // even if case sensitivity is disabled on the API.
Review Comment:
Could we remove/reword the comment after the bugfix is there?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Add case-insensitive field matching [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14729:
URL: https://github.com/apache/iceberg/pull/14729#discussion_r2704037393
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/CompareSchemasVisitor.java:
##########
@@ -43,23 +44,22 @@ public class CompareSchemasVisitor
extends SchemaWithPartnerVisitor<Integer, CompareSchemasVisitor.Result> {
private final Schema tableSchema;
+ private final boolean caseSensitive;
private final boolean dropUnusedColumns;
- private CompareSchemasVisitor(Schema tableSchema, boolean dropUnusedColumns) {
+ private CompareSchemasVisitor(
+ Schema tableSchema, boolean caseSensitive, boolean dropUnusedColumns) {
this.tableSchema = tableSchema;
+ this.caseSensitive = caseSensitive;
this.dropUnusedColumns = dropUnusedColumns;
}
- public static Result visit(Schema dataSchema, Schema tableSchema) {
- return visit(dataSchema, tableSchema, true, false);
- }
Review Comment:
We can't remove this ATM, as this is public. We need to deprecate.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Add case-insensitive field matching [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14729:
URL: https://github.com/apache/iceberg/pull/14729#discussion_r2704074407
##########
flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestEvolveSchemaVisitor.java:
##########
@@ -94,7 +94,7 @@ private static Types.NestedField[] primitiveFields(
public void testAddTopLevelPrimitives() {
Schema targetSchema = new Schema(primitiveFields(0, primitiveTypes()));
UpdateSchema updateApi = loadUpdateApi(new Schema());
- EvolveSchemaVisitor.visit(TABLE, updateApi, new Schema(), targetSchema, PRESERVE_COLUMNS);
+ EvolveSchemaVisitor.visit(TABLE, updateApi, new Schema(), targetSchema, true, PRESERVE_COLUMNS);
Review Comment:
Could we use a constant for this, similar to `PRESERVE_COLUMNS`. It will be more descriptive
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Add case-insensitive field matching [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #14729:
URL: https://github.com/apache/iceberg/pull/14729#discussion_r2705033372
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordProcessor.java:
##########
@@ -57,19 +58,21 @@ class DynamicRecordProcessor<T> extends ProcessFunction<T, DynamicRecordInternal
DynamicRecordGenerator<T> generator,
CatalogLoader catalogLoader,
boolean immediateUpdate,
- boolean dropUnusedColumns,
int cacheMaximumSize,
long cacheRefreshMs,
int inputSchemasPerTableCacheMaximumSize,
- TableCreator tableCreator) {
+ TableCreator tableCreator,
+ boolean caseSensitive,
+ boolean dropUnusedColumns) {
this.generator = generator;
this.catalogLoader = catalogLoader;
this.immediateUpdate = immediateUpdate;
- this.dropUnusedColumns = dropUnusedColumns;
Review Comment:
This parameter has been moved to the end, for consistency with the rest of the code base.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Add case-insensitive field matching [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on PR #14729:
URL: https://github.com/apache/iceberg/pull/14729#issuecomment-3755385687
Rebased.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Add case-insensitive field matching [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #14729:
URL: https://github.com/apache/iceberg/pull/14729#discussion_r2581105249
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/EvolveSchemaVisitor.java:
##########
@@ -89,14 +91,23 @@ public Boolean struct(Types.StructType struct, Integer partnerId, List<Boolean>
Types.StructType partnerStruct = findFieldType(partnerId).asStructType();
String after = null;
for (Types.NestedField targetField : struct.fields()) {
- Types.NestedField nestedField = partnerStruct.field(targetField.name());
+ Types.NestedField nestedField =
+ caseSensitive
+ ? partnerStruct.field(targetField.name())
+ : partnerStruct.caseInsensitiveField(targetField.name());
final String columnName;
if (nestedField != null) {
updateColumn(nestedField, targetField);
columnName = this.existingSchema.findColumnName(nestedField.fieldId());
} else {
addColumn(partnerId, targetField);
- columnName = this.targetSchema.findColumnName(targetField.fieldId());
+ // We need to lookup the proper parent name because the letter case might be different on
+ // table vs target parent field (e.g. struct1.FIELD2 vs STRUCT1.STRUCT2). This is actually a
+ // bug in SchemaUpdate#moveAfter(columnName, after) because added fields are case-sensitive,
+ // even if case sensitivity is disabled on the API.
Review Comment:
Fix here: https://github.com/apache/iceberg/pull/14734
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Add case-insensitive field matching [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14729:
URL: https://github.com/apache/iceberg/pull/14729#discussion_r2592118698
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicIcebergSink.java:
##########
@@ -336,6 +337,15 @@ public Builder<T> inputSchemasPerTableCacheMaxSize(int inputSchemasPerTableCache
return this;
}
+ /**
+ * Set whether schema field name matching should be case-sensitive. The default is to match the
+ * field names case-sensitive.
+ */
+ public Builder<T> caseSensitive(boolean newCaseSensitive) {
Review Comment:
So we are saying: If the user is providing wrong parameters, we can't help him in this case? There night be a silent data loss, but we can't help detecting it.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Add case-insensitive field matching [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #14729:
URL: https://github.com/apache/iceberg/pull/14729#issuecomment-3741143868
This pull request has been closed due to lack of activity. This is not a judgement on the merit of the PR in any way. It is just a way of keeping the PR queue manageable. If you think that is incorrect, or the pull request requires review, you can revive the PR at any time.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Add case-insensitive field matching [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14729:
URL: https://github.com/apache/iceberg/pull/14729#discussion_r2704989181
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicIcebergSink.java:
##########
@@ -418,11 +429,12 @@ public DataStreamSink<DynamicRecordInternal> append() {
.map(
new DynamicTableUpdateOperator(
catalogLoader,
- dropUnusedColumns,
Review Comment:
Why did we move the dropUnusedColumns?
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicIcebergSink.java:
##########
@@ -418,11 +429,12 @@ public DataStreamSink<DynamicRecordInternal> append() {
.map(
new DynamicTableUpdateOperator(
catalogLoader,
- dropUnusedColumns,
Review Comment:
Why did we move the `dropUnusedColumns`?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Add case-insensitive field matching [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on PR #14729:
URL: https://github.com/apache/iceberg/pull/14729#issuecomment-3769314077
Merged to main.
Thanks @mxm for adding the feature!
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Add case-insensitive field matching [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #14729:
URL: https://github.com/apache/iceberg/pull/14729#discussion_r2585505980
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableMetadataCache.java:
##########
@@ -52,13 +52,19 @@ class TableMetadataCache {
private final long refreshMs;
private final int inputSchemasPerTableCacheMaximumSize;
private final Map<TableIdentifier, CacheItem> tableCache;
+ private final boolean caseSensitive;
Review Comment:
We would have to probe the catalog for both casings to make sure that we do not load the wrong table. If the table exists for both cases, we would have to throw an error, unless it is the same table. Catalogs don't seem to have a notion of case insensitivity, in contrast to Schemas.
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicIcebergSink.java:
##########
@@ -336,6 +337,15 @@ public Builder<T> inputSchemasPerTableCacheMaxSize(int inputSchemasPerTableCache
return this;
}
+ /**
+ * Set whether schema field name matching should be case-sensitive. The default is to match the
+ * field names case-sensitive.
+ */
+ public Builder<T> caseSensitive(boolean newCaseSensitive) {
Review Comment:
Thinking about it more, I'm not sure we should extend case insensitivity to the table identifiers. It doesn't seem like catalogs currently have a notion of case insensitivity, in contrast to Schemas.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Add case-insensitive field matching [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #14729:
URL: https://github.com/apache/iceberg/pull/14729#discussion_r2580472866
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/CompareSchemasVisitor.java:
##########
@@ -88,7 +87,11 @@ public Result struct(Types.StructType struct, Integer tableSchemaId, List<Result
}
for (Types.NestedField tableField : tableSchemaType.asStructType().fields()) {
- if (tableField.isRequired() && struct.field(tableField.name()) == null) {
+ if (tableField.isRequired()
+ && (caseSensitive
+ ? struct.field(tableField.name())
+ : struct.caseInsensitiveField(tableField.name()))
+ == null) {
Review Comment:
Thanks, I'll consolidate this!
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Add case-insensitive field matching [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #14729:
URL: https://github.com/apache/iceberg/pull/14729#discussion_r2592071389
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicIcebergSink.java:
##########
@@ -336,6 +337,15 @@ public Builder<T> inputSchemasPerTableCacheMaxSize(int inputSchemasPerTableCache
return this;
}
+ /**
+ * Set whether schema field name matching should be case-sensitive. The default is to match the
+ * field names case-sensitive.
+ */
+ public Builder<T> caseSensitive(boolean newCaseSensitive) {
Review Comment:
>In cache, we are fine if we are insensitive - we are just adding more items to the cache.
It's not so simple. The cache itself uses the TableIdentifier to refresh the table. Are we refreshing using the user-provided TableIdentifer, or are we using the lower case variant? IMHO We need to use whatever the user provides to load the correct table from the catalog. When we load the table, the only way to tell would be to load both case sensitive and case insensitive; if both return results and they are not the same table => FAIL; if it's the same table or we get only one result => use that table. That's the same mechanism Schema uses.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Add case-insensitive field matching [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14729:
URL: https://github.com/apache/iceberg/pull/14729#discussion_r2592184883
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicIcebergSink.java:
##########
@@ -336,6 +337,15 @@ public Builder<T> inputSchemasPerTableCacheMaxSize(int inputSchemasPerTableCache
return this;
}
+ /**
+ * Set whether schema field name matching should be case-sensitive. The default is to match the
+ * field names case-sensitive.
+ */
+ public Builder<T> caseSensitive(boolean newCaseSensitive) {
Review Comment:
I don't think we would like to have a double call to the catalog on the hot path.
So I guess we have to skip this for now.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org