From 2be084caf148281ba5b07de8f169667df6e015cc Mon Sep 17 00:00:00 2001
From: Amogh Jahagirdar <amogh@tabular.io>
Date: Mon, 27 Nov 2023 19:26:03 -0800
Subject: [PATCH] Return empty table list in Iceberg REST Catalog when invalid
 namespace is provided

---
 .../catalog/rest/TrinoRestCatalog.java        | 22 +++++-----
 .../iceberg/catalog/BaseTrinoCatalogTest.java | 41 +++++++++++++++++++
 2 files changed, 53 insertions(+), 10 deletions(-)

diff --git a/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/catalog/rest/TrinoRestCatalog.java b/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/catalog/rest/TrinoRestCatalog.java
index 29f6ef8e2443..71cd72315e40 100644
--- a/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/catalog/rest/TrinoRestCatalog.java
+++ b/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/catalog/rest/TrinoRestCatalog.java
@@ -183,16 +183,7 @@ public void renameNamespace(ConnectorSession session, String source, String targ
     public List<SchemaTableName> listTables(ConnectorSession session, Optional<String> namespace)
     {
         SessionContext sessionContext = convert(session);
-        List<Namespace> namespaces;
-
-        if (namespace.isPresent() && namespaceExists(session, namespace.get())) {
-            namespaces = ImmutableList.of(Namespace.of(namespace.get()));
-        }
-        else {
-            namespaces = listNamespaces(session).stream()
-                    .map(Namespace::of)
-                    .collect(toImmutableList());
-        }
+        List<Namespace> namespaces = listNamespaces(session, namespace);
 
         ImmutableList.Builder<SchemaTableName> tables = ImmutableList.builder();
         for (Namespace restNamespace : namespaces) {
@@ -555,4 +546,15 @@ private static TableIdentifier toIdentifier(SchemaTableName schemaTableName)
     {
         return TableIdentifier.of(schemaTableName.getSchemaName(), schemaTableName.getTableName());
     }
+
+    private List<Namespace> listNamespaces(ConnectorSession session, Optional<String> namespace)
+    {
+        if (namespace.isEmpty()) {
+            return listNamespaces(session).stream()
+                    .map(Namespace::of)
+                    .collect(toImmutableList());
+        }
+
+        return ImmutableList.of(Namespace.of(namespace.get()));
+    }
 }
diff --git a/plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/catalog/BaseTrinoCatalogTest.java b/plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/catalog/BaseTrinoCatalogTest.java
index cd896dcd9895..69830786be64 100644
--- a/plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/catalog/BaseTrinoCatalogTest.java
+++ b/plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/catalog/BaseTrinoCatalogTest.java
@@ -385,6 +385,47 @@ public void testView()
         }
     }
 
+    @Test
+    public void testListTables()
+            throws Exception
+    {
+        TrinoCatalog catalog = createTrinoCatalog(false);
+        TrinoPrincipal principal = new TrinoPrincipal(PrincipalType.USER, SESSION.getUser());
+        String ns1 = "ns1";
+        String ns2 = "ns2";
+
+        catalog.createNamespace(SESSION, ns1, defaultNamespaceProperties(ns1), principal);
+        catalog.createNamespace(SESSION, ns2, defaultNamespaceProperties(ns2), principal);
+        SchemaTableName table1 = new SchemaTableName(ns1, "t1");
+        SchemaTableName table2 = new SchemaTableName(ns2, "t2");
+        catalog.newCreateTableTransaction(
+                        SESSION,
+                        table1,
+                        new Schema(Types.NestedField.of(1, true, "col1", Types.LongType.get())),
+                        PartitionSpec.unpartitioned(),
+                        SortOrder.unsorted(),
+                        arbitraryTableLocation(catalog, SESSION, table1),
+                        ImmutableMap.of())
+                .commitTransaction();
+
+        catalog.newCreateTableTransaction(
+                        SESSION,
+                        table2,
+                        new Schema(Types.NestedField.of(1, true, "col1", Types.LongType.get())),
+                        PartitionSpec.unpartitioned(),
+                        SortOrder.unsorted(),
+                        arbitraryTableLocation(catalog, SESSION, table2),
+                        ImmutableMap.of())
+                .commitTransaction();
+
+        // No namespace provided, all tables across all namespaces should be returned
+        assertThat(catalog.listTables(SESSION, Optional.empty())).containsAll(ImmutableList.of(table1, table2));
+        // Namespace is provided and exists
+        assertThat(catalog.listTables(SESSION, Optional.of(ns1))).isEqualTo(ImmutableList.of(table1));
+        // Namespace is provided and does not exist
+        assertThat(catalog.listTables(SESSION, Optional.of("non_existing"))).isEmpty();
+    }
+
     private String arbitraryTableLocation(TrinoCatalog catalog, ConnectorSession session, SchemaTableName schemaTableName)
             throws Exception
     {