From 1a4746176c1c1301dc1659c44927b5cfa4747a71 Mon Sep 17 00:00:00 2001 From: Pritam Kumar Mishra Date: Mon, 19 Oct 2026 13:52:22 +0530 Subject: [PATCH 1/5] Flink: Dynamic Sink: Create the namespace when the catalog reports it 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 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. --- .../flink/sink/dynamic/TableUpdater.java | 14 ++++++- .../flink/sink/dynamic/TestTableUpdater.java | 39 +++++++++++++++++++ .../flink/sink/dynamic/TableUpdater.java | 14 ++++++- .../flink/sink/dynamic/TestTableUpdater.java | 39 +++++++++++++++++++ .../flink/sink/dynamic/TableUpdater.java | 14 ++++++- .../flink/sink/dynamic/TestTableUpdater.java | 39 +++++++++++++++++++ 6 files changed, 156 insertions(+), 3 deletions(-) diff --git a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java index 25efa744a733..d215ad6d67b5 100644 --- a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java +++ b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java @@ -78,7 +78,7 @@ private void findOrCreateTable( TableIdentifier identifier, Schema schema, PartitionSpec spec, TableCreator tableCreator) { Tuple2 exists = cache.exists(identifier); if (Boolean.FALSE.equals(exists.f0)) { - if (exists.f1 instanceof NoSuchNamespaceException) { + if (exists.f1 instanceof NoSuchNamespaceException || namespaceIsMissing(identifier)) { SupportsNamespaces catalogWithNameSpace = (SupportsNamespaces) catalog; LOG.info("Namespace {} not found during table search. Creating namespace", identifier); try { @@ -100,6 +100,18 @@ private void findOrCreateTable( } } + /** + * Some catalogs, such as the Glue catalog, report a table in a namespace that does not exist as a + * {@link org.apache.iceberg.exceptions.NoSuchTableException} rather than a {@link + * NoSuchNamespaceException}, so the lookup alone cannot tell a missing table from a missing + * namespace. Ask the catalog directly. + */ + private boolean namespaceIsMissing(TableIdentifier identifier) { + return catalog instanceof SupportsNamespaces + && !identifier.namespace().isEmpty() + && !((SupportsNamespaces) catalog).namespaceExists(identifier.namespace()); + } + private void findOrCreateBranch(TableIdentifier identifier, String branch) { String fromCache = cache.branch(identifier, branch); if (fromCache == null) { diff --git a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java index 367250cf278c..a1e9fcc2fdb7 100644 --- a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java +++ b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java @@ -35,6 +35,7 @@ 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 +454,42 @@ void testTableCreationWithExistingNamespace() { assertThat(result.f0.resolvedTableSchema().sameSchema(SCHEMA)).isTrue(); assertThat(result.f0.compareResult()).isEqualTo(CompareSchemasVisitor.Result.SAME); } + + @Test + void testTableCreationWhenCatalogReportsMissingNamespaceAsNoSuchTable() { + // The Glue catalog answers a lookup in a database that does not exist with + // NoSuchTableException, + // not NoSuchNamespaceException, and then refuses to create the table. + 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); + + assertThat(catalog.namespaceExists(namespace)).isFalse(); + + Tuple2 result = + tableUpdater.update( + tableIdentifier, + SnapshotRef.MAIN_BRANCH, + SCHEMA, + PartitionSpec.unpartitioned(), + TableCreator.DEFAULT); + + assertThat(catalog.namespaceExists(namespace)).isTrue(); + assertThat(catalog.tableExists(tableIdentifier)).isTrue(); + assertThat(result.f0.resolvedTableSchema().sameSchema(SCHEMA)).isTrue(); + } } diff --git a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java index 25efa744a733..d215ad6d67b5 100644 --- a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java +++ b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java @@ -78,7 +78,7 @@ private void findOrCreateTable( TableIdentifier identifier, Schema schema, PartitionSpec spec, TableCreator tableCreator) { Tuple2 exists = cache.exists(identifier); if (Boolean.FALSE.equals(exists.f0)) { - if (exists.f1 instanceof NoSuchNamespaceException) { + if (exists.f1 instanceof NoSuchNamespaceException || namespaceIsMissing(identifier)) { SupportsNamespaces catalogWithNameSpace = (SupportsNamespaces) catalog; LOG.info("Namespace {} not found during table search. Creating namespace", identifier); try { @@ -100,6 +100,18 @@ private void findOrCreateTable( } } + /** + * Some catalogs, such as the Glue catalog, report a table in a namespace that does not exist as a + * {@link org.apache.iceberg.exceptions.NoSuchTableException} rather than a {@link + * NoSuchNamespaceException}, so the lookup alone cannot tell a missing table from a missing + * namespace. Ask the catalog directly. + */ + private boolean namespaceIsMissing(TableIdentifier identifier) { + return catalog instanceof SupportsNamespaces + && !identifier.namespace().isEmpty() + && !((SupportsNamespaces) catalog).namespaceExists(identifier.namespace()); + } + private void findOrCreateBranch(TableIdentifier identifier, String branch) { String fromCache = cache.branch(identifier, branch); if (fromCache == null) { diff --git a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java index 367250cf278c..a1e9fcc2fdb7 100644 --- a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java +++ b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java @@ -35,6 +35,7 @@ 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 +454,42 @@ void testTableCreationWithExistingNamespace() { assertThat(result.f0.resolvedTableSchema().sameSchema(SCHEMA)).isTrue(); assertThat(result.f0.compareResult()).isEqualTo(CompareSchemasVisitor.Result.SAME); } + + @Test + void testTableCreationWhenCatalogReportsMissingNamespaceAsNoSuchTable() { + // The Glue catalog answers a lookup in a database that does not exist with + // NoSuchTableException, + // not NoSuchNamespaceException, and then refuses to create the table. + 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); + + assertThat(catalog.namespaceExists(namespace)).isFalse(); + + Tuple2 result = + tableUpdater.update( + tableIdentifier, + SnapshotRef.MAIN_BRANCH, + SCHEMA, + PartitionSpec.unpartitioned(), + TableCreator.DEFAULT); + + assertThat(catalog.namespaceExists(namespace)).isTrue(); + assertThat(catalog.tableExists(tableIdentifier)).isTrue(); + assertThat(result.f0.resolvedTableSchema().sameSchema(SCHEMA)).isTrue(); + } } 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 25efa744a733..d215ad6d67b5 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 @@ -78,7 +78,7 @@ private void findOrCreateTable( TableIdentifier identifier, Schema schema, PartitionSpec spec, TableCreator tableCreator) { Tuple2 exists = cache.exists(identifier); if (Boolean.FALSE.equals(exists.f0)) { - if (exists.f1 instanceof NoSuchNamespaceException) { + if (exists.f1 instanceof NoSuchNamespaceException || namespaceIsMissing(identifier)) { SupportsNamespaces catalogWithNameSpace = (SupportsNamespaces) catalog; LOG.info("Namespace {} not found during table search. Creating namespace", identifier); try { @@ -100,6 +100,18 @@ private void findOrCreateTable( } } + /** + * Some catalogs, such as the Glue catalog, report a table in a namespace that does not exist as a + * {@link org.apache.iceberg.exceptions.NoSuchTableException} rather than a {@link + * NoSuchNamespaceException}, so the lookup alone cannot tell a missing table from a missing + * namespace. Ask the catalog directly. + */ + private boolean namespaceIsMissing(TableIdentifier identifier) { + return catalog instanceof SupportsNamespaces + && !identifier.namespace().isEmpty() + && !((SupportsNamespaces) catalog).namespaceExists(identifier.namespace()); + } + private void findOrCreateBranch(TableIdentifier identifier, String branch) { String fromCache = cache.branch(identifier, branch); if (fromCache == null) { 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 367250cf278c..a1e9fcc2fdb7 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 @@ -35,6 +35,7 @@ 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 +454,42 @@ void testTableCreationWithExistingNamespace() { assertThat(result.f0.resolvedTableSchema().sameSchema(SCHEMA)).isTrue(); assertThat(result.f0.compareResult()).isEqualTo(CompareSchemasVisitor.Result.SAME); } + + @Test + void testTableCreationWhenCatalogReportsMissingNamespaceAsNoSuchTable() { + // The Glue catalog answers a lookup in a database that does not exist with + // NoSuchTableException, + // not NoSuchNamespaceException, and then refuses to create the table. + 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); + + assertThat(catalog.namespaceExists(namespace)).isFalse(); + + Tuple2 result = + tableUpdater.update( + tableIdentifier, + SnapshotRef.MAIN_BRANCH, + SCHEMA, + PartitionSpec.unpartitioned(), + TableCreator.DEFAULT); + + assertThat(catalog.namespaceExists(namespace)).isTrue(); + assertThat(catalog.tableExists(tableIdentifier)).isTrue(); + assertThat(result.f0.resolvedTableSchema().sameSchema(SCHEMA)).isTrue(); + } } From 9a5407fbb7ea67514b145c0470d50b8e07f6a8f8 Mon Sep 17 00:00:00 2001 From: Pritam Kumar Mishra Date: Mon, 19 Oct 2026 20:03:08 +0530 Subject: [PATCH 2/5] Flink: Dynamic Sink: Ask the catalog for the namespace instead of also 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. --- .../flink/sink/dynamic/TableUpdater.java | 30 +++++++------------ .../flink/sink/dynamic/TableUpdater.java | 30 +++++++------------ .../flink/sink/dynamic/TableUpdater.java | 30 +++++++------------ 3 files changed, 33 insertions(+), 57 deletions(-) diff --git a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java index d215ad6d67b5..e733848c0a83 100644 --- a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java +++ b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java @@ -31,7 +31,6 @@ 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; @@ -78,13 +77,18 @@ private void findOrCreateTable( TableIdentifier identifier, Schema schema, PartitionSpec spec, TableCreator tableCreator) { Tuple2 exists = cache.exists(identifier); if (Boolean.FALSE.equals(exists.f0)) { - if (exists.f1 instanceof NoSuchNamespaceException || namespaceIsMissing(identifier)) { + // 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()) { SupportsNamespaces catalogWithNameSpace = (SupportsNamespaces) catalog; - LOG.info("Namespace {} not found during table search. Creating namespace", identifier); - try { - catalogWithNameSpace.createNamespace(identifier.namespace()); - } catch (AlreadyExistsException e) { - LOG.debug("Namespace {} created concurrently", identifier.namespace(), e); + if (!catalogWithNameSpace.namespaceExists(identifier.namespace())) { + LOG.info("Namespace {} not found during table search. Creating namespace", identifier); + try { + catalogWithNameSpace.createNamespace(identifier.namespace()); + } catch (AlreadyExistsException e) { + LOG.debug("Namespace {} created concurrently", identifier.namespace(), e); + } } } @@ -100,18 +104,6 @@ private void findOrCreateTable( } } - /** - * Some catalogs, such as the Glue catalog, report a table in a namespace that does not exist as a - * {@link org.apache.iceberg.exceptions.NoSuchTableException} rather than a {@link - * NoSuchNamespaceException}, so the lookup alone cannot tell a missing table from a missing - * namespace. Ask the catalog directly. - */ - private boolean namespaceIsMissing(TableIdentifier identifier) { - return catalog instanceof SupportsNamespaces - && !identifier.namespace().isEmpty() - && !((SupportsNamespaces) catalog).namespaceExists(identifier.namespace()); - } - private void findOrCreateBranch(TableIdentifier identifier, String branch) { String fromCache = cache.branch(identifier, branch); if (fromCache == null) { diff --git a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java index d215ad6d67b5..e733848c0a83 100644 --- a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java +++ b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java @@ -31,7 +31,6 @@ 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; @@ -78,13 +77,18 @@ private void findOrCreateTable( TableIdentifier identifier, Schema schema, PartitionSpec spec, TableCreator tableCreator) { Tuple2 exists = cache.exists(identifier); if (Boolean.FALSE.equals(exists.f0)) { - if (exists.f1 instanceof NoSuchNamespaceException || namespaceIsMissing(identifier)) { + // 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()) { SupportsNamespaces catalogWithNameSpace = (SupportsNamespaces) catalog; - LOG.info("Namespace {} not found during table search. Creating namespace", identifier); - try { - catalogWithNameSpace.createNamespace(identifier.namespace()); - } catch (AlreadyExistsException e) { - LOG.debug("Namespace {} created concurrently", identifier.namespace(), e); + if (!catalogWithNameSpace.namespaceExists(identifier.namespace())) { + LOG.info("Namespace {} not found during table search. Creating namespace", identifier); + try { + catalogWithNameSpace.createNamespace(identifier.namespace()); + } catch (AlreadyExistsException e) { + LOG.debug("Namespace {} created concurrently", identifier.namespace(), e); + } } } @@ -100,18 +104,6 @@ private void findOrCreateTable( } } - /** - * Some catalogs, such as the Glue catalog, report a table in a namespace that does not exist as a - * {@link org.apache.iceberg.exceptions.NoSuchTableException} rather than a {@link - * NoSuchNamespaceException}, so the lookup alone cannot tell a missing table from a missing - * namespace. Ask the catalog directly. - */ - private boolean namespaceIsMissing(TableIdentifier identifier) { - return catalog instanceof SupportsNamespaces - && !identifier.namespace().isEmpty() - && !((SupportsNamespaces) catalog).namespaceExists(identifier.namespace()); - } - private void findOrCreateBranch(TableIdentifier identifier, String branch) { String fromCache = cache.branch(identifier, branch); if (fromCache == null) { 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 d215ad6d67b5..e733848c0a83 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.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; @@ -78,13 +77,18 @@ private void findOrCreateTable( TableIdentifier identifier, Schema schema, PartitionSpec spec, TableCreator tableCreator) { Tuple2 exists = cache.exists(identifier); if (Boolean.FALSE.equals(exists.f0)) { - if (exists.f1 instanceof NoSuchNamespaceException || namespaceIsMissing(identifier)) { + // 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()) { SupportsNamespaces catalogWithNameSpace = (SupportsNamespaces) catalog; - LOG.info("Namespace {} not found during table search. Creating namespace", identifier); - try { - catalogWithNameSpace.createNamespace(identifier.namespace()); - } catch (AlreadyExistsException e) { - LOG.debug("Namespace {} created concurrently", identifier.namespace(), e); + if (!catalogWithNameSpace.namespaceExists(identifier.namespace())) { + LOG.info("Namespace {} not found during table search. Creating namespace", identifier); + try { + catalogWithNameSpace.createNamespace(identifier.namespace()); + } catch (AlreadyExistsException e) { + LOG.debug("Namespace {} created concurrently", identifier.namespace(), e); + } } } @@ -100,18 +104,6 @@ private void findOrCreateTable( } } - /** - * Some catalogs, such as the Glue catalog, report a table in a namespace that does not exist as a - * {@link org.apache.iceberg.exceptions.NoSuchTableException} rather than a {@link - * NoSuchNamespaceException}, so the lookup alone cannot tell a missing table from a missing - * namespace. Ask the catalog directly. - */ - private boolean namespaceIsMissing(TableIdentifier identifier) { - return catalog instanceof SupportsNamespaces - && !identifier.namespace().isEmpty() - && !((SupportsNamespaces) catalog).namespaceExists(identifier.namespace()); - } - private void findOrCreateBranch(TableIdentifier identifier, String branch) { String fromCache = cache.branch(identifier, branch); if (fromCache == null) { From bac0435eeb5a6d7e1a6cc34e2da8549be79e5f77 Mon Sep 17 00:00:00 2001 From: Pritam Kumar Mishra Date: Mon, 19 Oct 2026 21:06:24 +0530 Subject: [PATCH 3/5] Flink: Dynamic Sink: Fail clearly when the catalog cannot create the 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. --- .../flink/sink/dynamic/TableUpdater.java | 14 ++- .../flink/sink/dynamic/TestTableUpdater.java | 88 +++++++++++++++++++ .../flink/sink/dynamic/TableUpdater.java | 14 ++- .../flink/sink/dynamic/TestTableUpdater.java | 88 +++++++++++++++++++ .../flink/sink/dynamic/TableUpdater.java | 14 ++- .../flink/sink/dynamic/TestTableUpdater.java | 88 +++++++++++++++++++ 6 files changed, 300 insertions(+), 6 deletions(-) diff --git a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java index e733848c0a83..9a9305ea3ecc 100644 --- a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java +++ b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java @@ -31,6 +31,7 @@ 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; @@ -80,9 +81,12 @@ private void findOrCreateTable( // 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 (catalog instanceof SupportsNamespaces) { SupportsNamespaces catalogWithNameSpace = (SupportsNamespaces) catalog; - if (!catalogWithNameSpace.namespaceExists(identifier.namespace())) { + // A table without a namespace has none to create, and some catalogs, like the Hadoop + // catalog, do not accept the empty namespace in the call. + if (!identifier.namespace().isEmpty() + && !catalogWithNameSpace.namespaceExists(identifier.namespace())) { LOG.info("Namespace {} not found during table search. Creating namespace", identifier); try { catalogWithNameSpace.createNamespace(identifier.namespace()); @@ -90,6 +94,12 @@ private void findOrCreateTable( LOG.debug("Namespace {} created concurrently", identifier.namespace(), e); } } + } else if (exists.f1 instanceof NoSuchNamespaceException) { + throw new UnsupportedOperationException( + String.format( + "Cannot create the namespace of table %s: catalog %s does not support namespaces", + identifier, catalog.name()), + exists.f1); } LOG.info("Table {} not found during table search. Creating table.", identifier); diff --git a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java index a1e9fcc2fdb7..32c2bec206fb 100644 --- a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java +++ b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java @@ -19,9 +19,11 @@ package org.apache.iceberg.flink.sink.dynamic; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; import java.nio.file.Path; import java.time.LocalDate; +import java.util.List; import java.util.Map; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.table.data.GenericRowData; @@ -35,6 +37,7 @@ import org.apache.iceberg.catalog.Namespace; import org.apache.iceberg.catalog.SupportsNamespaces; import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.exceptions.NoSuchNamespaceException; import org.apache.iceberg.exceptions.NoSuchTableException; import org.apache.iceberg.flink.sink.TestFlinkIcebergSinkBase; import org.apache.iceberg.inmemory.InMemoryCatalog; @@ -159,6 +162,91 @@ void testBranchCreationAndCaching() { assertThat(cache.getInternalCache()).contains(Map.entry(tableIdentifier, cacheItem)); } + @Test + void testNamespaceIsCreatedWhenTheCacheAlreadyHoldsTheMissingTable() { + // A catalog that reports a table in a missing namespace as NoSuchNamespaceException. + InMemoryCatalog catalog = + new InMemoryCatalog() { + @Override + public Table loadTable(TableIdentifier identifier) { + if (!namespaceExists(identifier.namespace())) { + throw new NoSuchNamespaceException( + "Namespace does not exist: %s", identifier.namespace()); + } + + return super.loadTable(identifier); + } + }; + catalog.initialize("catalog", Map.of()); + Namespace namespace = Namespace.of("cached_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 asks the cache whether the table exists before it calls the updater, so + // the + // updater is given the cached answer, which does not carry the exception of the lookup. + Tuple2 first = cache.exists(tableIdentifier); + assertThat(first.f0).isFalse(); + assertThat(first.f1).isInstanceOf(NoSuchNamespaceException.class); + assertThat(cache.exists(tableIdentifier).f1).isNull(); + + tableUpdater.update( + tableIdentifier, + SnapshotRef.MAIN_BRANCH, + SCHEMA, + PartitionSpec.unpartitioned(), + TableCreator.DEFAULT); + + assertThat(catalog.namespaceExists(namespace)).isTrue(); + assertThat(catalog.tableExists(tableIdentifier)).isTrue(); + } + + @Test + void testMissingNamespaceOfCatalogWithoutNamespaceSupportFails() { + Catalog catalog = + new Catalog() { + @Override + public List listTables(Namespace namespace) { + throw new UnsupportedOperationException(); + } + + @Override + public Table loadTable(TableIdentifier identifier) { + throw new NoSuchNamespaceException( + "Namespace does not exist: %s", identifier.namespace()); + } + + @Override + public boolean dropTable(TableIdentifier identifier, boolean purge) { + throw new UnsupportedOperationException(); + } + + @Override + public void renameTable(TableIdentifier from, TableIdentifier to) { + throw new UnsupportedOperationException(); + } + }; + TableIdentifier tableIdentifier = + TableIdentifier.of(Namespace.of("missing_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); + + assertThatThrownBy( + () -> + tableUpdater.update( + tableIdentifier, + SnapshotRef.MAIN_BRANCH, + SCHEMA, + PartitionSpec.unpartitioned(), + TableCreator.DEFAULT)) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessageContaining("does not support namespaces") + .hasCauseInstanceOf(NoSuchNamespaceException.class); + } + @Test void testSpecCreation() { Catalog catalog = CATALOG_EXTENSION.catalog(); diff --git a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java index e733848c0a83..9a9305ea3ecc 100644 --- a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java +++ b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java @@ -31,6 +31,7 @@ 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; @@ -80,9 +81,12 @@ private void findOrCreateTable( // 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 (catalog instanceof SupportsNamespaces) { SupportsNamespaces catalogWithNameSpace = (SupportsNamespaces) catalog; - if (!catalogWithNameSpace.namespaceExists(identifier.namespace())) { + // A table without a namespace has none to create, and some catalogs, like the Hadoop + // catalog, do not accept the empty namespace in the call. + if (!identifier.namespace().isEmpty() + && !catalogWithNameSpace.namespaceExists(identifier.namespace())) { LOG.info("Namespace {} not found during table search. Creating namespace", identifier); try { catalogWithNameSpace.createNamespace(identifier.namespace()); @@ -90,6 +94,12 @@ private void findOrCreateTable( LOG.debug("Namespace {} created concurrently", identifier.namespace(), e); } } + } else if (exists.f1 instanceof NoSuchNamespaceException) { + throw new UnsupportedOperationException( + String.format( + "Cannot create the namespace of table %s: catalog %s does not support namespaces", + identifier, catalog.name()), + exists.f1); } LOG.info("Table {} not found during table search. Creating table.", identifier); diff --git a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java index a1e9fcc2fdb7..32c2bec206fb 100644 --- a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java +++ b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java @@ -19,9 +19,11 @@ package org.apache.iceberg.flink.sink.dynamic; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; import java.nio.file.Path; import java.time.LocalDate; +import java.util.List; import java.util.Map; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.table.data.GenericRowData; @@ -35,6 +37,7 @@ import org.apache.iceberg.catalog.Namespace; import org.apache.iceberg.catalog.SupportsNamespaces; import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.exceptions.NoSuchNamespaceException; import org.apache.iceberg.exceptions.NoSuchTableException; import org.apache.iceberg.flink.sink.TestFlinkIcebergSinkBase; import org.apache.iceberg.inmemory.InMemoryCatalog; @@ -159,6 +162,91 @@ void testBranchCreationAndCaching() { assertThat(cache.getInternalCache()).contains(Map.entry(tableIdentifier, cacheItem)); } + @Test + void testNamespaceIsCreatedWhenTheCacheAlreadyHoldsTheMissingTable() { + // A catalog that reports a table in a missing namespace as NoSuchNamespaceException. + InMemoryCatalog catalog = + new InMemoryCatalog() { + @Override + public Table loadTable(TableIdentifier identifier) { + if (!namespaceExists(identifier.namespace())) { + throw new NoSuchNamespaceException( + "Namespace does not exist: %s", identifier.namespace()); + } + + return super.loadTable(identifier); + } + }; + catalog.initialize("catalog", Map.of()); + Namespace namespace = Namespace.of("cached_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 asks the cache whether the table exists before it calls the updater, so + // the + // updater is given the cached answer, which does not carry the exception of the lookup. + Tuple2 first = cache.exists(tableIdentifier); + assertThat(first.f0).isFalse(); + assertThat(first.f1).isInstanceOf(NoSuchNamespaceException.class); + assertThat(cache.exists(tableIdentifier).f1).isNull(); + + tableUpdater.update( + tableIdentifier, + SnapshotRef.MAIN_BRANCH, + SCHEMA, + PartitionSpec.unpartitioned(), + TableCreator.DEFAULT); + + assertThat(catalog.namespaceExists(namespace)).isTrue(); + assertThat(catalog.tableExists(tableIdentifier)).isTrue(); + } + + @Test + void testMissingNamespaceOfCatalogWithoutNamespaceSupportFails() { + Catalog catalog = + new Catalog() { + @Override + public List listTables(Namespace namespace) { + throw new UnsupportedOperationException(); + } + + @Override + public Table loadTable(TableIdentifier identifier) { + throw new NoSuchNamespaceException( + "Namespace does not exist: %s", identifier.namespace()); + } + + @Override + public boolean dropTable(TableIdentifier identifier, boolean purge) { + throw new UnsupportedOperationException(); + } + + @Override + public void renameTable(TableIdentifier from, TableIdentifier to) { + throw new UnsupportedOperationException(); + } + }; + TableIdentifier tableIdentifier = + TableIdentifier.of(Namespace.of("missing_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); + + assertThatThrownBy( + () -> + tableUpdater.update( + tableIdentifier, + SnapshotRef.MAIN_BRANCH, + SCHEMA, + PartitionSpec.unpartitioned(), + TableCreator.DEFAULT)) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessageContaining("does not support namespaces") + .hasCauseInstanceOf(NoSuchNamespaceException.class); + } + @Test void testSpecCreation() { Catalog catalog = CATALOG_EXTENSION.catalog(); 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 e733848c0a83..9a9305ea3ecc 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,6 +31,7 @@ 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; @@ -80,9 +81,12 @@ private void findOrCreateTable( // 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 (catalog instanceof SupportsNamespaces) { SupportsNamespaces catalogWithNameSpace = (SupportsNamespaces) catalog; - if (!catalogWithNameSpace.namespaceExists(identifier.namespace())) { + // A table without a namespace has none to create, and some catalogs, like the Hadoop + // catalog, do not accept the empty namespace in the call. + if (!identifier.namespace().isEmpty() + && !catalogWithNameSpace.namespaceExists(identifier.namespace())) { LOG.info("Namespace {} not found during table search. Creating namespace", identifier); try { catalogWithNameSpace.createNamespace(identifier.namespace()); @@ -90,6 +94,12 @@ private void findOrCreateTable( LOG.debug("Namespace {} created concurrently", identifier.namespace(), e); } } + } else if (exists.f1 instanceof NoSuchNamespaceException) { + throw new UnsupportedOperationException( + String.format( + "Cannot create the namespace of table %s: catalog %s does not support namespaces", + identifier, catalog.name()), + exists.f1); } LOG.info("Table {} not found during table search. Creating table.", identifier); 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 a1e9fcc2fdb7..32c2bec206fb 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 @@ -19,9 +19,11 @@ package org.apache.iceberg.flink.sink.dynamic; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; import java.nio.file.Path; import java.time.LocalDate; +import java.util.List; import java.util.Map; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.table.data.GenericRowData; @@ -35,6 +37,7 @@ import org.apache.iceberg.catalog.Namespace; import org.apache.iceberg.catalog.SupportsNamespaces; import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.exceptions.NoSuchNamespaceException; import org.apache.iceberg.exceptions.NoSuchTableException; import org.apache.iceberg.flink.sink.TestFlinkIcebergSinkBase; import org.apache.iceberg.inmemory.InMemoryCatalog; @@ -159,6 +162,91 @@ void testBranchCreationAndCaching() { assertThat(cache.getInternalCache()).contains(Map.entry(tableIdentifier, cacheItem)); } + @Test + void testNamespaceIsCreatedWhenTheCacheAlreadyHoldsTheMissingTable() { + // A catalog that reports a table in a missing namespace as NoSuchNamespaceException. + InMemoryCatalog catalog = + new InMemoryCatalog() { + @Override + public Table loadTable(TableIdentifier identifier) { + if (!namespaceExists(identifier.namespace())) { + throw new NoSuchNamespaceException( + "Namespace does not exist: %s", identifier.namespace()); + } + + return super.loadTable(identifier); + } + }; + catalog.initialize("catalog", Map.of()); + Namespace namespace = Namespace.of("cached_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 asks the cache whether the table exists before it calls the updater, so + // the + // updater is given the cached answer, which does not carry the exception of the lookup. + Tuple2 first = cache.exists(tableIdentifier); + assertThat(first.f0).isFalse(); + assertThat(first.f1).isInstanceOf(NoSuchNamespaceException.class); + assertThat(cache.exists(tableIdentifier).f1).isNull(); + + tableUpdater.update( + tableIdentifier, + SnapshotRef.MAIN_BRANCH, + SCHEMA, + PartitionSpec.unpartitioned(), + TableCreator.DEFAULT); + + assertThat(catalog.namespaceExists(namespace)).isTrue(); + assertThat(catalog.tableExists(tableIdentifier)).isTrue(); + } + + @Test + void testMissingNamespaceOfCatalogWithoutNamespaceSupportFails() { + Catalog catalog = + new Catalog() { + @Override + public List listTables(Namespace namespace) { + throw new UnsupportedOperationException(); + } + + @Override + public Table loadTable(TableIdentifier identifier) { + throw new NoSuchNamespaceException( + "Namespace does not exist: %s", identifier.namespace()); + } + + @Override + public boolean dropTable(TableIdentifier identifier, boolean purge) { + throw new UnsupportedOperationException(); + } + + @Override + public void renameTable(TableIdentifier from, TableIdentifier to) { + throw new UnsupportedOperationException(); + } + }; + TableIdentifier tableIdentifier = + TableIdentifier.of(Namespace.of("missing_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); + + assertThatThrownBy( + () -> + tableUpdater.update( + tableIdentifier, + SnapshotRef.MAIN_BRANCH, + SCHEMA, + PartitionSpec.unpartitioned(), + TableCreator.DEFAULT)) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessageContaining("does not support namespaces") + .hasCauseInstanceOf(NoSuchNamespaceException.class); + } + @Test void testSpecCreation() { Catalog catalog = CATALOG_EXTENSION.catalog(); From 27b36d21623b270208763a3a76bc6aaa4b5e6b90 Mon Sep 17 00:00:00 2001 From: Pritam Kumar Mishra Date: Tue, 20 Oct 2026 17:52:28 +0530 Subject: [PATCH 4/5] Flink: Dynamic Sink: Fail on an unexpected lookup exception and reword 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. --- .../flink/sink/dynamic/TableUpdater.java | 15 ++++++++-- .../flink/sink/dynamic/TestTableUpdater.java | 28 +++++++++++++++++++ .../flink/sink/dynamic/TableUpdater.java | 15 ++++++++-- .../flink/sink/dynamic/TestTableUpdater.java | 28 +++++++++++++++++++ .../flink/sink/dynamic/TableUpdater.java | 15 ++++++++-- .../flink/sink/dynamic/TestTableUpdater.java | 28 +++++++++++++++++++ 6 files changed, 120 insertions(+), 9 deletions(-) diff --git a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java index 9a9305ea3ecc..838d5bd42006 100644 --- a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java +++ b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java @@ -32,6 +32,7 @@ import org.apache.iceberg.exceptions.AlreadyExistsException; import org.apache.iceberg.exceptions.CommitFailedException; import org.apache.iceberg.exceptions.NoSuchNamespaceException; +import org.apache.iceberg.exceptions.NoSuchTableException; import org.apache.iceberg.flink.FlinkSchemaUtil; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -78,9 +79,17 @@ private void findOrCreateTable( TableIdentifier identifier, Schema schema, PartitionSpec spec, TableCreator tableCreator) { Tuple2 exists = cache.exists(identifier); if (Boolean.FALSE.equals(exists.f0)) { - // 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 (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); + } + + // Create the namespace if it is missing. Tasks that get here at the same time can all find it + // missing and try to create it, which is not an error. The lookup of the table is not used to + // tell: which exception it throws for a missing namespace is up to the catalog (Glue throws + // NoSuchTableException), and an answer from the cache carries none. if (catalog instanceof SupportsNamespaces) { SupportsNamespaces catalogWithNameSpace = (SupportsNamespaces) catalog; // A table without a namespace has none to create, and some catalogs, like the Hadoop diff --git a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java index 32c2bec206fb..f8ad73670812 100644 --- a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java +++ b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java @@ -247,6 +247,34 @@ public void renameTable(TableIdentifier from, TableIdentifier to) { .hasCauseInstanceOf(NoSuchNamespaceException.class); } + @Test + void testUnexpectedFailureOfTheTableLookupIsNotSwallowed() { + Catalog catalog = CATALOG_EXTENSION.catalog(); + TableIdentifier tableIdentifier = TableIdentifier.parse("myTable"); + RuntimeException failure = new RuntimeException("catalog unavailable"); + TableMetadataCache cache = + new TableMetadataCache(catalog, 10, Long.MAX_VALUE, 10, CASE_SENSITIVE, PRESERVE_COLUMNS) { + @Override + Tuple2 exists(TableIdentifier identifier) { + return Tuple2.of(false, failure); + } + }; + TableUpdater tableUpdater = new TableUpdater(cache, catalog, CASE_SENSITIVE, PRESERVE_COLUMNS); + + assertThatThrownBy( + () -> + tableUpdater.update( + tableIdentifier, + SnapshotRef.MAIN_BRANCH, + SCHEMA, + PartitionSpec.unpartitioned(), + TableCreator.DEFAULT)) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("Unexpected failure looking up table") + .hasCause(failure); + assertThat(catalog.tableExists(tableIdentifier)).isFalse(); + } + @Test void testSpecCreation() { Catalog catalog = CATALOG_EXTENSION.catalog(); diff --git a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java index 9a9305ea3ecc..838d5bd42006 100644 --- a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java +++ b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java @@ -32,6 +32,7 @@ import org.apache.iceberg.exceptions.AlreadyExistsException; import org.apache.iceberg.exceptions.CommitFailedException; import org.apache.iceberg.exceptions.NoSuchNamespaceException; +import org.apache.iceberg.exceptions.NoSuchTableException; import org.apache.iceberg.flink.FlinkSchemaUtil; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -78,9 +79,17 @@ private void findOrCreateTable( TableIdentifier identifier, Schema schema, PartitionSpec spec, TableCreator tableCreator) { Tuple2 exists = cache.exists(identifier); if (Boolean.FALSE.equals(exists.f0)) { - // 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 (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); + } + + // Create the namespace if it is missing. Tasks that get here at the same time can all find it + // missing and try to create it, which is not an error. The lookup of the table is not used to + // tell: which exception it throws for a missing namespace is up to the catalog (Glue throws + // NoSuchTableException), and an answer from the cache carries none. if (catalog instanceof SupportsNamespaces) { SupportsNamespaces catalogWithNameSpace = (SupportsNamespaces) catalog; // A table without a namespace has none to create, and some catalogs, like the Hadoop diff --git a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java index 32c2bec206fb..f8ad73670812 100644 --- a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java +++ b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java @@ -247,6 +247,34 @@ public void renameTable(TableIdentifier from, TableIdentifier to) { .hasCauseInstanceOf(NoSuchNamespaceException.class); } + @Test + void testUnexpectedFailureOfTheTableLookupIsNotSwallowed() { + Catalog catalog = CATALOG_EXTENSION.catalog(); + TableIdentifier tableIdentifier = TableIdentifier.parse("myTable"); + RuntimeException failure = new RuntimeException("catalog unavailable"); + TableMetadataCache cache = + new TableMetadataCache(catalog, 10, Long.MAX_VALUE, 10, CASE_SENSITIVE, PRESERVE_COLUMNS) { + @Override + Tuple2 exists(TableIdentifier identifier) { + return Tuple2.of(false, failure); + } + }; + TableUpdater tableUpdater = new TableUpdater(cache, catalog, CASE_SENSITIVE, PRESERVE_COLUMNS); + + assertThatThrownBy( + () -> + tableUpdater.update( + tableIdentifier, + SnapshotRef.MAIN_BRANCH, + SCHEMA, + PartitionSpec.unpartitioned(), + TableCreator.DEFAULT)) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("Unexpected failure looking up table") + .hasCause(failure); + assertThat(catalog.tableExists(tableIdentifier)).isFalse(); + } + @Test void testSpecCreation() { Catalog catalog = CATALOG_EXTENSION.catalog(); 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 9a9305ea3ecc..838d5bd42006 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 @@ -32,6 +32,7 @@ import org.apache.iceberg.exceptions.AlreadyExistsException; import org.apache.iceberg.exceptions.CommitFailedException; import org.apache.iceberg.exceptions.NoSuchNamespaceException; +import org.apache.iceberg.exceptions.NoSuchTableException; import org.apache.iceberg.flink.FlinkSchemaUtil; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -78,9 +79,17 @@ private void findOrCreateTable( TableIdentifier identifier, Schema schema, PartitionSpec spec, TableCreator tableCreator) { Tuple2 exists = cache.exists(identifier); if (Boolean.FALSE.equals(exists.f0)) { - // 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 (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); + } + + // Create the namespace if it is missing. Tasks that get here at the same time can all find it + // missing and try to create it, which is not an error. The lookup of the table is not used to + // tell: which exception it throws for a missing namespace is up to the catalog (Glue throws + // NoSuchTableException), and an answer from the cache carries none. if (catalog instanceof SupportsNamespaces) { SupportsNamespaces catalogWithNameSpace = (SupportsNamespaces) catalog; // A table without a namespace has none to create, and some catalogs, like the Hadoop 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 32c2bec206fb..f8ad73670812 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 @@ -247,6 +247,34 @@ public void renameTable(TableIdentifier from, TableIdentifier to) { .hasCauseInstanceOf(NoSuchNamespaceException.class); } + @Test + void testUnexpectedFailureOfTheTableLookupIsNotSwallowed() { + Catalog catalog = CATALOG_EXTENSION.catalog(); + TableIdentifier tableIdentifier = TableIdentifier.parse("myTable"); + RuntimeException failure = new RuntimeException("catalog unavailable"); + TableMetadataCache cache = + new TableMetadataCache(catalog, 10, Long.MAX_VALUE, 10, CASE_SENSITIVE, PRESERVE_COLUMNS) { + @Override + Tuple2 exists(TableIdentifier identifier) { + return Tuple2.of(false, failure); + } + }; + TableUpdater tableUpdater = new TableUpdater(cache, catalog, CASE_SENSITIVE, PRESERVE_COLUMNS); + + assertThatThrownBy( + () -> + tableUpdater.update( + tableIdentifier, + SnapshotRef.MAIN_BRANCH, + SCHEMA, + PartitionSpec.unpartitioned(), + TableCreator.DEFAULT)) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("Unexpected failure looking up table") + .hasCause(failure); + assertThat(catalog.tableExists(tableIdentifier)).isFalse(); + } + @Test void testSpecCreation() { Catalog catalog = CATALOG_EXTENSION.catalog(); From 0f08bbe17413283c2304c476d477ed8763080333 Mon Sep 17 00:00:00 2001 From: Pritam Kumar Mishra Date: Tue, 20 Oct 2026 18:21:57 +0530 Subject: [PATCH 5/5] Flink: Dynamic Sink: Always fail for a catalog without namespace support 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. --- .../flink/sink/dynamic/TableUpdater.java | 4 +- .../flink/sink/dynamic/TestTableUpdater.java | 74 +++++++++++++------ .../flink/sink/dynamic/TableUpdater.java | 4 +- .../flink/sink/dynamic/TestTableUpdater.java | 74 +++++++++++++------ .../flink/sink/dynamic/TableUpdater.java | 4 +- .../flink/sink/dynamic/TestTableUpdater.java | 74 +++++++++++++------ 6 files changed, 162 insertions(+), 72 deletions(-) diff --git a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java index 838d5bd42006..a2d2807485d5 100644 --- a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java +++ b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java @@ -103,10 +103,10 @@ private void findOrCreateTable( LOG.debug("Namespace {} created concurrently", identifier.namespace(), e); } } - } else if (exists.f1 instanceof NoSuchNamespaceException) { + } else { throw new UnsupportedOperationException( String.format( - "Cannot create the namespace of table %s: catalog %s does not support namespaces", + "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); } diff --git a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java index f8ad73670812..b8663e6e8ad1 100644 --- a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java +++ b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java @@ -25,6 +25,7 @@ import java.time.LocalDate; import java.util.List; import java.util.Map; +import java.util.function.Supplier; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.table.data.GenericRowData; import org.apache.flink.table.data.RowData; @@ -203,31 +204,35 @@ public Table loadTable(TableIdentifier identifier) { assertThat(catalog.tableExists(tableIdentifier)).isTrue(); } + private static Catalog catalogWithoutNamespaceSupport(Supplier lookupFailure) { + return new Catalog() { + @Override + public List listTables(Namespace namespace) { + throw new UnsupportedOperationException(); + } + + @Override + public Table loadTable(TableIdentifier identifier) { + throw lookupFailure.get(); + } + + @Override + public boolean dropTable(TableIdentifier identifier, boolean purge) { + throw new UnsupportedOperationException(); + } + + @Override + public void renameTable(TableIdentifier from, TableIdentifier to) { + throw new UnsupportedOperationException(); + } + }; + } + @Test void testMissingNamespaceOfCatalogWithoutNamespaceSupportFails() { Catalog catalog = - new Catalog() { - @Override - public List listTables(Namespace namespace) { - throw new UnsupportedOperationException(); - } - - @Override - public Table loadTable(TableIdentifier identifier) { - throw new NoSuchNamespaceException( - "Namespace does not exist: %s", identifier.namespace()); - } - - @Override - public boolean dropTable(TableIdentifier identifier, boolean purge) { - throw new UnsupportedOperationException(); - } - - @Override - public void renameTable(TableIdentifier from, TableIdentifier to) { - throw new UnsupportedOperationException(); - } - }; + catalogWithoutNamespaceSupport( + () -> new NoSuchNamespaceException("Namespace does not exist")); TableIdentifier tableIdentifier = TableIdentifier.of(Namespace.of("missing_namespace"), "myTable"); TableMetadataCache cache = @@ -247,6 +252,31 @@ public void renameTable(TableIdentifier from, TableIdentifier to) { .hasCauseInstanceOf(NoSuchNamespaceException.class); } + @Test + void testCatalogWithoutNamespaceSupportFailsWhateverTheLookupReported() { + Catalog catalog = + catalogWithoutNamespaceSupport(() -> new NoSuchTableException("Table does not exist")); + TableIdentifier tableIdentifier = TableIdentifier.of(Namespace.of("some_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 first answer carries the exception of the lookup, the second one comes from the cache and + // carries none: both must fail the same way. + assertThat(cache.exists(tableIdentifier).f1).isInstanceOf(NoSuchTableException.class); + assertThat(cache.exists(tableIdentifier).f1).isNull(); + assertThatThrownBy( + () -> + tableUpdater.update( + tableIdentifier, + SnapshotRef.MAIN_BRANCH, + SCHEMA, + PartitionSpec.unpartitioned(), + TableCreator.DEFAULT)) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessageContaining("does not support namespaces"); + } + @Test void testUnexpectedFailureOfTheTableLookupIsNotSwallowed() { Catalog catalog = CATALOG_EXTENSION.catalog(); diff --git a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java index 838d5bd42006..a2d2807485d5 100644 --- a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java +++ b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java @@ -103,10 +103,10 @@ private void findOrCreateTable( LOG.debug("Namespace {} created concurrently", identifier.namespace(), e); } } - } else if (exists.f1 instanceof NoSuchNamespaceException) { + } else { throw new UnsupportedOperationException( String.format( - "Cannot create the namespace of table %s: catalog %s does not support namespaces", + "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); } diff --git a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java index f8ad73670812..b8663e6e8ad1 100644 --- a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java +++ b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java @@ -25,6 +25,7 @@ import java.time.LocalDate; import java.util.List; import java.util.Map; +import java.util.function.Supplier; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.table.data.GenericRowData; import org.apache.flink.table.data.RowData; @@ -203,31 +204,35 @@ public Table loadTable(TableIdentifier identifier) { assertThat(catalog.tableExists(tableIdentifier)).isTrue(); } + private static Catalog catalogWithoutNamespaceSupport(Supplier lookupFailure) { + return new Catalog() { + @Override + public List listTables(Namespace namespace) { + throw new UnsupportedOperationException(); + } + + @Override + public Table loadTable(TableIdentifier identifier) { + throw lookupFailure.get(); + } + + @Override + public boolean dropTable(TableIdentifier identifier, boolean purge) { + throw new UnsupportedOperationException(); + } + + @Override + public void renameTable(TableIdentifier from, TableIdentifier to) { + throw new UnsupportedOperationException(); + } + }; + } + @Test void testMissingNamespaceOfCatalogWithoutNamespaceSupportFails() { Catalog catalog = - new Catalog() { - @Override - public List listTables(Namespace namespace) { - throw new UnsupportedOperationException(); - } - - @Override - public Table loadTable(TableIdentifier identifier) { - throw new NoSuchNamespaceException( - "Namespace does not exist: %s", identifier.namespace()); - } - - @Override - public boolean dropTable(TableIdentifier identifier, boolean purge) { - throw new UnsupportedOperationException(); - } - - @Override - public void renameTable(TableIdentifier from, TableIdentifier to) { - throw new UnsupportedOperationException(); - } - }; + catalogWithoutNamespaceSupport( + () -> new NoSuchNamespaceException("Namespace does not exist")); TableIdentifier tableIdentifier = TableIdentifier.of(Namespace.of("missing_namespace"), "myTable"); TableMetadataCache cache = @@ -247,6 +252,31 @@ public void renameTable(TableIdentifier from, TableIdentifier to) { .hasCauseInstanceOf(NoSuchNamespaceException.class); } + @Test + void testCatalogWithoutNamespaceSupportFailsWhateverTheLookupReported() { + Catalog catalog = + catalogWithoutNamespaceSupport(() -> new NoSuchTableException("Table does not exist")); + TableIdentifier tableIdentifier = TableIdentifier.of(Namespace.of("some_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 first answer carries the exception of the lookup, the second one comes from the cache and + // carries none: both must fail the same way. + assertThat(cache.exists(tableIdentifier).f1).isInstanceOf(NoSuchTableException.class); + assertThat(cache.exists(tableIdentifier).f1).isNull(); + assertThatThrownBy( + () -> + tableUpdater.update( + tableIdentifier, + SnapshotRef.MAIN_BRANCH, + SCHEMA, + PartitionSpec.unpartitioned(), + TableCreator.DEFAULT)) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessageContaining("does not support namespaces"); + } + @Test void testUnexpectedFailureOfTheTableLookupIsNotSwallowed() { Catalog catalog = CATALOG_EXTENSION.catalog(); 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 838d5bd42006..a2d2807485d5 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 @@ -103,10 +103,10 @@ private void findOrCreateTable( LOG.debug("Namespace {} created concurrently", identifier.namespace(), e); } } - } else if (exists.f1 instanceof NoSuchNamespaceException) { + } else { throw new UnsupportedOperationException( String.format( - "Cannot create the namespace of table %s: catalog %s does not support namespaces", + "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); } 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 f8ad73670812..b8663e6e8ad1 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 @@ -25,6 +25,7 @@ import java.time.LocalDate; import java.util.List; import java.util.Map; +import java.util.function.Supplier; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.table.data.GenericRowData; import org.apache.flink.table.data.RowData; @@ -203,31 +204,35 @@ public Table loadTable(TableIdentifier identifier) { assertThat(catalog.tableExists(tableIdentifier)).isTrue(); } + private static Catalog catalogWithoutNamespaceSupport(Supplier lookupFailure) { + return new Catalog() { + @Override + public List listTables(Namespace namespace) { + throw new UnsupportedOperationException(); + } + + @Override + public Table loadTable(TableIdentifier identifier) { + throw lookupFailure.get(); + } + + @Override + public boolean dropTable(TableIdentifier identifier, boolean purge) { + throw new UnsupportedOperationException(); + } + + @Override + public void renameTable(TableIdentifier from, TableIdentifier to) { + throw new UnsupportedOperationException(); + } + }; + } + @Test void testMissingNamespaceOfCatalogWithoutNamespaceSupportFails() { Catalog catalog = - new Catalog() { - @Override - public List listTables(Namespace namespace) { - throw new UnsupportedOperationException(); - } - - @Override - public Table loadTable(TableIdentifier identifier) { - throw new NoSuchNamespaceException( - "Namespace does not exist: %s", identifier.namespace()); - } - - @Override - public boolean dropTable(TableIdentifier identifier, boolean purge) { - throw new UnsupportedOperationException(); - } - - @Override - public void renameTable(TableIdentifier from, TableIdentifier to) { - throw new UnsupportedOperationException(); - } - }; + catalogWithoutNamespaceSupport( + () -> new NoSuchNamespaceException("Namespace does not exist")); TableIdentifier tableIdentifier = TableIdentifier.of(Namespace.of("missing_namespace"), "myTable"); TableMetadataCache cache = @@ -247,6 +252,31 @@ public void renameTable(TableIdentifier from, TableIdentifier to) { .hasCauseInstanceOf(NoSuchNamespaceException.class); } + @Test + void testCatalogWithoutNamespaceSupportFailsWhateverTheLookupReported() { + Catalog catalog = + catalogWithoutNamespaceSupport(() -> new NoSuchTableException("Table does not exist")); + TableIdentifier tableIdentifier = TableIdentifier.of(Namespace.of("some_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 first answer carries the exception of the lookup, the second one comes from the cache and + // carries none: both must fail the same way. + assertThat(cache.exists(tableIdentifier).f1).isInstanceOf(NoSuchTableException.class); + assertThat(cache.exists(tableIdentifier).f1).isNull(); + assertThatThrownBy( + () -> + tableUpdater.update( + tableIdentifier, + SnapshotRef.MAIN_BRANCH, + SCHEMA, + PartitionSpec.unpartitioned(), + TableCreator.DEFAULT)) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessageContaining("does not support namespaces"); + } + @Test void testUnexpectedFailureOfTheTableLookupIsNotSwallowed() { Catalog catalog = CATALOG_EXTENSION.catalog();