From aedfa9e7042b0abfc62725800d87623e8b1f8bf0 Mon Sep 17 00:00:00 2001
From: Gabor Kaszab <gaborkaszab@cloudera.com>
Date: Mon, 9 Dec 2024 21:10:50 +0100
Subject: [PATCH] Core: Fix loading a table in CachingCatalog with metadata
 table name

If a regular table had a metadata table name then CachingCatalog throws a
NoSuchTableException when loading that table.

Co-authored-by: Manu Zhang <OwenZhang1990@gmail.com>
---
 .../org/apache/iceberg/CachingCatalog.java    | 12 +++---
 .../iceberg/hadoop/TestCachingCatalog.java    | 38 +++++++++++++++++++
 2 files changed, 45 insertions(+), 5 deletions(-)

diff --git a/core/src/main/java/org/apache/iceberg/CachingCatalog.java b/core/src/main/java/org/apache/iceberg/CachingCatalog.java
index 1043e3e7205c..913f1a9482e1 100644
--- a/core/src/main/java/org/apache/iceberg/CachingCatalog.java
+++ b/core/src/main/java/org/apache/iceberg/CachingCatalog.java
@@ -144,14 +144,16 @@ public Table loadTable(TableIdentifier ident) {
       return cached;
     }
 
-    if (MetadataTableUtils.hasMetadataTableName(canonicalized)) {
+    Table table = tableCache.get(canonicalized, catalog::loadTable);
+
+    if (table instanceof BaseMetadataTable) {
+      // Cache underlying table
       TableIdentifier originTableIdentifier =
           TableIdentifier.of(canonicalized.namespace().levels());
       Table originTable = tableCache.get(originTableIdentifier, catalog::loadTable);
 
-      // share TableOperations instance of origin table for all metadata tables, so that metadata
-      // table instances are
-      // also refreshed as well when origin table instance is refreshed.
+      // Share TableOperations instance of origin table for all metadata tables, so that metadata
+      // table instances are refreshed as well when origin table instance is refreshed.
       if (originTable instanceof HasTableOperations) {
         TableOperations ops = ((HasTableOperations) originTable).operations();
         MetadataTableType type = MetadataTableType.from(canonicalized.name());
@@ -164,7 +166,7 @@ public Table loadTable(TableIdentifier ident) {
       }
     }
 
-    return tableCache.get(canonicalized, catalog::loadTable);
+    return table;
   }
 
   @Override
diff --git a/core/src/test/java/org/apache/iceberg/hadoop/TestCachingCatalog.java b/core/src/test/java/org/apache/iceberg/hadoop/TestCachingCatalog.java
index cb7ca641ea05..ce88c46973ce 100644
--- a/core/src/test/java/org/apache/iceberg/hadoop/TestCachingCatalog.java
+++ b/core/src/test/java/org/apache/iceberg/hadoop/TestCachingCatalog.java
@@ -32,6 +32,7 @@
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.iceberg.BaseMetadataTable;
 import org.apache.iceberg.CachingCatalog;
 import org.apache.iceberg.CatalogProperties;
 import org.apache.iceberg.MetadataTableType;
@@ -41,6 +42,7 @@
 import org.apache.iceberg.catalog.Catalog;
 import org.apache.iceberg.catalog.Namespace;
 import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.exceptions.NoSuchTableException;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.util.FakeTicker;
@@ -166,6 +168,42 @@ public void testTableName() throws Exception {
         .isEqualTo("hadoop.db.ns1.ns2.tbl.snapshots");
   }
 
+  @Test
+  public void testNonExistingTable() throws Exception {
+    Catalog catalog = CachingCatalog.wrap(hadoopCatalog());
+
+    TableIdentifier tableIdent = TableIdentifier.of("otherDB", "otherTbl");
+
+    assertThatThrownBy(() -> catalog.loadTable(tableIdent))
+        .isInstanceOf(NoSuchTableException.class);
+  }
+
+  @Test
+  public void testTableWithMetadataTableName() throws Exception {
+    TestableCachingCatalog catalog =
+        TestableCachingCatalog.wrap(hadoopCatalog(), EXPIRATION_TTL, ticker);
+    TableIdentifier tableIdent = TableIdentifier.of("db", "ns1", "ns2", "partitions");
+    TableIdentifier metaTableIdent =
+        TableIdentifier.of("db", "ns1", "ns2", "partitions", "partitions");
+
+    catalog.createTable(tableIdent, SCHEMA, SPEC, ImmutableMap.of("key2", "value2"));
+    catalog.cache().invalidateAll();
+
+    Table table = catalog.loadTable(tableIdent);
+    assertThat(table.name()).isEqualTo("hadoop.db.ns1.ns2.partitions");
+    assertThat(catalog.cache().asMap()).containsKey(tableIdent);
+    assertThat(catalog.cache().asMap()).doesNotContainKey(metaTableIdent);
+
+    catalog.cache().invalidateAll();
+    assertThat(catalog.cache().asMap()).doesNotContainKey(tableIdent);
+
+    Table metaTable = catalog.loadTable(metaTableIdent);
+    assertThat(metaTable).isInstanceOf(BaseMetadataTable.class);
+    assertThat(metaTable.name()).isEqualTo("hadoop.db.ns1.ns2.partitions.partitions");
+    assertThat(catalog.cache().asMap()).containsKey(tableIdent);
+    assertThat(catalog.cache().asMap()).containsKey(metaTableIdent);
+  }
+
   @Test
   public void testTableExpiresAfterInterval() throws IOException {
     TestableCachingCatalog catalog =