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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
