adityaksolves commented on code in PR #18517:
URL: https://github.com/apache/iceberg/pull/18517#discussion_r4409259468
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableMetadataCache.java:
##########
@@ -259,26 +277,84 @@ public void invalidate(TableIdentifier identifier) {
tableCache.remove(identifier);
}
+ /**
+ * Orders the schemas with the current schema first, followed by the rest
from newest to oldest.
+ * The current schema is not necessarily the one with the highest id, for
example after a
+ * rollback.
+ */
+ static Map<Integer, Schema> orderSchemas(Map<Integer, Schema> tableSchemas,
int currentSchemaId) {
+ Map<Integer, Schema> ordered =
Maps.newLinkedHashMapWithExpectedSize(tableSchemas.size());
+ Schema current = tableSchemas.get(currentSchemaId);
+ if (current != null) {
+ ordered.put(currentSchemaId, current);
+ }
+
+ tableSchemas.entrySet().stream()
+ .filter(e -> e.getKey() != currentSchemaId)
+ .sorted(Map.Entry.<Integer, Schema>comparingByKey().reversed())
+ .forEach(e -> ordered.put(e.getKey(), e.getValue()));
+ return ordered;
+ }
+
+ /**
+ * Splits the ordered schemas into two tiers, keeping their order within
each tier.
+ *
+ * <p>The second tier holds the schemas which use a field name that belongs
to a different field
+ * id in the current schema, for example after a column was dropped and
added again, or renamed
+ * and its name reused. Data written with such a schema is stored under
field ids which are not
+ * read through the current schema, so these schemas are only used when no
schema of the first
+ * tier matches the input.
+ */
+ static List<List<Schema>> schemaTiers(
+ Map<Integer, Schema> orderedSchemas, Schema currentSchema, boolean
caseSensitive) {
+ List<Schema> preferred = Lists.newArrayList();
+ List<Schema> fallback = Lists.newArrayList();
+ for (Schema schema : orderedSchemas.values()) {
+ if (reusesFieldNames(schema, currentSchema, caseSensitive)) {
+ fallback.add(schema);
+ } else {
+ preferred.add(schema);
+ }
+ }
+
+ return ImmutableList.of(preferred, fallback);
+ }
+
+ private static boolean reusesFieldNames(
+ Schema schema, Schema currentSchema, boolean caseSensitive) {
+ for (Map.Entry<Integer, String> field :
TypeUtil.indexNameById(schema.asStruct()).entrySet()) {
+ Types.NestedField currentField =
+ caseSensitive
+ ? currentSchema.findField(field.getValue())
+ : currentSchema.caseInsensitiveFindField(field.getValue());
Review Comment:
Added `testSchemaMatchingOfReusedFieldNameHonorsCaseSensitivity` in e67fff9.
It renames `data` to `legacy` and adds `Data`. The case-insensitive cache
resolves to the new field id, the case-sensitive one still resolves to the old
schema. I also checked that flipping the branch makes the test fail.
--
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]
