This is an automated email from the ASF dual-hosted git repository.
mxm 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 dd17f677d8 Flink: Dynamic Sink: Create the namespace when the catalog
reports it as a missing table (#18582)
dd17f677d8 is described below
commit dd17f677d8ef39d6ad3085d90ce76f20274cc58b
Author: kumarpritam863 <[email protected]>
AuthorDate: Fri Nov 6 13:39:45 2026 +0530
Flink: Dynamic Sink: Create the namespace when the catalog reports it as a
missing table (#18582)
---
.../flink/sink/dynamic/DynamicRecordProcessor.java | 2 +-
.../flink/sink/dynamic/TableMetadataCache.java | 15 +++---
.../iceberg/flink/sink/dynamic/TableUpdater.java | 13 ++---
.../flink/sink/dynamic/TestTableUpdater.java | 61 ++++++++++++++++++++++
4 files changed, 75 insertions(+), 16 deletions(-)
diff --git
a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordProcessor.java
b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordProcessor.java
index 1f4b7cb1ca..e17fdbc65b 100644
---
a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordProcessor.java
+++
b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordProcessor.java
@@ -132,7 +132,7 @@ class DynamicRecordProcessor<T> extends ProcessFunction<T,
DynamicRecordInternal
DynamicRecordWithConfig data = dynamicRecordWithConfig.wrap(inputData);
boolean isForward = isForwardEligible(data);
- boolean exists = tableCache.exists(data.tableIdentifier()).f0;
+ boolean exists = tableCache.exists(data.tableIdentifier());
String foundBranch = exists ? tableCache.branch(data.tableIdentifier(),
data.branch()) : null;
TableMetadataCache.ResolvedSchemaInfo foundSchema =
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 9d836d56fb..99a75ec5b6 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
@@ -24,7 +24,6 @@ import java.util.Map;
import java.util.Set;
import org.apache.flink.annotation.Internal;
import org.apache.flink.annotation.VisibleForTesting;
-import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
import org.apache.iceberg.Table;
@@ -51,8 +50,6 @@ import org.slf4j.LoggerFactory;
class TableMetadataCache {
private static final Logger LOG =
LoggerFactory.getLogger(TableMetadataCache.class);
- private static final Tuple2<Boolean, Exception> EXISTS = Tuple2.of(true,
null);
- private static final Tuple2<Boolean, Exception> NOT_EXISTS =
Tuple2.of(false, null);
static final ResolvedSchemaInfo NOT_FOUND =
new ResolvedSchemaInfo(
null, CompareSchemasVisitor.Result.SCHEMA_UPDATE_NEEDED,
DataConverter.identity());
@@ -100,14 +97,14 @@ class TableMetadataCache {
this.cacheRefreshClock = cacheRefreshClock;
}
- Tuple2<Boolean, Exception> exists(TableIdentifier identifier) {
+ boolean exists(TableIdentifier identifier) {
CacheItem cached = tableCache.get(identifier);
if (cached != null && Boolean.TRUE.equals(cached.tableExists)) {
- return EXISTS;
+ return true;
} else if (needsRefresh(identifier, cached, true)) {
return refreshTable(identifier);
} else {
- return NOT_EXISTS;
+ return false;
}
}
@@ -241,16 +238,16 @@ class TableMetadataCache {
}
}
- private Tuple2<Boolean, Exception> refreshTable(TableIdentifier identifier) {
+ private boolean refreshTable(TableIdentifier identifier) {
try {
Table table = catalog.loadTable(identifier);
update(identifier, table);
- return EXISTS;
+ return true;
} catch (NoSuchTableException | NoSuchNamespaceException e) {
LOG.debug("Table or namespace doesn't exist {}", identifier, e);
tableCache.put(
identifier, new CacheItem(cacheRefreshClock.millis(), false, null,
null, null, 1));
- return Tuple2.of(false, e);
+ return false;
}
}
diff --git
a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java
b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java
index 25efa744a7..67efd2c223 100644
---
a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java
+++
b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java
@@ -31,7 +31,6 @@ import org.apache.iceberg.catalog.SupportsNamespaces;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.exceptions.AlreadyExistsException;
import org.apache.iceberg.exceptions.CommitFailedException;
-import org.apache.iceberg.exceptions.NoSuchNamespaceException;
import org.apache.iceberg.flink.FlinkSchemaUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -76,13 +75,15 @@ class TableUpdater {
private void findOrCreateTable(
TableIdentifier identifier, Schema schema, PartitionSpec spec,
TableCreator tableCreator) {
- Tuple2<Boolean, Exception> exists = cache.exists(identifier);
- if (Boolean.FALSE.equals(exists.f0)) {
- if (exists.f1 instanceof NoSuchNamespaceException) {
- SupportsNamespaces catalogWithNameSpace = (SupportsNamespaces) catalog;
+ if (!cache.exists(identifier)) {
+ // The catalog is asked rather than the lookup exception: catalogs
differ in what they throw
+ // for a missing namespace (Glue throws NoSuchTableException) and cached
lookups carry none.
+ if (catalog instanceof SupportsNamespaces catalogWithNamespaces
+ && !identifier.namespace().isEmpty()
+ && !catalogWithNamespaces.namespaceExists(identifier.namespace())) {
LOG.info("Namespace {} not found during table search. Creating
namespace", identifier);
try {
- catalogWithNameSpace.createNamespace(identifier.namespace());
+ catalogWithNamespaces.createNamespace(identifier.namespace());
} catch (AlreadyExistsException e) {
LOG.debug("Namespace {} created concurrently",
identifier.namespace(), e);
}
diff --git
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java
index 367250cf27..f8e74ea5fe 100644
---
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java
+++
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java
@@ -27,6 +27,7 @@ import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.StringData;
+import org.apache.iceberg.CachingCatalog;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
import org.apache.iceberg.SnapshotRef;
@@ -35,6 +36,7 @@ import org.apache.iceberg.catalog.Catalog;
import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.SupportsNamespaces;
import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.exceptions.NoSuchTableException;
import org.apache.iceberg.flink.sink.TestFlinkIcebergSinkBase;
import org.apache.iceberg.inmemory.InMemoryCatalog;
import org.apache.iceberg.types.Types;
@@ -453,4 +455,63 @@ public class TestTableUpdater extends
TestFlinkIcebergSinkBase {
assertThat(result.f0.resolvedTableSchema().sameSchema(SCHEMA)).isTrue();
assertThat(result.f0.compareResult()).isEqualTo(CompareSchemasVisitor.Result.SAME);
}
+
+ @Test
+ void testNamespaceCreatedWhenCatalogReportsMissingNamespaceAsMissingTable() {
+ // Like the Glue catalog, which throws NoSuchTableException for a table in
a missing namespace
+ InMemoryCatalog catalog =
+ new InMemoryCatalog() {
+ @Override
+ public Table loadTable(TableIdentifier identifier) {
+ if (!namespaceExists(identifier.namespace())) {
+ throw new NoSuchTableException("Table does not exist: %s",
identifier);
+ }
+
+ return super.loadTable(identifier);
+ }
+ };
+ catalog.initialize("catalog", Map.of());
+ Namespace namespace = Namespace.of("glue_like_namespace");
+ TableIdentifier tableIdentifier = TableIdentifier.of(namespace, "myTable");
+ TableMetadataCache cache =
+ new TableMetadataCache(catalog, 10, Long.MAX_VALUE, 10,
CASE_SENSITIVE, PRESERVE_COLUMNS);
+ TableUpdater tableUpdater = new TableUpdater(cache, catalog,
CASE_SENSITIVE, PRESERVE_COLUMNS);
+
+ // The record processor looks the table up before calling the updater, so
the updater only
+ // sees the cached answer, which carries no exception
+ cache.exists(tableIdentifier);
+
+ tableUpdater.update(
+ tableIdentifier,
+ SnapshotRef.MAIN_BRANCH,
+ SCHEMA,
+ PartitionSpec.unpartitioned(),
+ TableCreator.DEFAULT);
+
+ assertThat(catalog.namespaceExists(namespace)).isTrue();
+ assertThat(catalog.tableExists(tableIdentifier)).isTrue();
+ }
+
+ @Test
+ void testTableCreatedForCatalogWithoutNamespaceSupport() {
+ InMemoryCatalog inMemoryCatalog = new InMemoryCatalog();
+ inMemoryCatalog.initialize("catalog", Map.of());
+ Namespace namespace = Namespace.of("some_namespace");
+ inMemoryCatalog.createNamespace(namespace);
+ Catalog catalog = CachingCatalog.wrap(inMemoryCatalog);
+ assertThat(catalog).isNotInstanceOf(SupportsNamespaces.class);
+ TableIdentifier tableIdentifier = TableIdentifier.of(namespace, "myTable");
+ TableMetadataCache cache =
+ new TableMetadataCache(catalog, 10, Long.MAX_VALUE, 10,
CASE_SENSITIVE, PRESERVE_COLUMNS);
+ TableUpdater tableUpdater = new TableUpdater(cache, catalog,
CASE_SENSITIVE, PRESERVE_COLUMNS);
+
+ tableUpdater.update(
+ tableIdentifier,
+ SnapshotRef.MAIN_BRANCH,
+ SCHEMA,
+ PartitionSpec.unpartitioned(),
+ TableCreator.DEFAULT);
+
+ assertThat(catalog.tableExists(tableIdentifier)).isTrue();
+ }
}
