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, if the catalog is a SupportsNamespaces and 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). If the catalog does not support namespaces it fails with a message naming the catalog, whatever the lookup reported (the old code threw a ClassCastException, and only when the lookup had just reported a missing namespace), and an exception in the lookup result other than NoSuchTableException and NoSuchNamespaceException is thrown instead of being ignored.

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 is thrown.
All tests of org.apache.iceberg.flink.sink.dynamic pass in the three versions (297, 317, 320 tests); spotlessApply is clean.

… as a missing table TableUpdater.findOrCreateTable creates a missing namespace only when the table lookup throws NoSuchNamespaceException. The Glue catalog does not: a lookup in a database that does not exist returns no Glue table, which loadTable reports as NoSuchTableException. The namespace is then never created, and creating the table fails with "Cannot commit <table> because Glue cannot find the requested entity", so a job writing to a new database restarts on every record. Also ask the catalog whether the namespace exists before creating the table, and create it when it does not. A namespace created concurrently is still tolerated.

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.

…o checking the lookup exception The namespaceExists check covers the NoSuchNamespaceException case too, so the exception check and the helper method are inlined into one condition.

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

…namespace, and test the cached lookup A catalog that does not support namespaces and reports a missing namespace now fails with an UnsupportedOperationException that names it, instead of a ClassCastException. The empty namespace is still not asked about, as the Hadoop catalog rejects it. Adds a test for the flow of DynamicRecordProcessor, where the cache already holds the missing table when the updater runs, and one for the catalog without namespace support.

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.

…d the namespace comment findOrCreateTable only expects NoSuchTableException or NoSuchNamespaceException from the table lookup. Any other exception in the lookup result is now thrown instead of being ignored. The comment on the namespace check now describes the usual case, tasks creating the namespace at the same time, and why the catalog is asked instead of the lookup.

A catalog that is not a SupportsNamespaces cannot be asked whether the namespace of a new table exists or be made to create it, so TableUpdater now fails with a message naming the catalog whatever the table lookup reported, instead of only when the lookup had just reported a missing namespace. The answer of TableMetadataCache.exists carries the exception only when it has just looked the table up, so the result no longer depends on whether the cache held the answer.

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

mxm reviewed Oct 19, 2026

View reviewed changes

Flink: Dynamic Sink: Ask the catalog for the namespace instead of als…

9a5407f

…o checking the lookup exception The namespaceExists check covers the NoSuchNamespaceException case too, so the exception check and the helper method are inlined into one condition.

mxm reviewed Oct 19, 2026

View reviewed changes

mxm reviewed Oct 20, 2026

View reviewed changes

Pritam Kumar Mishra added 2 commits October 20, 2026 17:52

Navigation Menu

Repository navigation

Flink: Dynamic Sink: Create the namespace when the catalog reports it as a missing table - #18582

Flink: Dynamic Sink: Create the namespace when the catalog reports it as a missing table#18582
kumarpritam863 wants to merge 5 commits into
apache:mainfrom
kumarpritam863:fix/flink-dynamic-sink-create-missing-namespace

kumarpritam863 commented Oct 19, 2026 •

edited

Loading

Uh oh!

mxm left a comment

Uh oh!

mxm Oct 19, 2026

Uh oh!

kumarpritam863 Oct 19, 2026

Uh oh!

mxm Oct 19, 2026

Uh oh!

mxm Oct 19, 2026

Uh oh!

kumarpritam863 Oct 19, 2026

Uh oh!

mxm Oct 19, 2026

Uh oh!

kumarpritam863 Oct 19, 2026

Uh oh!

mxm Oct 20, 2026

Uh oh!

kumarpritam863 Oct 20, 2026

Uh oh!

mxm Oct 20, 2026

Uh oh!

kumarpritam863 Oct 20, 2026

Uh oh!

This branch has not been deployed

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

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!

This branch has not been deployed

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

kumarpritam863 commented Oct 19, 2026 •

edited

Loading