mxm commented on code in PR #18517:
URL: https://github.com/apache/iceberg/pull/18517#discussion_r4400137743
##########
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:
Could we add a test for the case-insensitive branch?
##########
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);
+ }
Review Comment:
NIT: I don't think it is possible for the current schema to be null. But if
we want to be extra safe:
```suggestion
Schema current = tableSchemas.get(currentSchemaId);
Preconditions.checkArgument(current != null, "Current schema %s not
found", currentSchemaId);
ordered.put(currentSchemaId, current);
```
##########
flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableMetadataCache.java:
##########
@@ -239,4 +241,138 @@ void testCaseSensitiveCachingDoesNotMatch() {
assertThat(cache.schema(tableIdentifier, SCHEMA_MIXEDCASE))
.isEqualTo(TableMetadataCache.NOT_FOUND);
}
+
+ @Test
+ void testSchemaPrefersCurrentSchemaAfterDropAndReAddColumn() {
+ Catalog catalog = CATALOG_EXTENSION.catalog();
+ TableIdentifier tableIdentifier = TableIdentifier.parse("default.myTable");
+ Table table = catalog.createTable(tableIdentifier, SCHEMA);
+ table.updateSchema().addColumn("extra", Types.StringType.get()).commit();
+ int retiredFieldId = table.schema().findField("extra").fieldId();
+ table.updateSchema().deleteColumn("extra").commit();
+ table.updateSchema().addColumn("extra", Types.StringType.get()).commit();
+ int currentFieldId = table.schema().findField("extra").fieldId();
+ assertThat(currentFieldId).isNotEqualTo(retiredFieldId);
+
+ TableMetadataCache cache =
+ new TableMetadataCache(catalog, 10, Long.MAX_VALUE, 10,
CASE_SENSITIVE, PRESERVE_COLUMNS);
+
+ Schema resolved = cache.schema(tableIdentifier,
table.schema()).resolvedTableSchema();
+
+
assertThat(resolved.findField("extra").fieldId()).isEqualTo(currentFieldId);
+ }
+
+ @Test
+ void testSchemaStillMatchesHistoricalSchemaWhenDroppedColumnIsNotReAdded() {
+ Catalog catalog = CATALOG_EXTENSION.catalog();
+ TableIdentifier tableIdentifier = TableIdentifier.parse("default.myTable");
+ Table table = catalog.createTable(tableIdentifier, SCHEMA);
+ table.updateSchema().addColumn("extra", Types.StringType.get()).commit();
+ Schema withExtra = table.schema();
+ table.updateSchema().deleteColumn("extra").commit();
+
+ TableMetadataCache cache =
+ new TableMetadataCache(catalog, 10, Long.MAX_VALUE, 10,
CASE_SENSITIVE, PRESERVE_COLUMNS);
+
+ Schema resolved = cache.schema(tableIdentifier,
withExtra).resolvedTableSchema();
+
+ assertThat(resolved.findField("extra")).isNotNull();
Review Comment:
`isNotNull()` passes for any schema that has `extra`. Could we assert the
field id of the old schema, as the other tests do?
```suggestion
assertThat(resolved.findField("extra").fieldId())
.isEqualTo(withExtra.findField("extra").fieldId());
```
##########
flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableMetadataCache.java:
##########
@@ -239,4 +241,138 @@ void testCaseSensitiveCachingDoesNotMatch() {
assertThat(cache.schema(tableIdentifier, SCHEMA_MIXEDCASE))
.isEqualTo(TableMetadataCache.NOT_FOUND);
}
+
+ @Test
+ void testSchemaPrefersCurrentSchemaAfterDropAndReAddColumn() {
+ Catalog catalog = CATALOG_EXTENSION.catalog();
+ TableIdentifier tableIdentifier = TableIdentifier.parse("default.myTable");
+ Table table = catalog.createTable(tableIdentifier, SCHEMA);
+ table.updateSchema().addColumn("extra", Types.StringType.get()).commit();
+ int retiredFieldId = table.schema().findField("extra").fieldId();
+ table.updateSchema().deleteColumn("extra").commit();
+ table.updateSchema().addColumn("extra", Types.StringType.get()).commit();
+ int currentFieldId = table.schema().findField("extra").fieldId();
+ assertThat(currentFieldId).isNotEqualTo(retiredFieldId);
+
+ TableMetadataCache cache =
+ new TableMetadataCache(catalog, 10, Long.MAX_VALUE, 10,
CASE_SENSITIVE, PRESERVE_COLUMNS);
+
+ Schema resolved = cache.schema(tableIdentifier,
table.schema()).resolvedTableSchema();
+
+
assertThat(resolved.findField("extra").fieldId()).isEqualTo(currentFieldId);
+ }
+
+ @Test
+ void testSchemaStillMatchesHistoricalSchemaWhenDroppedColumnIsNotReAdded() {
+ Catalog catalog = CATALOG_EXTENSION.catalog();
+ TableIdentifier tableIdentifier = TableIdentifier.parse("default.myTable");
+ Table table = catalog.createTable(tableIdentifier, SCHEMA);
+ table.updateSchema().addColumn("extra", Types.StringType.get()).commit();
+ Schema withExtra = table.schema();
+ table.updateSchema().deleteColumn("extra").commit();
+
+ TableMetadataCache cache =
+ new TableMetadataCache(catalog, 10, Long.MAX_VALUE, 10,
CASE_SENSITIVE, PRESERVE_COLUMNS);
+
+ Schema resolved = cache.schema(tableIdentifier,
withExtra).resolvedTableSchema();
+
+ assertThat(resolved.findField("extra")).isNotNull();
+ }
+
+ @Test
+ void testSchemasAreOrderedCurrentFirstThenNewestToOldest() {
+ Map<Integer, Schema> schemas = Maps.newHashMap();
+ for (int id = 0; id < 4; id++) {
+ schemas.put(id, SCHEMA);
+ }
+
+ assertThat(TableMetadataCache.orderSchemas(schemas,
3).keySet()).containsExactly(3, 2, 1, 0);
+
+ // after a rollback the current schema is not the one with the highest id
+ assertThat(TableMetadataCache.orderSchemas(schemas,
1).keySet()).containsExactly(1, 3, 2, 0);
+ }
+
+ @Test
+ void testSchemaPrefersCurrentSchemaWhenColumnIsReAddedAfterAnotherChange() {
+ Catalog catalog = CATALOG_EXTENSION.catalog();
+ TableIdentifier tableIdentifier = TableIdentifier.parse("default.myTable");
+ Table table = catalog.createTable(tableIdentifier, SCHEMA);
+ table.updateSchema().addColumn("c", Types.StringType.get()).commit();
+ Schema oldShape = table.schema();
+ int retiredFieldId = oldShape.findField("c").fieldId();
+ table.updateSchema().deleteColumn("c").commit();
+ table.updateSchema().addColumn("d", Types.StringType.get()).commit();
+ table.updateSchema().addColumn("c", Types.StringType.get()).commit();
+ int currentFieldId = table.schema().findField("c").fieldId();
+ assertThat(currentFieldId).isNotEqualTo(retiredFieldId);
+
+ TableMetadataCache cache =
+ new TableMetadataCache(catalog, 10, Long.MAX_VALUE, 10,
CASE_SENSITIVE, PRESERVE_COLUMNS);
+
+ Schema resolved = cache.schema(tableIdentifier,
oldShape).resolvedTableSchema();
+
+ assertThat(resolved.findField("c").fieldId()).isEqualTo(currentFieldId);
+ }
+
+ @Test
+ void testSchemaPrefersCurrentSchemaWhenColumnIsReAddedWithWiderType() {
+ Catalog catalog = CATALOG_EXTENSION.catalog();
+ TableIdentifier tableIdentifier = TableIdentifier.parse("default.myTable");
+ Table table = catalog.createTable(tableIdentifier, SCHEMA);
+ table.updateSchema().addColumn("c", Types.IntegerType.get()).commit();
+ Schema oldShape = table.schema();
+ int retiredFieldId = oldShape.findField("c").fieldId();
+ table.updateSchema().deleteColumn("c").commit();
+ table.updateSchema().addColumn("c", Types.LongType.get()).commit();
+ int currentFieldId = table.schema().findField("c").fieldId();
+ assertThat(currentFieldId).isNotEqualTo(retiredFieldId);
+
+ TableMetadataCache cache =
+ new TableMetadataCache(catalog, 10, Long.MAX_VALUE, 10,
CASE_SENSITIVE, PRESERVE_COLUMNS);
+
+ Schema resolved = cache.schema(tableIdentifier,
oldShape).resolvedTableSchema();
+
+ assertThat(resolved.findField("c").fieldId()).isEqualTo(currentFieldId);
+ assertThat(resolved.findField("c").type()).isEqualTo(Types.LongType.get());
Review Comment:
Could we add one `TestDynamicIcebergSink` test that writes an old-shape
record after the column was re-added and reads the value back through the
current schema?
--
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]
