You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "kumarpritam863 (via GitHub)" <gi...@apache.org> on 2026/10/19 08:36:38 UTC
[PR] Flink: Dynamic Sink: Create the namespace when the catalog reports it as a missing table [iceberg]
kumarpritam863 opened a new pull request, #18582:
URL: https://github.com/apache/iceberg/pull/18582
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` creates the namespace when the lookup reported `NoSuchNamespaceException` (as before) or when `namespaceIsMissing(identifier)` is true: the catalog is a `SupportsNamespaces`, the identifier has a namespace, and `namespaceExists` is false. The extra call happens only on the path that is about to create a table. A namespace created concurrently is still tolerated.
The change and its test are identical in the three Flink version directories: `v1.20`, `v2.2`, `v2.3`.
### Testing
- New `TestTableUpdater#testTableCreationWhenCatalogReportsMissingNamespaceAsNoSuchTable`: an `InMemoryCatalog` whose `loadTable` throws `NoSuchTableException` for a missing namespace, as Glue does. It fails without the change (`NoSuchNamespaceException: Cannot create table glue_like_namespace.myTable. Namespace does not exist`) and passes with it.
- `TestTableUpdater` passes in all three versions (15 tests each).
- `spotlessApply` is clean.
Not run against a real Glue catalog; the behaviour of `getGlueTable` and `GlueTableOperations` is taken from the source.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Create the namespace when the catalog reports it as a missing table [iceberg]
Posted by "mxm (via GitHub) via issues" <is...@iceberg.apache.org>.
mxm commented on PR #18582:
URL: https://github.com/apache/iceberg/pull/18582#issuecomment-6680303113
I usually wait a day to see if there are any other comments. That said, the PR looks great.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Create the namespace when the catalog reports it as a missing table [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #18582:
URL: https://github.com/apache/iceberg/pull/18582#discussion_r4315417275
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java:
##########
@@ -78,13 +77,18 @@ 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) {
+ // 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()) {
Review Comment:
Do we need the isEmpty check?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Create the namespace when the catalog reports it as a missing table [iceberg]
Posted by "kumarpritam863 (via GitHub) via issues" <is...@iceberg.apache.org>.
kumarpritam863 commented on PR #18582:
URL: https://github.com/apache/iceberg/pull/18582#issuecomment-6663747640
@mxm Thanks for the review and valuable feedback. Can we please merge if the changes look good.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Create the namespace when the catalog reports it as a missing table [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #18582:
URL: https://github.com/apache/iceberg/pull/18582#discussion_r4334970131
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java:
##########
@@ -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);
+ }
Review Comment:
Let's use Preconidition.checkState instead.
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java:
##########
@@ -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);
Review Comment:
Let's move this into a `Precondition.checkState(catalog instanceof SupportsNamespace), ...);` check before line 93. We can then remove the if condition.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Create the namespace when the catalog reports it as a missing table [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #18582:
URL: https://github.com/apache/iceberg/pull/18582#discussion_r4334970131
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java:
##########
@@ -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);
+ }
Review Comment:
Let's use Precondition.checkState instead.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Create the namespace when the catalog reports it as a missing table [iceberg]
Posted by "kumarpritam863 (via GitHub)" <gi...@apache.org>.
kumarpritam863 commented on code in PR #18582:
URL: https://github.com/apache/iceberg/pull/18582#discussion_r4325082257
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java:
##########
@@ -78,14 +78,28 @@ 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) {
+ // 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) {
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Create the namespace when the catalog reports it as a missing table [iceberg]
Posted by "mxm (via GitHub) via issues" <is...@iceberg.apache.org>.
mxm commented on code in PR #18582:
URL: https://github.com/apache/iceberg/pull/18582#discussion_r4469745823
##########
flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java:
##########
@@ -78,8 +80,28 @@ 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) {
Review Comment:
That makes sense! Thanks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Create the namespace when the catalog reports it as a missing table [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #18582:
URL: https://github.com/apache/iceberg/pull/18582#discussion_r4315456895
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java:
##########
@@ -78,13 +77,18 @@ 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) {
+ // 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);
+ }
}
Review Comment:
I think we must throw if the catalog doesn't implement SupportsNamespaces (else branch) or we have another type of exception.
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java:
##########
@@ -78,13 +77,18 @@ 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) {
+ // 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()) {
Review Comment:
```java
if (exists.f1 instanceof NoSuchNamespaceException || exists.f1 instanceof NoSuchTableException) {
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Create the namespace when the catalog reports it as a missing table [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #18582:
URL: https://github.com/apache/iceberg/pull/18582#discussion_r4313816763
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java:
##########
@@ -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());
Review Comment:
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?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Create the namespace when the catalog reports it as a missing table [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #18582:
URL: https://github.com/apache/iceberg/pull/18582#discussion_r4336117964
##########
flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java:
##########
@@ -78,8 +80,28 @@ 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) {
Review Comment:
We still need this check. There is no need to run the code below when `exists.f1 == null` .
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Create the namespace when the catalog reports it as a missing table [iceberg]
Posted by "kumarpritam863 (via GitHub) via issues" <is...@iceberg.apache.org>.
kumarpritam863 commented on code in PR #18582:
URL: https://github.com/apache/iceberg/pull/18582#discussion_r4470275241
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java:
##########
@@ -78,11 +78,20 @@ 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) {
Review Comment:
Agreed, nothing reads it anymore. I removed it from TableMetadataCache.exists() altogether rather than just from the local variable, so it now returns a boolean. The EXISTS/NOT_EXISTS tuple constants went with it, and DynamicRecordProcessor no longer needs .f0. If you only meant the local variable, I can revert the cache part.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Create the namespace when the catalog reports it as a missing table [iceberg]
Posted by "kumarpritam863 (via GitHub)" <gi...@apache.org>.
kumarpritam863 commented on code in PR #18582:
URL: https://github.com/apache/iceberg/pull/18582#discussion_r4324949397
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java:
##########
@@ -78,14 +78,28 @@ 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) {
+ // 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) {
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 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);
}
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Create the namespace when the catalog reports it as a missing table [iceberg]
Posted by "kumarpritam863 (via GitHub) via issues" <is...@iceberg.apache.org>.
kumarpritam863 commented on code in PR #18582:
URL: https://github.com/apache/iceberg/pull/18582#discussion_r4470268687
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java:
##########
@@ -78,11 +78,20 @@ private void findOrCreateTable(
TableIdentifier identifier, Schema schema, PartitionSpec spec, TableCreator tableCreator) {
Tuple2<Boolean, Exception> exists = cache.exists(identifier);
if (Boolean.FALSE.equals(exists.f0)) {
- if (exists.f1 instanceof NoSuchNamespaceException) {
- SupportsNamespaces catalogWithNameSpace = (SupportsNamespaces) catalog;
+ Preconditions.checkState(
+ catalog instanceof SupportsNamespaces,
Review Comment:
Good point, that was a regression: before this PR such a catalog could still create tables, since the cast only happened inside the NoSuchNamespaceException branch. Removed the checkState. The namespace step now only runs for catalogs that implement SupportsNamespaces; others go straight to createTable.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Create the namespace when the catalog reports it as a missing table [iceberg]
Posted by "mxm (via GitHub) via issues" <is...@iceberg.apache.org>.
mxm commented on PR #18582:
URL: https://github.com/apache/iceberg/pull/18582#issuecomment-6680316588
Excellent work @kumarpritam863! Do you want to do the backport?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Create the namespace when the catalog reports it as a missing table [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #18582:
URL: https://github.com/apache/iceberg/pull/18582#discussion_r4323063158
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java:
##########
@@ -78,14 +78,28 @@ 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) {
+ // 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) {
Review Comment:
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.
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java:
##########
@@ -78,14 +78,28 @@ 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) {
+ // 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) {
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 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);
}
Review Comment:
We should also throw in case of other exceptions in `exists.f1`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Create the namespace when the catalog reports it as a missing table [iceberg]
Posted by "kumarpritam863 (via GitHub) via issues" <is...@iceberg.apache.org>.
kumarpritam863 commented on code in PR #18582:
URL: https://github.com/apache/iceberg/pull/18582#discussion_r4436258979
##########
flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java:
##########
@@ -78,8 +80,28 @@ 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) {
Review Comment:
- DynamicRecordProcessor.processRecord calls tableCache.exists() before it calls the updater. That caches the "not found" answer.
- When TableUpdater asks again, it gets the cached answer with f1 == null. Gating on f1 would skip namespace creation on the immediate-update path.
- Glue also throws NoSuchTableException for a missing namespace, not NoSuchNamespaceException.
This block only runs when the table is missing, so it runs once per table creation. A single namespaceExists call is negligible there. The new testNamespaceCreatedWhenCatalogReportsMissingNamespaceAsMissingTable primes the cache with cache.exists(...) first, as the processor does. It fails on the old code.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Create the namespace when the catalog reports it as a missing table [iceberg]
Posted by "kumarpritam863 (via GitHub) via issues" <is...@iceberg.apache.org>.
kumarpritam863 commented on code in PR #18582:
URL: https://github.com/apache/iceberg/pull/18582#discussion_r4436265630
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java:
##########
@@ -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);
Review Comment:
Done. The check is now the first statement in the Boolean.FALSE.equals(exists.f0) block, and the else branch is gone.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Create the namespace when the catalog reports it as a missing table [iceberg]
Posted by "kumarpritam863 (via GitHub) via issues" <is...@iceberg.apache.org>.
kumarpritam863 commented on code in PR #18582:
URL: https://github.com/apache/iceberg/pull/18582#discussion_r4436267141
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java:
##########
@@ -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);
+ }
Review Comment:
The remaining check is now a Preconditions.checkState.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Create the namespace when the catalog reports it as a missing table [iceberg]
Posted by "mxm (via GitHub) via issues" <is...@iceberg.apache.org>.
mxm merged PR #18582:
URL: https://github.com/apache/iceberg/pull/18582
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Create the namespace when the catalog reports it as a missing table [iceberg]
Posted by "mxm (via GitHub) via issues" <is...@iceberg.apache.org>.
mxm commented on code in PR #18582:
URL: https://github.com/apache/iceberg/pull/18582#discussion_r4469689855
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java:
##########
@@ -78,11 +78,20 @@ private void findOrCreateTable(
TableIdentifier identifier, Schema schema, PartitionSpec spec, TableCreator tableCreator) {
Tuple2<Boolean, Exception> exists = cache.exists(identifier);
if (Boolean.FALSE.equals(exists.f0)) {
- if (exists.f1 instanceof NoSuchNamespaceException) {
- SupportsNamespaces catalogWithNameSpace = (SupportsNamespaces) catalog;
+ Preconditions.checkState(
+ catalog instanceof SupportsNamespaces,
Review Comment:
What if the catalog does not implement `SupportsNamespaces`? I don't think we should throw then.
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java:
##########
@@ -78,11 +78,20 @@ 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) {
Review Comment:
If we don't use f1 anymore, we should remove it.
##########
flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java:
##########
@@ -453,4 +456,62 @@ void testTableCreationWithExistingNamespace() {
assertThat(result.f0.resolvedTableSchema().sameSchema(SCHEMA)).isTrue();
assertThat(result.f0.compareResult()).isEqualTo(CompareSchemasVisitor.Result.SAME);
}
+
+ @Test
+ void testNamespaceCreatedWhenCatalogReportsMissingNamespaceAsMissingTable() {
+ // Like the Glue catalog, which throws NoSuchTableException for a table in a missing namespace
+ InMemoryCatalog catalog =
+ new InMemoryCatalog() {
+ @Override
+ public Table loadTable(TableIdentifier identifier) {
+ if (!namespaceExists(identifier.namespace())) {
+ throw new NoSuchTableException("Table does not exist: %s", identifier);
+ }
+
+ return super.loadTable(identifier);
+ }
+ };
+ catalog.initialize("catalog", Map.of());
+ Namespace namespace = Namespace.of("glue_like_namespace");
+ TableIdentifier tableIdentifier = TableIdentifier.of(namespace, "myTable");
+ TableMetadataCache cache =
+ new TableMetadataCache(catalog, 10, Long.MAX_VALUE, 10, CASE_SENSITIVE, PRESERVE_COLUMNS);
+ TableUpdater tableUpdater = new TableUpdater(cache, catalog, CASE_SENSITIVE, PRESERVE_COLUMNS);
+
+ // The record processor looks the table up before calling the updater, so the updater only
+ // sees the cached answer, which carries no exception
+ cache.exists(tableIdentifier);
+
+ tableUpdater.update(
+ tableIdentifier,
+ SnapshotRef.MAIN_BRANCH,
+ SCHEMA,
+ PartitionSpec.unpartitioned(),
+ TableCreator.DEFAULT);
+
+ assertThat(catalog.namespaceExists(namespace)).isTrue();
+ assertThat(catalog.tableExists(tableIdentifier)).isTrue();
+ }
+
+ @Test
+ void testTableCreationFailsForCatalogWithoutNamespaceSupport() {
+ Catalog catalog = Mockito.mock(Catalog.class);
Review Comment:
Could we remove the mock? `CachingCatalog.wrap(inMemoryCatalog)` is a catalog without `SupportsNamespaces`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Create the namespace when the catalog reports it as a missing table [iceberg]
Posted by "kumarpritam863 (via GitHub) via issues" <is...@iceberg.apache.org>.
kumarpritam863 commented on code in PR #18582:
URL: https://github.com/apache/iceberg/pull/18582#discussion_r4470281578
##########
flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java:
##########
@@ -453,4 +456,62 @@ void testTableCreationWithExistingNamespace() {
assertThat(result.f0.resolvedTableSchema().sameSchema(SCHEMA)).isTrue();
assertThat(result.f0.compareResult()).isEqualTo(CompareSchemasVisitor.Result.SAME);
}
+
+ @Test
+ void testNamespaceCreatedWhenCatalogReportsMissingNamespaceAsMissingTable() {
+ // Like the Glue catalog, which throws NoSuchTableException for a table in a missing namespace
+ InMemoryCatalog catalog =
+ new InMemoryCatalog() {
+ @Override
+ public Table loadTable(TableIdentifier identifier) {
+ if (!namespaceExists(identifier.namespace())) {
+ throw new NoSuchTableException("Table does not exist: %s", identifier);
+ }
+
+ return super.loadTable(identifier);
+ }
+ };
+ catalog.initialize("catalog", Map.of());
+ Namespace namespace = Namespace.of("glue_like_namespace");
+ TableIdentifier tableIdentifier = TableIdentifier.of(namespace, "myTable");
+ TableMetadataCache cache =
+ new TableMetadataCache(catalog, 10, Long.MAX_VALUE, 10, CASE_SENSITIVE, PRESERVE_COLUMNS);
+ TableUpdater tableUpdater = new TableUpdater(cache, catalog, CASE_SENSITIVE, PRESERVE_COLUMNS);
+
+ // The record processor looks the table up before calling the updater, so the updater only
+ // sees the cached answer, which carries no exception
+ cache.exists(tableIdentifier);
+
+ tableUpdater.update(
+ tableIdentifier,
+ SnapshotRef.MAIN_BRANCH,
+ SCHEMA,
+ PartitionSpec.unpartitioned(),
+ TableCreator.DEFAULT);
+
+ assertThat(catalog.namespaceExists(namespace)).isTrue();
+ assertThat(catalog.tableExists(tableIdentifier)).isTrue();
+ }
+
+ @Test
+ void testTableCreationFailsForCatalogWithoutNamespaceSupport() {
+ Catalog catalog = Mockito.mock(Catalog.class);
Review Comment:
Done. The Mockito test is replaced by testTableCreatedForCatalogWithoutNamespaceSupport, which wraps an InMemoryCatalog in CachingCatalog and asserts the table gets created. It asserts the wrapper isn't a SupportsNamespaces, so it will flag it if that ever changes. The namespace is created on the inner catalog first, because InMemoryCatalog rejects tables in a missing namespace.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Create the namespace when the catalog reports it as a missing table [iceberg]
Posted by "kumarpritam863 (via GitHub)" <gi...@apache.org>.
kumarpritam863 commented on code in PR #18582:
URL: https://github.com/apache/iceberg/pull/18582#discussion_r4315856313
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java:
##########
@@ -78,13 +77,18 @@ 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) {
+ // 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()) {
Review Comment:
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.
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java:
##########
@@ -78,13 +77,18 @@ 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) {
+ // 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);
+ }
}
Review Comment:
For a catalog without SupportsNamespaces it now throws a clear UnsupportedOperationException when the lookup reported a missing namespace
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Create the namespace when the catalog reports it as a missing table [iceberg]
Posted by "kumarpritam863 (via GitHub)" <gi...@apache.org>.
kumarpritam863 commented on code in PR #18582:
URL: https://github.com/apache/iceberg/pull/18582#discussion_r4315112054
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java:
##########
@@ -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());
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Create the namespace when the catalog reports it as a missing table [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #18582:
URL: https://github.com/apache/iceberg/pull/18582#discussion_r4334981709
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java:
##########
@@ -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);
Review Comment:
Let's move this into a `Precondition.checkState(catalog instanceof SupportsNamespace, ...);` check before line 93. We can then remove the if condition.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Create the namespace when the catalog reports it as a missing table [iceberg]
Posted by "kumarpritam863 (via GitHub) via issues" <is...@iceberg.apache.org>.
kumarpritam863 commented on PR #18582:
URL: https://github.com/apache/iceberg/pull/18582#issuecomment-6680706828
Thanks @mxm for all the help and guidance via the crucial feedbacks. Sure I will take up the backport to other versions.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Dynamic Sink: Create the namespace when the catalog reports it as a missing table [iceberg]
Posted by "kumarpritam863 (via GitHub) via issues" <is...@iceberg.apache.org>.
kumarpritam863 commented on PR #18582:
URL: https://github.com/apache/iceberg/pull/18582#issuecomment-6681084173
@mxm please review the backport PR https://github.com/apache/iceberg/pull/18814
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org