Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
kumarpritam863 wants to merge 5 commits into
base: main
Choose a base branch
from
+642 −18
Open

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
1a47461
Flink: Dynamic Sink: Create the namespace when the catalog reports it…
Oct 19, 2026
9a5407f
Flink: Dynamic Sink: Ask the catalog for the namespace instead of als…
Oct 19, 2026
bac0435
Flink: Dynamic Sink: Fail clearly when the catalog cannot create the …
Oct 19, 2026
27b36d2
Flink: Dynamic Sink: Fail on an unexpected lookup exception and rewor…
Oct 20, 2026
0f08bbe
Flink: Dynamic Sink: Always fail for a catalog without namespace support
Oct 20, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 29 additions & 6 deletions flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -78,14 +79,36 @@ private void findOrCreateTable(
TableIdentifier identifier, Schema schema, PartitionSpec spec, TableCreator tableCreator) {
Tuple2<Boolean, Exception> exists = cache.exists(identifier);
if (Boolean.FALSE.equals(exists.f0)) {
if (exists.f1 instanceof NoSuchNamespaceException) {
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;
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);
// 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());
} catch (AlreadyExistsException e) {
LOG.debug("Namespace {} created concurrently", identifier.namespace(), e);
}
}
} 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);
}

LOG.info("Table {} not found during table search. Creating table.", identifier);
Expand Down
185 changes: 185 additions & 0 deletions flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,10 +19,13 @@
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 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;
Expand All @@ -35,6 +38,8 @@
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;
import org.apache.iceberg.types.Types;
Expand Down Expand Up @@ -158,6 +163,148 @@ 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<Boolean, Exception> 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();
}

private static Catalog catalogWithoutNamespaceSupport(Supplier<RuntimeException> lookupFailure) {
return new Catalog() {
@Override
public List<TableIdentifier> 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 =
catalogWithoutNamespaceSupport(
() -> new NoSuchNamespaceException("Namespace does not exist"));
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 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();
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<Boolean, Exception> 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();
Expand Down Expand Up @@ -453,4 +600,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<TableMetadataCache.ResolvedSchemaInfo, PartitionSpec> 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();
}
}
35 changes: 29 additions & 6 deletions flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -78,14 +79,36 @@ private void findOrCreateTable(
TableIdentifier identifier, Schema schema, PartitionSpec spec, TableCreator tableCreator) {
Tuple2<Boolean, Exception> exists = cache.exists(identifier);
if (Boolean.FALSE.equals(exists.f0)) {
if (exists.f1 instanceof NoSuchNamespaceException) {
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;
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);
// 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());
} catch (AlreadyExistsException e) {
LOG.debug("Namespace {} created concurrently", identifier.namespace(), e);
}
}
} 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);
}

LOG.info("Table {} not found during table search. Creating table.", identifier);
Expand Down
Loading
Loading