You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by pv...@apache.org on 2026/10/20 12:10:11 UTC
(iceberg) branch main updated: Flink: Refresh unresolved input schemas in dynamic sink metadata cache (#18541)
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 42f5d46ce6 Flink: Refresh unresolved input schemas in dynamic sink metadata cache (#18541)
42f5d46ce6 is described below
commit 42f5d46ce6fe68643e9d65b9011b788187866b35
Author: Sergei Nikolaev <ki...@gmail.com>
AuthorDate: Tue Oct 20 14:10:04 2026 +0200
Flink: Refresh unresolved input schemas in dynamic sink metadata cache (#18541)
---
.../flink/sink/dynamic/TableMetadataCache.java | 77 ++++++++++--------
.../sink/dynamic/TestDynamicRecordProcessor.java | 95 ++++++++++++++++++++++
.../flink/sink/dynamic/TestTableMetadataCache.java | 22 +++++
3 files changed, 159 insertions(+), 35 deletions(-)
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 dd8f73c166..76f5909e65 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
@@ -145,56 +145,63 @@ class TableMetadataCache {
private ResolvedSchemaInfo schema(
TableIdentifier identifier, Schema input, boolean allowRefresh) {
CacheItem cached = tableCache.get(identifier);
- Schema compatible = null;
if (cached != null && cached.tableExists) {
// This only works if the {@link Schema#equals(Object)} returns true for the old schema
// and a new schema. Performance is paramount as this code is on the hot path. Every other
// way for comparing 2 schemas were performing worse than the
// {@link CompareByNameVisitor#visit(Schema, Schema, boolean)}, so caching was useless.
ResolvedSchemaInfo lastResult = cached.inputSchemas.get(input);
- if (lastResult != null) {
- return lastResult;
- }
-
- for (Map.Entry<Integer, Schema> tableSchema : cached.tableSchemas.entrySet()) {
- CompareSchemasVisitor.Result result =
- CompareSchemasVisitor.visit(
- input, tableSchema.getValue(), caseSensitive, dropUnusedColumns);
- if (result == CompareSchemasVisitor.Result.SAME) {
- ResolvedSchemaInfo newResult =
- new ResolvedSchemaInfo(
- tableSchema.getValue(),
- CompareSchemasVisitor.Result.SAME,
- DataConverter.identity());
- cached.inputSchemas.put(input, newResult);
- return newResult;
- } else if (compatible == null
- && result == CompareSchemasVisitor.Result.DATA_CONVERSION_NEEDED) {
- compatible = tableSchema.getValue();
+ if (lastResult == null) {
+ ResolvedSchemaInfo newResult = resolveSchema(identifier, cached, input, allowRefresh);
+ if (newResult == null) {
+ refreshTable(identifier);
+ return schema(identifier, input, false);
}
+
+ cached.inputSchemas.put(input, newResult);
+ return newResult;
+ } else if (lastResult != NOT_FOUND) {
+ return lastResult;
}
}
if (needsRefresh(identifier, cached, allowRefresh)) {
refreshTable(identifier);
return schema(identifier, input, false);
- } else if (compatible != null) {
- ResolvedSchemaInfo newResult =
- new ResolvedSchemaInfo(
- compatible,
- CompareSchemasVisitor.Result.DATA_CONVERSION_NEEDED,
- DataConverter.get(
- FlinkSchemaUtil.convert(input),
- FlinkSchemaUtil.convert(compatible),
- caseSensitive));
- cached.inputSchemas.put(input, newResult);
- return newResult;
- } else if (cached != null && cached.tableExists) {
- cached.inputSchemas.put(input, NOT_FOUND);
- return NOT_FOUND;
- } else {
+ }
+
+ return NOT_FOUND;
+ }
+
+ /** Returns null if the table metadata must be refreshed before resolving the input schema. */
+ 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;
+ }
+ }
+
+ if (needsRefresh(identifier, cached, allowRefresh)) {
+ return null;
+ }
+
+ if (compatible == null) {
return NOT_FOUND;
}
+
+ return new ResolvedSchemaInfo(
+ compatible,
+ CompareSchemasVisitor.Result.DATA_CONVERSION_NEEDED,
+ DataConverter.get(
+ FlinkSchemaUtil.convert(input), FlinkSchemaUtil.convert(compatible), caseSensitive));
}
private PartitionSpec spec(TableIdentifier identifier, PartitionSpec spec, boolean allowRefresh) {
diff --git a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicRecordProcessor.java b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicRecordProcessor.java
index 298a338ea9..48b94c14c7 100644
--- a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicRecordProcessor.java
+++ b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicRecordProcessor.java
@@ -18,23 +18,38 @@
*/
package org.apache.iceberg.flink.sink.dynamic;
+import static org.apache.iceberg.TestHelpers.waitUntilAfter;
+import static org.apache.iceberg.flink.TestFixtures.DATABASE;
+import static org.apache.iceberg.flink.TestFixtures.TABLE;
import static org.assertj.core.api.Assertions.assertThat;
import java.util.Collections;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
+import org.apache.flink.streaming.util.ProcessFunctionTestHarnesses;
import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.StringData;
+import org.apache.flink.util.OutputTag;
import org.apache.iceberg.DistributionMode;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
import org.apache.iceberg.SnapshotRef;
+import org.apache.iceberg.Table;
import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.flink.HadoopCatalogExtension;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.relocated.com.google.common.collect.Sets;
import org.apache.iceberg.types.Types;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
class TestDynamicRecordProcessor {
+ @RegisterExtension
+ private static final HadoopCatalogExtension CATALOG_EXTENSION =
+ new HadoopCatalogExtension(DATABASE, TABLE);
+
private static final Schema SCHEMA =
new Schema(
Types.NestedField.required(1, "id", Types.IntegerType.get()),
@@ -49,6 +64,7 @@ class TestDynamicRecordProcessor {
private static final TableIdentifier TABLE_IDENTIFIER = TableIdentifier.of("default", "table");
private static final String BRANCH = SnapshotRef.MAIN_BRANCH;
+ private static final int CACHE_REFRESH_MS = 10;
@Test
void testForwardEligibleWhenNoEqualityFields() {
@@ -87,6 +103,85 @@ class TestDynamicRecordProcessor {
assertThat(DynamicRecordProcessor.isForwardEligible(record)).isFalse();
}
+ @Test
+ void routesRecordsToMainOutputAfterTableCreationAndCacheRefresh() throws Exception {
+ DynamicRecord record = recordWithDistributionMode(SCHEMA, DistributionMode.NONE);
+
+ OutputTag<DynamicRecordInternal> updateStream = updateStream();
+ try (OneInputStreamOperatorTestHarness<DynamicRecord, DynamicRecordInternal> harness =
+ ProcessFunctionTestHarnesses.forProcessFunction(processorWithoutImmediateUpdate())) {
+ harness.processElement(record, 0L);
+ assertThat(harness.getSideOutput(updateStream)).hasSize(1);
+ assertThat(harness.extractOutputValues()).isEmpty();
+
+ Table table = CATALOG_EXTENSION.catalog().createTable(TABLE_IDENTIFIER, SCHEMA);
+ table.manageSnapshots().createBranch(BRANCH).commit();
+ waitUntilAfter(System.currentTimeMillis() + CACHE_REFRESH_MS);
+
+ harness.processElement(record, 1L);
+ assertThat(harness.getSideOutput(updateStream)).hasSize(1);
+ assertThat(harness.extractOutputValues()).hasSize(1);
+ }
+ }
+
+ @Test
+ void routesRecordsToMainOutputAfterSchemaUpdateAndCacheRefresh() throws Exception {
+ Table table = CATALOG_EXTENSION.catalog().createTable(TABLE_IDENTIFIER, SCHEMA);
+ table.newAppend().commit();
+ Schema schemaWithNewColumn =
+ new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.required(2, "data", Types.StringType.get()),
+ Types.NestedField.optional(3, "extra", Types.StringType.get()));
+ DynamicRecord record =
+ new DynamicRecord(
+ TABLE_IDENTIFIER,
+ BRANCH,
+ schemaWithNewColumn,
+ GenericRowData.of(1, StringData.fromString("foo"), StringData.fromString("bar")),
+ PartitionSpec.unpartitioned(),
+ DistributionMode.NONE,
+ 2);
+
+ OutputTag<DynamicRecordInternal> updateStream = updateStream();
+ try (OneInputStreamOperatorTestHarness<DynamicRecord, DynamicRecordInternal> harness =
+ ProcessFunctionTestHarnesses.forProcessFunction(processorWithoutImmediateUpdate())) {
+ harness.processElement(record, 0L);
+ assertThat(harness.getSideOutput(updateStream)).hasSize(1);
+ assertThat(harness.extractOutputValues()).isEmpty();
+
+ table.updateSchema().addColumn("extra", Types.StringType.get()).commit();
+ waitUntilAfter(System.currentTimeMillis() + CACHE_REFRESH_MS);
+
+ harness.processElement(record, 1L);
+ assertThat(harness.getSideOutput(updateStream)).hasSize(1);
+ assertThat(harness.extractOutputValues()).hasSize(1);
+ }
+ }
+
+ private static DynamicRecordProcessor<DynamicRecord> processorWithoutImmediateUpdate() {
+ return new DynamicRecordProcessor<>(
+ (input, out) -> out.collect(input),
+ CATALOG_EXTENSION.catalogLoader(),
+ TableCreator.DEFAULT,
+ new FlinkDynamicSinkConf(
+ ImmutableMap.of(
+ FlinkDynamicSinkOptions.IMMEDIATE_TABLE_UPDATE.key(),
+ "false",
+ FlinkDynamicSinkOptions.CACHE_REFRESH_MS.key(),
+ String.valueOf(CACHE_REFRESH_MS)),
+ new Configuration()),
+ Collections.emptyMap(),
+ new Configuration());
+ }
+
+ private static OutputTag<DynamicRecordInternal> updateStream() {
+ return new OutputTag<>(
+ DynamicRecordProcessor.DYNAMIC_TABLE_UPDATE_STREAM,
+ new DynamicRecordInternalType(
+ CATALOG_EXTENSION.catalogLoader(), true, CACHE_REFRESH_MS)) {};
+ }
+
private static DynamicRecord recordWithDistributionMode(
Schema schema, DistributionMode distributionMode) {
return new DynamicRecord(
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 8a17c707f8..bdf8045774 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
@@ -18,6 +18,7 @@
*/
package org.apache.iceberg.flink.sink.dynamic;
+import static org.apache.iceberg.TestHelpers.waitUntilAfter;
import static org.assertj.core.api.Assertions.assertThat;
import java.time.Clock;
@@ -156,6 +157,27 @@ public class TestTableMetadataCache extends TestFlinkIcebergSinkBase {
assertThat(cacheItem.inputSchemas()).containsKeys(SCHEMA, SCHEMA2);
}
+ @Test
+ void refreshesUnresolvedSchemaAfterRefreshInterval() {
+ Catalog catalog = CATALOG_EXTENSION.catalog();
+ TableIdentifier tableIdentifier = TableIdentifier.parse("default.myTable");
+ Table table = catalog.createTable(tableIdentifier, SCHEMA);
+ table.newAppend().commit();
+
+ long refreshMs = 10L;
+ TableMetadataCache cache =
+ new TableMetadataCache(catalog, 10, refreshMs, 10, CASE_SENSITIVE, PRESERVE_COLUMNS);
+ cache.update(tableIdentifier, table);
+ assertThat(cache.schema(tableIdentifier, SCHEMA2).compareResult())
+ .isEqualTo(CompareSchemasVisitor.Result.SCHEMA_UPDATE_NEEDED);
+
+ table.updateSchema().addColumn("extra", Types.StringType.get()).commit();
+ waitUntilAfter(System.currentTimeMillis() + refreshMs);
+
+ assertThat(cache.schema(tableIdentifier, SCHEMA2).compareResult())
+ .isEqualTo(CompareSchemasVisitor.Result.SAME);
+ }
+
@Test
void testNoSuchNamespaceExceptionHandling() {
Catalog catalog = CATALOG_EXTENSION.catalog();