Flink: Dynamic Sink: Create the namespace when the catalog reports it as a missing table by kumarpritam863 · Pull Request #18582 · apache/iceberg

kumarpritam863 · 2026-10-19T08:36:34Z

The Flink dynamic sink's TableUpdater.findOrCreateTable creates a missing namespace only when the table lookup throws NoSuchNamespaceException. The Glue catalog never does: GlueTableOperations.getGlueTable returns null on EntityNotFoundException, so a table in a database that does not exist looks like a table that does not exist yet and loadTable throws NoSuchTableException. The namespace is never created, and createTable then fails with

Cannot commit <catalog>.<db>.<table> because Glue cannot find the requested entity

(GlueTableOperations, EntityNotFoundException -> NotFoundException). A job that writes to a database that does not exist yet fails on every record and restarts. Seen on a cluster with a Flink dynamic sink writing to a new Glue database; the task manager log carried the message above.

Change

findOrCreateTable no longer relies on the lookup to report a missing namespace. Before creating a table it checks, with Preconditions.checkState, that

the lookup result holds no exception other than NoSuchTableException or NoSuchNamespaceException, and
the catalog is a SupportsNamespaces: a catalog that is not cannot be asked about, or made to create, the namespace of the table, so it fails with a message naming the catalog whatever the lookup reported.

Then, if the identifier has a namespace, it asks the catalog with namespaceExists and creates the namespace when it is missing, tolerating a concurrent creation (AlreadyExistsException). The lookup is not used for this because it cannot be: Glue reports the missing namespace as NoSuchTableException, and TableMetadataCache.exists returns the exception only when it has just looked the table up, not when it answers from the cache, which is what DynamicRecordProcessor hands to the updater (it asks the cache first). The old check only ran in the first case, so in the processor flow the namespace was not created even for a catalog that throws NoSuchNamespaceException.

The change and its tests are identical in the three Flink version directories: v1.20, v2.2, v2.3.

Testing

TestTableUpdater#testTableCreationWhenCatalogReportsMissingNamespaceAsNoSuchTable: an InMemoryCatalog whose loadTable throws NoSuchTableException for a missing namespace, as Glue does. It fails without the change and passes with it.
TestTableUpdater#testNamespaceIsCreatedWhenTheCacheAlreadyHoldsTheMissingTable: the flow of DynamicRecordProcessor: the cache is asked first, so the updater gets the cached answer with no exception, for a catalog that reports NoSuchNamespaceException. The namespace was not created before this change in that flow, and it is now.
TestTableUpdater#testMissingNamespaceOfCatalogWithoutNamespaceSupportFails and #testCatalogWithoutNamespaceSupportFailsWhateverTheLookupReported: a catalog without SupportsNamespaces fails with the message above, whether the lookup reported NoSuchNamespaceException, NoSuchTableException, or the answer came from the cache.
TestTableUpdater#testUnexpectedFailureOfTheTableLookupIsNotSwallowed: an unexpected exception in the lookup result fails the check.
All tests of org.apache.iceberg.flink.sink.dynamic pass in the three versions (297, 317, 320 tests); spotlessApply is clean.

mxm

Thanks for the PR @kumarpritam863! We usually implement changes for the newest Flink version first, then backport the changes. The changes are reasonable small, so we can skip this here.

mxm · 2026-10-19T12:30:12Z

