This is an automated email from the ASF dual-hosted git repository.
pvary 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 b59c059430 Flink: Fix DynamicIcebergSink TableMetadataCache resolving
retired field ids (#18517)
b59c059430 is described below
commit b59c059430d334d6a381c484fb73bb5094d00741
Author: Aditya Kumar <[email protected]>
AuthorDate: Thu Nov 5 16:12:24 2026 +0530
Flink: Fix DynamicIcebergSink TableMetadataCache resolving retired field
ids (#18517)
---
docs/docs/flink-writes.md | 4 +
.../flink/sink/dynamic/TableMetadataCache.java | 102 ++++++++++--
.../flink/sink/dynamic/TestDynamicIcebergSink.java | 33 ++++
.../flink/sink/dynamic/TestTableMetadataCache.java | 176 +++++++++++++++++++++
4 files changed, 302 insertions(+), 13 deletions(-)
diff --git a/docs/docs/flink-writes.md b/docs/docs/flink-writes.md
index 6e1a4ec095..deece55bb8 100644
--- a/docs/docs/flink-writes.md
+++ b/docs/docs/flink-writes.md
@@ -524,6 +524,8 @@ The dynamic sink tries to match the schema provided in
`DynamicRecord` with the
- If there is no direct match, DynamicSink tries to adapt the provided schema
such that it matches one of table schemas. For example, if there is an
additional optional column in the table schema, a null value will be added to
the RowData provided via DynamicRecord.
- Otherwise, we evolve the table schema to match the input schema, within the
constraints described below.
+Schemas are matched by column name. The table schemas are checked starting
with the current schema, then from newest to oldest. A table schema which
matches the record exactly is preferred over one which needs a conversion (for
example `Integer` to `Long`), so a record which has the shape of an older table
schema is written with that schema, even if it would also fit the current one.
The exceptions are older table schemas which use a column name that belongs to
a different column in the [...]
+
The dynamic sink maintains an LRU cache for both table metadata and incoming
schemas, with eviction based on size and time constraints. When a DynamicRecord
contains a schema that is incompatible with the current table schema, a schema
update is triggered. This update can occur either immediately or via a
centralized executor, depending on the `immediateTableUpdate` configuration.
While centralized updates reduce load on the Catalog, they may introduce
backpressure on the sink.
#### Supported schema updates
@@ -541,6 +543,8 @@ regular queries won't be able to reference the column. If
the field was to re-ap
entirely new column would be added, which apart from the name, has nothing in
common with the old column, i.e. queries
for the new column will never return data of the old column.
+Records which still carry the old column are written to the new column when
their values can be converted to its type. If they cannot be converted (for
example because the column was re-added with an incompatible type), they are
still written using the older table schema.
+
##### Unsupported schema updates
- Renaming columns
diff --git
a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableMetadataCache.java
b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableMetadataCache.java
index 76f5909e65..9d836d56fb 100644
---
a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableMetadataCache.java
+++
b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableMetadataCache.java
@@ -19,6 +19,7 @@
package org.apache.iceberg.flink.sink.dynamic;
import java.time.Clock;
+import java.util.List;
import java.util.Map;
import java.util.Set;
import org.apache.flink.annotation.Internal;
@@ -32,6 +33,12 @@ import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.exceptions.NoSuchNamespaceException;
import org.apache.iceberg.exceptions.NoSuchTableException;
import org.apache.iceberg.flink.FlinkSchemaUtil;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.types.TypeUtil;
+import org.apache.iceberg.types.Types;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -123,7 +130,10 @@ class TableMetadataCache {
cacheRefreshClock.millis(),
true,
table.refs().keySet(),
- table.schemas(),
+ schemaTiers(
+ orderSchemas(table.schemas(), table.schema().schemaId()),
+ table.schema(),
+ caseSensitive),
table.specs(),
inputSchemasPerTableCacheMaximumSize));
}
@@ -177,15 +187,24 @@ class TableMetadataCache {
private ResolvedSchemaInfo resolveSchema(
TableIdentifier identifier, CacheItem cached, Schema input, boolean
allowRefresh) {
Schema compatible = null;
- for (Schema tableSchema : cached.tableSchemas.values()) {
- CompareSchemasVisitor.Result result =
- CompareSchemasVisitor.visit(input, tableSchema, caseSensitive,
dropUnusedColumns);
- if (result == CompareSchemasVisitor.Result.SAME) {
- return new ResolvedSchemaInfo(
- tableSchema, CompareSchemasVisitor.Result.SAME,
DataConverter.identity());
- } else if (compatible == null
- && result == CompareSchemasVisitor.Result.DATA_CONVERSION_NEEDED) {
- compatible = tableSchema;
+ // The schemas of a tier are only used if no schema of the previous tiers
matched the input.
+ // Within a tier, the first schema which is the same as the input wins,
otherwise the first
+ // schema which the input can be converted to. See schemaTiers for the
order of the schemas.
+ for (List<Schema> tableSchemas : cached.schemaTiers) {
+ for (Schema tableSchema : tableSchemas) {
+ CompareSchemasVisitor.Result result =
+ CompareSchemasVisitor.visit(input, tableSchema, caseSensitive,
dropUnusedColumns);
+ if (result == CompareSchemasVisitor.Result.SAME) {
+ return new ResolvedSchemaInfo(
+ tableSchema, CompareSchemasVisitor.Result.SAME,
DataConverter.identity());
+ } else if (compatible == null
+ && result == CompareSchemasVisitor.Result.DATA_CONVERSION_NEEDED) {
+ compatible = tableSchema;
+ }
+ }
+
+ if (compatible != null) {
+ break;
}
}
@@ -259,12 +278,69 @@ class TableMetadataCache {
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);
+ Preconditions.checkArgument(current != null, "Current schema %s not
found", currentSchemaId);
+ 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());
+ if (currentField != null && currentField.fieldId() != field.getKey()) {
+ return true;
+ }
+ }
+
+ return false;
+ }
+
/** Handles timeout for missing items only. Caffeine performance causes
noticeable delays. */
static class CacheItem {
private final long createdTimestampMillis;
private final boolean tableExists;
private final Set<String> branches;
- private final Map<Integer, Schema> tableSchemas;
+ private final List<List<Schema>> schemaTiers;
private final Map<Integer, PartitionSpec> specs;
private final Map<Schema, ResolvedSchemaInfo> inputSchemas;
@@ -272,13 +348,13 @@ class TableMetadataCache {
long createdTimestampMillis,
boolean tableExists,
Set<String> branches,
- Map<Integer, Schema> tableSchemas,
+ List<List<Schema>> schemaTiers,
Map<Integer, PartitionSpec> specs,
int inputSchemaCacheMaximumSize) {
this.createdTimestampMillis = createdTimestampMillis;
this.tableExists = tableExists;
this.branches = branches;
- this.tableSchemas = tableSchemas;
+ this.schemaTiers = schemaTiers;
this.specs = specs;
this.inputSchemas =
new LRUCache<>(inputSchemaCacheMaximumSize,
CacheItem::inputSchemaEvictionListener);
diff --git
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java
index 5cdbab9087..69730d9adb 100644
---
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java
+++
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java
@@ -2110,6 +2110,39 @@ class TestDynamicIcebergSink extends
TestFlinkIcebergSinkBase {
}
}
+ @Test
+ void testWriteOldShapeRecordAfterColumnReAddedWithWiderType() throws
Exception {
+ Schema base =
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "data", Types.StringType.get()));
+ Schema oldShape =
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "data", Types.StringType.get()),
+ Types.NestedField.optional(3, "c", Types.IntegerType.get()));
+
+ Catalog catalog = CATALOG_EXTENSION.catalog();
+ TableIdentifier tableIdentifier = TableIdentifier.of(DATABASE, "t1");
+ Table table = catalog.createTable(tableIdentifier, base);
+ table.updateSchema().addColumn("c", Types.IntegerType.get()).commit();
+ table.updateSchema().deleteColumn("c").commit();
+ table.updateSchema().addColumn("c", Types.LongType.get()).commit();
+
+ DynamicIcebergDataImpl record =
+ new DynamicIcebergDataImpl(
+ oldShape, "t1", SnapshotRef.MAIN_BRANCH,
PartitionSpec.unpartitioned());
+ record.rowProvided = Row.of(1, "a", 5);
+
+ executeDynamicSink(Lists.newArrayList(record), env, true, 1, null);
+
+ table.refresh();
+ List<Record> records =
Lists.newArrayList(IcebergGenerics.read(table).build());
+ assertThat(records).hasSize(1);
+ assertThat(records.get(0).getField("data")).isEqualTo("a");
+ assertThat(records.get(0).getField("c")).isEqualTo(5L);
+ }
+
private void runTest(List<DynamicIcebergDataImpl> dynamicData) throws
Exception {
runTest(dynamicData, this.env, 2);
}
diff --git
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableMetadataCache.java
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableMetadataCache.java
index bdf8045774..3bb62b45fb 100644
---
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableMetadataCache.java
+++
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableMetadataCache.java
@@ -24,6 +24,7 @@ import static org.assertj.core.api.Assertions.assertThat;
import java.time.Clock;
import java.time.Instant;
import java.time.ZoneId;
+import java.util.Map;
import org.apache.commons.lang3.SerializationUtils;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
@@ -32,6 +33,7 @@ import org.apache.iceberg.Table;
import org.apache.iceberg.catalog.Catalog;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.flink.sink.TestFlinkIcebergSinkBase;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.types.Types;
import org.junit.jupiter.api.Test;
@@ -239,4 +241,178 @@ public class TestTableMetadataCache extends
TestFlinkIcebergSinkBase {
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").fieldId())
+ .isEqualTo(withExtra.findField("extra").fieldId());
+ }
+
+ @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());
+ }
+
+ @Test
+ void testSchemaPrefersCurrentSchemaWhenFieldNameIsReusedAfterRename() {
+ Catalog catalog = CATALOG_EXTENSION.catalog();
+ TableIdentifier tableIdentifier = TableIdentifier.parse("default.myTable");
+ Table table = catalog.createTable(tableIdentifier, SCHEMA);
+ int renamedFieldId = table.schema().findField("data").fieldId();
+ table.updateSchema().renameColumn("data", "legacy").commit();
+ table.updateSchema().addColumn("data", Types.StringType.get()).commit();
+ int currentFieldId = table.schema().findField("data").fieldId();
+ assertThat(currentFieldId).isNotEqualTo(renamedFieldId);
+
+ TableMetadataCache cache =
+ new TableMetadataCache(catalog, 10, Long.MAX_VALUE, 10,
CASE_SENSITIVE, PRESERVE_COLUMNS);
+
+ Schema resolved = cache.schema(tableIdentifier,
SCHEMA).resolvedTableSchema();
+
+ assertThat(resolved.findField("data").fieldId()).isEqualTo(currentFieldId);
+ }
+
+ @Test
+ void testSchemaPrefersExactMatchOnOlderSchemaOverConversionToCurrentSchema()
{
+ Catalog catalog = CATALOG_EXTENSION.catalog();
+ TableIdentifier tableIdentifier = TableIdentifier.parse("default.myTable");
+ Table table = catalog.createTable(tableIdentifier, SCHEMA);
+ Schema oldShape = table.schema();
+ table.updateSchema().updateColumn("id", Types.LongType.get()).commit();
+
assertThat(table.schema().findField("id").type()).isEqualTo(Types.LongType.get());
+
+ TableMetadataCache cache =
+ new TableMetadataCache(catalog, 10, Long.MAX_VALUE, 10,
CASE_SENSITIVE, PRESERVE_COLUMNS);
+
+ Schema resolved = cache.schema(tableIdentifier,
oldShape).resolvedTableSchema();
+
+
assertThat(resolved.findField("id").type()).isEqualTo(Types.IntegerType.get());
+ }
+
+ @Test
+ void
testSchemaFallsBackToHistoricalSchemaWhenReAddedColumnTypeIsIncompatible() {
+ Catalog catalog = CATALOG_EXTENSION.catalog();
+ TableIdentifier tableIdentifier = TableIdentifier.parse("default.myTable");
+ Table table = catalog.createTable(tableIdentifier, SCHEMA);
+ table.updateSchema().addColumn("myfield", Types.StringType.get()).commit();
+ Schema oldShape = table.schema();
+ int retiredFieldId = oldShape.findField("myfield").fieldId();
+ table.updateSchema().deleteColumn("myfield").commit();
+ table.updateSchema().addColumn("myfield", Types.DoubleType.get()).commit();
+
assertThat(table.schema().findField("myfield").fieldId()).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("myfield").fieldId()).isEqualTo(retiredFieldId);
+
assertThat(resolved.findField("myfield").type()).isEqualTo(Types.StringType.get());
+ }
+
+ @Test
+ void testSchemaMatchingOfReusedFieldNameHonorsCaseSensitivity() {
+ Catalog catalog = CATALOG_EXTENSION.catalog();
+ TableIdentifier tableIdentifier = TableIdentifier.parse("default.myTable");
+ Table table = catalog.createTable(tableIdentifier, SCHEMA);
+ int renamedFieldId = table.schema().findField("data").fieldId();
+ table.updateSchema().renameColumn("data", "legacy").commit();
+ table.updateSchema().addColumn("Data", Types.StringType.get()).commit();
+ int currentFieldId = table.schema().findField("Data").fieldId();
+ assertThat(currentFieldId).isNotEqualTo(renamedFieldId);
+
+ TableMetadataCache caseInsensitive =
+ new TableMetadataCache(catalog, 10, Long.MAX_VALUE, 10,
CASE_INSENSITIVE, PRESERVE_COLUMNS);
+ Schema resolved = caseInsensitive.schema(tableIdentifier,
SCHEMA).resolvedTableSchema();
+
assertThat(resolved.caseInsensitiveFindField("data").fieldId()).isEqualTo(currentFieldId);
+
+ TableMetadataCache caseSensitive =
+ new TableMetadataCache(catalog, 10, Long.MAX_VALUE, 10,
CASE_SENSITIVE, PRESERVE_COLUMNS);
+ resolved = caseSensitive.schema(tableIdentifier,
SCHEMA).resolvedTableSchema();
+ assertThat(resolved.findField("data").fieldId()).isEqualTo(renamedFieldId);
+ }
}