+  private boolean namespaceIsMissing(TableIdentifier identifier) {
+    return catalog instanceof SupportsNamespaces
+        && !identifier.namespace().isEmpty()
+        && !((SupportsNamespaces) catalog).namespaceExists(identifier.namespace());


Wouldn't this cover all scenarios? Do we still need the old exists.f1 instanceof NoSuchNamespaceException check? In that case, can we inline the method?

Thanks @mxm for the review. Yes that one check is enough and i have replaced it with an inline function call. Also I will make sure from the next time that the changes are for latest version only.

mxm · 2026-10-19T15:04:10Z

+      // The lookup cannot tell a missing table from a missing namespace: the Glue catalog, for
+      // instance, reports a table in a namespace that does not exist as NoSuchTableException. Ask
+      // the catalog directly.
+      if (catalog instanceof SupportsNamespaces && !identifier.namespace().isEmpty()) {


if (exists.f1 instanceof NoSuchNamespaceException || exists.f1 instanceof NoSuchTableException) {

mxm · 2026-10-19T15:05:00Z

+      // The lookup cannot tell a missing table from a missing namespace: the Glue catalog, for
+      // instance, reports a table in a namespace that does not exist as NoSuchTableException. Ask
+      // the catalog directly.
+      if (catalog instanceof SupportsNamespaces && !identifier.namespace().isEmpty()) {


Do we need the isEmpty check?

I think it is needed as a table identifier without a namespace is valid (TableIdentifier.parse("myTable") in the tests), and without the check HadoopCatalog.namespaceExists(Namespace.empty()) throws IllegalArgumentException: Can not create a Path from an empty string; three existing tests fail. I added a comment.

mxm · 2026-10-19T15:08:40Z

+          } catch (AlreadyExistsException e) {
+            LOG.debug("Namespace {} created concurrently", identifier.namespace(), e);
+          }
        }


I think we must throw if the catalog doesn't implement SupportsNamespaces (else branch) or we have another type of exception.

For a catalog without SupportsNamespaces it now throws a clear UnsupportedOperationException when the lookup reported a missing namespace

mxm · 2026-10-20T07:36:48Z

+                "Cannot create the namespace of table %s: catalog %s does not support namespaces",
+                identifier, catalog.name()),
+            exists.f1);
      }


We should also throw in case of other exceptions in exists.f1.

done. A lookup result other than NoSuchTableException or NoSuchNamespaceException is now thrown as an IllegalStateException with the exception as the cause. Today TableMetadataCache.exists() can only return those two (anything else propagates out of refreshTable), so this is a guard for that contract, with a test that feeds it a different exception.

mxm · 2026-10-20T08:39:15Z

+      // The lookup cannot tell a missing table from a missing namespace: the Glue catalog, for
+      // instance, reports a table in a namespace that does not exist as NoSuchTableException. Ask
+      // the catalog directly.
+      if (catalog instanceof SupportsNamespaces) {


This comment is misleading because it only describes an edge case. The normal case would be a race condition with other tasks which try to create the namespace at the same time.

agreed, reworded. It now says that tasks arriving at the same time can all find the namespace missing and try to create it (the AlreadyExistsException is ignored), and that the catalog is asked instead of the lookup because catalogs differ in which exception they throw (Glue throws NoSuchTableException) and a cached answer carries none.

mxm · 2026-10-21T09:06:52Z

+      if (exists.f1 != null
+          && !(exists.f1 instanceof NoSuchTableException)
+          && !(exists.f1 instanceof NoSuchNamespaceException)) {
+        throw new IllegalStateException(
+            String.format("Unexpected failure looking up table %s", identifier), exists.f1);
+      }


Let's use Precondition.checkState instead.

The remaining check is now a Preconditions.checkState.

mxm · 2026-10-21T09:08:10Z

+      } else {
+        throw new UnsupportedOperationException(
+            String.format(
+                "Cannot create table %s: catalog %s does not support namespaces, so the namespace of the table cannot be created if it is missing",
+                identifier, catalog.name()),
+            exists.f1);


Let's move this into a Precondition.checkState(catalog instanceof SupportsNamespace, ...); check before line 93. We can then remove the if condition.

Done. The check is now the first statement in the Boolean.FALSE.equals(exists.f0) block, and the else branch is gone.

mxm · 2026-10-21T11:15:52Z

      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) {


We still need this check. There is no need to run the code below when exists.f1 == null .

DynamicRecordProcessor.processRecord calls tableCache.exists() before it calls the updater. That caches the "not found" answer.

When TableUpdater asks again, it gets the cached answer with f1 == null. Gating on f1 would skip namespace creation on the immediate-update path.

Glue also throws NoSuchTableException for a missing namespace, not NoSuchNamespaceException.
This block only runs when the table is missing, so it runs once per table creation. A single namespaceExists call is negligible there. The new testNamespaceCreatedWhenCatalogReportsMissingNamespaceAsMissingTable primes the cache with cache.exists(...) first, as the processor does. It fails on the old code.

That makes sense! Thanks.

… as a missing table TableUpdater created a missing namespace only when the table lookup threw NoSuchNamespaceException. Catalogs such as Glue report a table in a missing namespace as NoSuchTableException, and the processor's earlier lookup leaves a cached answer with no exception, so the namespace was never created and createTable failed on every record. Ask the catalog whether the namespace exists instead, and fail clearly when the catalog does not support namespaces.

mxm

Thanks @kumarpritam863!

mxm · 2026-11-04T15:41:23Z

-      if (exists.f1 instanceof NoSuchNamespaceException) {
-        SupportsNamespaces catalogWithNameSpace = (SupportsNamespaces) catalog;
+      Preconditions.checkState(
+          catalog instanceof SupportsNamespaces,


What if the catalog does not implement SupportsNamespaces? I don't think we should throw then.

Good point, that was a regression: before this PR such a catalog could still create tables, since the cast only happened inside the NoSuchNamespaceException branch. Removed the checkState. The namespace step now only runs for catalogs that implement SupportsNamespaces; others go straight to createTable.

mxm · 2026-11-04T15:42:20Z

      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) {


If we don't use f1 anymore, we should remove it.

Agreed, nothing reads it anymore. I removed it from TableMetadataCache.exists() altogether rather than just from the local variable, so it now returns a boolean. The EXISTS/NOT_EXISTS tuple constants went with it, and DynamicRecordProcessor no longer needs .f0. If you only meant the local variable, I can revert the cache part.

mxm · 2026-11-04T15:45:28Z

+
+  @Test
+  void testTableCreationFailsForCatalogWithoutNamespaceSupport() {
+    Catalog catalog = Mockito.mock(Catalog.class);


Could we remove the mock? CachingCatalog.wrap(inMemoryCatalog) is a catalog without SupportsNamespaces.

Done. The Mockito test is replaced by testTableCreatedForCatalogWithoutNamespaceSupport, which wraps an InMemoryCatalog in CachingCatalog and asserts the table gets created. It asserts the wrapper isn't a SupportsNamespaces, so it will flag it if that ever changes. The namespace is created on the inner catalog first, because InMemoryCatalog rejects tables in a missing namespace.

A catalog that does not implement SupportsNamespaces could create tables before the namespace fix, so failing for it is a regression. Skip the namespace step for such catalogs instead of throwing. TableMetadataCache.exists() no longer returns the lookup exception, as nothing reads it any more, so it returns a boolean.

github-actions Bot added the flink label Oct 19, 2026

mxm reviewed Oct 19, 2026

View reviewed changes

mxm reviewed Oct 20, 2026

View reviewed changes

mxm reviewed Oct 21, 2026

View reviewed changes

kumarpritam863 force-pushed the fix/flink-dynamic-sink-create-missing-namespace branch from 312234d to 180718f Compare November 1, 2026 15:33

kumarpritam863 force-pushed the fix/flink-dynamic-sink-create-missing-namespace branch from 180718f to 9bfdec2 Compare November 1, 2026 15:43

kumarpritam863 requested a review from mxm November 2, 2026 17:13

mxm reviewed Nov 4, 2026

View reviewed changes

Conversation

kumarpritam863 commented Oct 19, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Change

Testing

Uh oh!

mxm left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mxm Oct 21, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mxm Oct 21, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mxm left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

This branch has not been deployed

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

kumarpritam863 commented Oct 19, 2026 •

edited

Loading

mxm Oct 21, 2026 •

edited

Loading

mxm Oct 21, 2026 •

edited

Loading