From 2046a40c4b9d38c95d2cacd5b22efd8e987887ca Mon Sep 17 00:00:00 2001
From: ismail simsek <6005685+ismailsimsek@users.noreply.github.com>
Date: Wed, 16 Aug 2023 22:12:34 +0200
Subject: [PATCH 01/11] JDBC catalog fix namespaceExists check

---
 .../org/apache/iceberg/jdbc/JdbcUtil.java     | 25 ++++++++++++++++---
 .../apache/iceberg/jdbc/TestJdbcCatalog.java  | 21 ++++++++++++++++
 2 files changed, 42 insertions(+), 4 deletions(-)

diff --git a/core/src/main/java/org/apache/iceberg/jdbc/JdbcUtil.java b/core/src/main/java/org/apache/iceberg/jdbc/JdbcUtil.java
index 3ffa47d2ea68..748f0ea73c7f 100644
--- a/core/src/main/java/org/apache/iceberg/jdbc/JdbcUtil.java
+++ b/core/src/main/java/org/apache/iceberg/jdbc/JdbcUtil.java
@@ -134,8 +134,13 @@ final class JdbcUtil {
           + " WHERE "
           + CATALOG_NAME
           + " = ? AND "
+          + " ( "
           + TABLE_NAMESPACE
-          + " LIKE ? LIMIT 1";
+          + " = ? OR "
+          + TABLE_NAMESPACE
+          + " LIKE ? ESCAPE '\\' "
+          + " ) "
+          + " LIMIT 1";
   static final String LIST_NAMESPACES_SQL =
       "SELECT DISTINCT "
           + TABLE_NAMESPACE
@@ -204,8 +209,12 @@ final class JdbcUtil {
           + " WHERE "
           + CATALOG_NAME
           + " = ? AND "
+          + " ( "
+          + NAMESPACE_NAME
+          + " = ? OR "
           + NAMESPACE_NAME
-          + " LIKE ? LIMIT 1";
+          + " LIKE ? ESCAPE '\\' "
+          + " ) ";
   static final String INSERT_NAMESPACE_PROPERTIES_SQL =
       "INSERT INTO "
           + NAMESPACE_PROPERTIES_TABLE_NAME
@@ -345,11 +354,18 @@ public static String deletePropertiesStatement(Set<String> properties) {
 
   static boolean namespaceExists(
       String catalogName, JdbcClientPool connections, Namespace namespace) {
+
+    String namespaceEquals = JdbcUtil.namespaceToString(namespace);
+    // when namespace has sub-namespace then additionally checking it with LIKE statement.
+    // catalog.db can exists as: catalog.db.ns1 or catalog.db.ns1.ns2
+    String namespaceStartsWith =
+        namespaceEquals.replace("\\", "\\\\").replace("_", "\\_").replace("%", "\\%") + ".%";
     if (exists(
         connections,
         JdbcUtil.GET_NAMESPACE_SQL,
         catalogName,
-        JdbcUtil.namespaceToString(namespace) + "%")) {
+        namespaceEquals,
+        namespaceStartsWith)) {
       return true;
     }
 
@@ -357,7 +373,8 @@ static boolean namespaceExists(
         connections,
         JdbcUtil.GET_NAMESPACE_PROPERTIES_SQL,
         catalogName,
-        JdbcUtil.namespaceToString(namespace) + "%")) {
+        namespaceEquals,
+        namespaceStartsWith)) {
       return true;
     }
 
diff --git a/core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java b/core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java
index 4634de57073d..47ef59385043 100644
--- a/core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java
+++ b/core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java
@@ -604,6 +604,27 @@ public void testCreateNamespace() {
     // Test with no metadata
     catalog.createNamespace(testNamespace);
     assertThat(catalog.namespaceExists(testNamespace)).isTrue();
+    assertThat(catalog.namespaceExists(Namespace.of("testDb"))).isTrue();
+    assertThat(catalog.namespaceExists(Namespace.of("testDb", "ns1"))).isTrue();
+    assertThat(catalog.namespaceExists(Namespace.of("testDb", "ns1", "ns2"))).isTrue();
+    assertThat(catalog.namespaceExists(Namespace.of("testDb_"))).isFalse();
+    assertThat(catalog.namespaceExists(Namespace.of("testD_"))).isFalse();
+    assertThat(catalog.namespaceExists(Namespace.of("testDb."))).isFalse();
+    assertThat(catalog.namespaceExists(Namespace.of("testDb.ns"))).isFalse();
+    assertThat(catalog.namespaceExists(Namespace.of("testDb.ns_"))).isFalse();
+    assertThat(catalog.namespaceExists(Namespace.of("testDb%"))).isFalse();
+
+    Namespace underscore = Namespace.of("te%t_b");
+    assertThat(catalog.namespaceExists(underscore))
+        .as("Should false to namespace doesn't exist")
+        .isFalse();
+
+    Namespace underscore2 = Namespace.of("special_name%space");
+    catalog.createNamespace(underscore2);
+    assertThat(catalog.namespaceExists(underscore2)).isTrue();
+    assertThat(catalog.namespaceExists(Namespace.of("special_name%spa_e")))
+        .as("Should false to namespace doesn't exist")
+        .isFalse();
   }
 
   @Test

From a01295015387d928b54f23880caa810a3cc1b5a1 Mon Sep 17 00:00:00 2001
From: ismail simsek <6005685+ismailsimsek@users.noreply.github.com>
Date: Fri, 20 Oct 2023 18:21:39 +0200
Subject: [PATCH 02/11] Improve testCreateNamespace test

---
 .../apache/iceberg/jdbc/TestJdbcCatalog.java  | 54 ++++++++++++++++---
 1 file changed, 47 insertions(+), 7 deletions(-)

diff --git a/core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java b/core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java
index 47ef59385043..6a542dee5093 100644
--- a/core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java
+++ b/core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java
@@ -601,27 +601,67 @@ public void testDropNamespace() {
   public void testCreateNamespace() {
     Namespace testNamespace = Namespace.of("testDb", "ns1", "ns2");
     assertThat(catalog.namespaceExists(testNamespace)).isFalse();
-    // Test with no metadata
     catalog.createNamespace(testNamespace);
     assertThat(catalog.namespaceExists(testNamespace)).isTrue();
     assertThat(catalog.namespaceExists(Namespace.of("testDb"))).isTrue();
     assertThat(catalog.namespaceExists(Namespace.of("testDb", "ns1"))).isTrue();
+    assertThat(catalog.namespaceExists(Namespace.of("testDb", "ns%"))).isFalse();
+    assertThat(catalog.namespaceExists(Namespace.of("testDb", "ns_"))).isFalse();
     assertThat(catalog.namespaceExists(Namespace.of("testDb", "ns1", "ns2"))).isTrue();
-    assertThat(catalog.namespaceExists(Namespace.of("testDb_"))).isFalse();
-    assertThat(catalog.namespaceExists(Namespace.of("testD_"))).isFalse();
-    assertThat(catalog.namespaceExists(Namespace.of("testDb."))).isFalse();
-    assertThat(catalog.namespaceExists(Namespace.of("testDb.ns"))).isFalse();
-    assertThat(catalog.namespaceExists(Namespace.of("testDb.ns_"))).isFalse();
-    assertThat(catalog.namespaceExists(Namespace.of("testDb%"))).isFalse();
+    assertThat(catalog.namespaceExists(Namespace.of("testDb", "ns1", "ns2", "ns3"))).isFalse();
+
+    testNamespace = Namespace.of("testDb_");
+    assertThat(catalog.namespaceExists(testNamespace)).isFalse();
+    catalog.createNamespace(testNamespace);
+    assertThat(catalog.namespaceExists(testNamespace)).isTrue();
+
+    testNamespace = Namespace.of("testD_");
+    assertThat(catalog.namespaceExists(testNamespace)).isFalse();
+    catalog.createNamespace(testNamespace);
+    assertThat(catalog.namespaceExists(testNamespace)).isTrue();
+
+    testNamespace = Namespace.of("testDb.");
+    assertThat(catalog.namespaceExists(testNamespace)).isFalse();
+    catalog.createNamespace(testNamespace);
+    assertThat(catalog.namespaceExists(testNamespace)).isTrue();
+
+    testNamespace = Namespace.of("testDb.ns");
+    assertThat(catalog.namespaceExists(testNamespace)).isFalse();
+    catalog.createNamespace(testNamespace);
+    assertThat(catalog.namespaceExists(testNamespace)).isTrue();
+
+    testNamespace = Namespace.of("testDb.ns_");
+    assertThat(catalog.namespaceExists(testNamespace)).isFalse();
+    catalog.createNamespace(testNamespace);
+    assertThat(catalog.namespaceExists(testNamespace)).isTrue();
+
+    testNamespace = Namespace.of("testDb%");
+    assertThat(catalog.namespaceExists(testNamespace)).isFalse();
+    catalog.createNamespace(testNamespace);
+    assertThat(catalog.namespaceExists(testNamespace)).isTrue();
+
+    testNamespace = Namespace.of("test\\Db", "ns\\1", "ns3");
+    assertThat(catalog.namespaceExists(testNamespace)).isFalse();
+    catalog.createNamespace(testNamespace);
+    assertThat(catalog.namespaceExists(testNamespace)).isTrue();
+
+    testNamespace = Namespace.of("test\\%Db","ns\\.1");
+    assertThat(catalog.namespaceExists(testNamespace)).isFalse();
+    catalog.createNamespace(testNamespace);
+    assertThat(catalog.namespaceExists(testNamespace)).isTrue();
 
     Namespace underscore = Namespace.of("te%t_b");
+    assertThat(catalog.namespaceExists(Namespace.of("te\\%t_b"))).isFalse();
+    assertThat(catalog.namespaceExists(Namespace.of("te%t\\_b"))).isFalse();
     assertThat(catalog.namespaceExists(underscore))
         .as("Should false to namespace doesn't exist")
         .isFalse();
 
     Namespace underscore2 = Namespace.of("special_name%space");
+    assertThat(catalog.namespaceExists(underscore2)).isFalse();
     catalog.createNamespace(underscore2);
     assertThat(catalog.namespaceExists(underscore2)).isTrue();
+    assertThat(catalog.namespaceExists(Namespace.of("special_name%"))).isFalse();
     assertThat(catalog.namespaceExists(Namespace.of("special_name%spa_e")))
         .as("Should false to namespace doesn't exist")
         .isFalse();

From 4aa72898e82599103621b6a6f702b23b49d3e9f5 Mon Sep 17 00:00:00 2001
From: ismail simsek <6005685+ismailsimsek@users.noreply.github.com>
Date: Sat, 21 Oct 2023 09:17:29 +0200
Subject: [PATCH 03/11] Improve testCreateNamespace test

---
 .../apache/iceberg/jdbc/TestJdbcCatalog.java    | 17 +++++++++++++++--
 1 file changed, 15 insertions(+), 2 deletions(-)

diff --git a/core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java b/core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java
index 6a542dee5093..d92e43fd32a9 100644
--- a/core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java
+++ b/core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java
@@ -645,7 +645,7 @@ public void testCreateNamespace() {
     catalog.createNamespace(testNamespace);
     assertThat(catalog.namespaceExists(testNamespace)).isTrue();
 
-    testNamespace = Namespace.of("test\\%Db","ns\\.1");
+    testNamespace = Namespace.of("test\\%Db", "ns\\.1");
     assertThat(catalog.namespaceExists(testNamespace)).isFalse();
     catalog.createNamespace(testNamespace);
     assertThat(catalog.namespaceExists(testNamespace)).isTrue();
@@ -665,12 +665,25 @@ public void testCreateNamespace() {
     assertThat(catalog.namespaceExists(Namespace.of("special_name%spa_e")))
         .as("Should false to namespace doesn't exist")
         .isFalse();
+
+    underscore2 =
+        Namespace.of("special_nested_name%space", "names\\pace1", "names_pace1", "name%s_pace1");
+    assertThat(catalog.namespaceExists(underscore2)).isFalse();
+    catalog.createNamespace(underscore2);
+    assertThat(catalog.namespaceExists(underscore2)).isTrue();
+    assertThat(catalog.namespaceExists(Namespace.of("special_nested_name%space", "names\\pace1")))
+        .isTrue();
+    assertThat(
+            catalog.namespaceExists(
+                Namespace.of("special_nested_name%space", "names\\pace1", "names_pace1")))
+        .isTrue();
+    assertThat(catalog.namespaceExists(Namespace.of("special_nested_name%space"))).isTrue();
   }
 
   @Test
   public void testCreateTableInNonExistingNamespace() {
     try (JdbcCatalog jdbcCatalog = initCatalog("non_strict_jdbc_catalog", ImmutableMap.of())) {
-      Namespace namespace = Namespace.of("testDb", "ns1", "ns2");
+      Namespace namespace = Namespace.of("test\\D_b%", "ns1", "ns2");
       TableIdentifier identifier = TableIdentifier.of(namespace, "someTable");
       Assertions.assertThat(jdbcCatalog.namespaceExists(namespace)).isFalse();
       Assertions.assertThat(jdbcCatalog.tableExists(identifier)).isFalse();

From cfc874c6866bab5956b792e245aa975e76470bfd Mon Sep 17 00:00:00 2001
From: ismail simsek <6005685+ismailsimsek@users.noreply.github.com>
Date: Sat, 21 Oct 2023 09:19:11 +0200
Subject: [PATCH 04/11] Improve testCreateNamespace test

---
 core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java b/core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java
index d92e43fd32a9..754cc6db5098 100644
--- a/core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java
+++ b/core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java
@@ -645,7 +645,7 @@ public void testCreateNamespace() {
     catalog.createNamespace(testNamespace);
     assertThat(catalog.namespaceExists(testNamespace)).isTrue();
 
-    testNamespace = Namespace.of("test\\%Db", "ns\\.1");
+    testNamespace = Namespace.of("test\\%Db","ns\\.1");
     assertThat(catalog.namespaceExists(testNamespace)).isFalse();
     catalog.createNamespace(testNamespace);
     assertThat(catalog.namespaceExists(testNamespace)).isTrue();

From ec6d06294595548d440dd873faa6e021e216e336 Mon Sep 17 00:00:00 2001
From: ismail simsek <6005685+ismailsimsek@users.noreply.github.com>
Date: Sat, 21 Oct 2023 09:20:10 +0200
Subject: [PATCH 05/11] Improve testCreateNamespace test

---
 core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java b/core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java
index 754cc6db5098..d92e43fd32a9 100644
--- a/core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java
+++ b/core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java
@@ -645,7 +645,7 @@ public void testCreateNamespace() {
     catalog.createNamespace(testNamespace);
     assertThat(catalog.namespaceExists(testNamespace)).isTrue();
 
-    testNamespace = Namespace.of("test\\%Db","ns\\.1");
+    testNamespace = Namespace.of("test\\%Db", "ns\\.1");
     assertThat(catalog.namespaceExists(testNamespace)).isFalse();
     catalog.createNamespace(testNamespace);
     assertThat(catalog.namespaceExists(testNamespace)).isTrue();

From 14c2a5e4833bc6d9b6fdf62038474594e5a94774 Mon Sep 17 00:00:00 2001
From: ismail simsek <6005685+ismailsimsek@users.noreply.github.com>
Date: Sat, 21 Oct 2023 09:21:53 +0200
Subject: [PATCH 06/11] Improve testCreateNamespace test

---
 core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java | 1 +
 1 file changed, 1 insertion(+)

diff --git a/core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java b/core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java
index d92e43fd32a9..1f8c0dc9a162 100644
--- a/core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java
+++ b/core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java
@@ -649,6 +649,7 @@ public void testCreateNamespace() {
     assertThat(catalog.namespaceExists(testNamespace)).isFalse();
     catalog.createNamespace(testNamespace);
     assertThat(catalog.namespaceExists(testNamespace)).isTrue();
+    assertThat(catalog.namespaceExists( Namespace.of("test\\%Db"))).isTrue();
 
     Namespace underscore = Namespace.of("te%t_b");
     assertThat(catalog.namespaceExists(Namespace.of("te\\%t_b"))).isFalse();

From 6508e2bca7ec10666073a1ca2a0ee8d07b16b6b3 Mon Sep 17 00:00:00 2001
From: ismail simsek <6005685+ismailsimsek@users.noreply.github.com>
Date: Sat, 21 Oct 2023 09:33:34 +0200
Subject: [PATCH 07/11] Improve testCreateNamespace test

---
 core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java b/core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java
index 1f8c0dc9a162..141a8faaef06 100644
--- a/core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java
+++ b/core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java
@@ -649,7 +649,7 @@ public void testCreateNamespace() {
     assertThat(catalog.namespaceExists(testNamespace)).isFalse();
     catalog.createNamespace(testNamespace);
     assertThat(catalog.namespaceExists(testNamespace)).isTrue();
-    assertThat(catalog.namespaceExists( Namespace.of("test\\%Db"))).isTrue();
+    assertThat(catalog.namespaceExists(Namespace.of("test\\%Db"))).isTrue();
 
     Namespace underscore = Namespace.of("te%t_b");
     assertThat(catalog.namespaceExists(Namespace.of("te\\%t_b"))).isFalse();

From 0718c8324ed1d08682a781f362938f80fde8342b Mon Sep 17 00:00:00 2001
From: ismail simsek <6005685+ismailsimsek@users.noreply.github.com>
Date: Wed, 25 Oct 2023 09:08:36 +0200
Subject: [PATCH 08/11] Split tests to testCreateNamespaceWithSpecialCharacter

---
 .../java/org/apache/iceberg/jdbc/TestJdbcCatalog.java    | 9 +++++++++
 1 file changed, 9 insertions(+)

diff --git a/core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java b/core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java
index 141a8faaef06..774f216a83cd 100644
--- a/core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java
+++ b/core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java
@@ -629,6 +629,15 @@ public void testCreateNamespace() {
     assertThat(catalog.namespaceExists(testNamespace)).isFalse();
     catalog.createNamespace(testNamespace);
     assertThat(catalog.namespaceExists(testNamespace)).isTrue();
+  }
+
+  @Test
+  public void testCreateNamespaceWithSpecialCharacter() {
+    // creating namespaces with special characters
+    // and existence check with special characters should succeed
+    Namespace testNamespace = Namespace.of("testDb", "ns1", "ns2");
+    catalog.createNamespace(testNamespace);
+    assertThat(catalog.namespaceExists(testNamespace)).isTrue();
 
     testNamespace = Namespace.of("testDb.ns_");
     assertThat(catalog.namespaceExists(testNamespace)).isFalse();

From 05d6d847a8e115c08b92ae70f5c40c6e6c69a364 Mon Sep 17 00:00:00 2001
From: ismail simsek <6005685+ismailsimsek@users.noreply.github.com>
Date: Wed, 25 Oct 2023 13:13:02 +0200
Subject: [PATCH 09/11] Organize and split testCreateNamespace tests

---
 .../apache/iceberg/jdbc/TestJdbcCatalog.java  | 104 ++++++++----------
 1 file changed, 43 insertions(+), 61 deletions(-)

diff --git a/core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java b/core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java
index 774f216a83cd..9e6c7e944733 100644
--- a/core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java
+++ b/core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java
@@ -601,93 +601,75 @@ public void testDropNamespace() {
   public void testCreateNamespace() {
     Namespace testNamespace = Namespace.of("testDb", "ns1", "ns2");
     assertThat(catalog.namespaceExists(testNamespace)).isFalse();
+    assertThat(catalog.namespaceExists(Namespace.of("testDb", "ns1"))).isFalse();
     catalog.createNamespace(testNamespace);
     assertThat(catalog.namespaceExists(testNamespace)).isTrue();
     assertThat(catalog.namespaceExists(Namespace.of("testDb"))).isTrue();
     assertThat(catalog.namespaceExists(Namespace.of("testDb", "ns1"))).isTrue();
+    assertThat(catalog.namespaceExists(Namespace.of("ns1", "ns2"))).isFalse();
     assertThat(catalog.namespaceExists(Namespace.of("testDb", "ns%"))).isFalse();
     assertThat(catalog.namespaceExists(Namespace.of("testDb", "ns_"))).isFalse();
     assertThat(catalog.namespaceExists(Namespace.of("testDb", "ns1", "ns2"))).isTrue();
     assertThat(catalog.namespaceExists(Namespace.of("testDb", "ns1", "ns2", "ns3"))).isFalse();
+  }
 
-    testNamespace = Namespace.of("testDb_");
-    assertThat(catalog.namespaceExists(testNamespace)).isFalse();
-    catalog.createNamespace(testNamespace);
-    assertThat(catalog.namespaceExists(testNamespace)).isTrue();
-
-    testNamespace = Namespace.of("testD_");
-    assertThat(catalog.namespaceExists(testNamespace)).isFalse();
-    catalog.createNamespace(testNamespace);
-    assertThat(catalog.namespaceExists(testNamespace)).isTrue();
-
-    testNamespace = Namespace.of("testDb.");
-    assertThat(catalog.namespaceExists(testNamespace)).isFalse();
-    catalog.createNamespace(testNamespace);
-    assertThat(catalog.namespaceExists(testNamespace)).isTrue();
+  @Test
+  public void testCreateNamespaceWithBackslashCharacter() {
 
-    testNamespace = Namespace.of("testDb.ns");
+    Namespace testNamespace = Namespace.of("test\\Db", "ns\\1", "ns3");
     assertThat(catalog.namespaceExists(testNamespace)).isFalse();
     catalog.createNamespace(testNamespace);
     assertThat(catalog.namespaceExists(testNamespace)).isTrue();
+    assertThat(catalog.namespaceExists(Namespace.of("test\\Db", "ns\\1"))).isTrue();
+    //
+    assertThat(catalog.namespaceExists(Namespace.of("test\\%Db", "ns\\.1"))).isFalse();
+    assertThat(catalog.namespaceExists(Namespace.of("test%Db", "ns\\.1"))).isFalse();
+    assertThat(catalog.namespaceExists(Namespace.of("test%Db"))).isFalse();
+    assertThat(catalog.namespaceExists(Namespace.of("test%"))).isFalse();
+    assertThat(catalog.namespaceExists(Namespace.of("test\\%"))).isFalse();
+    assertThat(catalog.namespaceExists(Namespace.of("test_Db", "ns\\.1"))).isFalse();
   }
 
   @Test
-  public void testCreateNamespaceWithSpecialCharacter() {
-    // creating namespaces with special characters
-    // and existence check with special characters should succeed
-    Namespace testNamespace = Namespace.of("testDb", "ns1", "ns2");
-    catalog.createNamespace(testNamespace);
-    assertThat(catalog.namespaceExists(testNamespace)).isTrue();
+  public void testCreateNamespaceWithPercentCharacter() {
 
-    testNamespace = Namespace.of("testDb.ns_");
-    assertThat(catalog.namespaceExists(testNamespace)).isFalse();
-    catalog.createNamespace(testNamespace);
-    assertThat(catalog.namespaceExists(testNamespace)).isTrue();
-
-    testNamespace = Namespace.of("testDb%");
+    Namespace testNamespace = Namespace.of("testDb%", "ns%1");
     assertThat(catalog.namespaceExists(testNamespace)).isFalse();
     catalog.createNamespace(testNamespace);
     assertThat(catalog.namespaceExists(testNamespace)).isTrue();
+    assertThat(catalog.namespaceExists(Namespace.of("testDb%"))).isTrue();
+    //
+    assertThat(catalog.namespaceExists(Namespace.of("testDb\\%"))).isFalse();
+    assertThat(catalog.namespaceExists(Namespace.of("testDb"))).isFalse();
+    assertThat(catalog.namespaceExists(Namespace.of("tes%Db%"))).isFalse();
+    assertThat(catalog.namespaceExists(Namespace.of("testDb%", "ns%"))).isFalse();
+  }
 
-    testNamespace = Namespace.of("test\\Db", "ns\\1", "ns3");
-    assertThat(catalog.namespaceExists(testNamespace)).isFalse();
+  @Test
+  public void testCreateNamespaceWithUnderscoreCharacter() {
+    Namespace testNamespace = Namespace.of("test_Db", "ns_1", "ns_");
     catalog.createNamespace(testNamespace);
     assertThat(catalog.namespaceExists(testNamespace)).isTrue();
+    assertThat(catalog.namespaceExists(Namespace.of("test_Db", "ns_1"))).isTrue();
+    //
+    assertThat(catalog.namespaceExists(Namespace.of("test_Db"))).isTrue();
+    assertThat(catalog.namespaceExists(Namespace.of("test_D_"))).isFalse();
+    assertThat(catalog.namespaceExists(Namespace.of("test_D%"))).isFalse();
+    assertThat(catalog.namespaceExists(Namespace.of("test_Db", "ns_"))).isFalse();
+    assertThat(catalog.namespaceExists(Namespace.of("test_Db", "ns_%"))).isFalse();
+  }
 
-    testNamespace = Namespace.of("test\\%Db", "ns\\.1");
-    assertThat(catalog.namespaceExists(testNamespace)).isFalse();
+  @Test
+  public void testCreateNamespaceWithDotCharacter() {
+    Namespace testNamespace = Namespace.of("test.Db", "ns1", "ns2");
     catalog.createNamespace(testNamespace);
     assertThat(catalog.namespaceExists(testNamespace)).isTrue();
-    assertThat(catalog.namespaceExists(Namespace.of("test\\%Db"))).isTrue();
-
-    Namespace underscore = Namespace.of("te%t_b");
-    assertThat(catalog.namespaceExists(Namespace.of("te\\%t_b"))).isFalse();
-    assertThat(catalog.namespaceExists(Namespace.of("te%t\\_b"))).isFalse();
-    assertThat(catalog.namespaceExists(underscore))
-        .as("Should false to namespace doesn't exist")
-        .isFalse();
-
-    Namespace underscore2 = Namespace.of("special_name%space");
-    assertThat(catalog.namespaceExists(underscore2)).isFalse();
-    catalog.createNamespace(underscore2);
-    assertThat(catalog.namespaceExists(underscore2)).isTrue();
-    assertThat(catalog.namespaceExists(Namespace.of("special_name%"))).isFalse();
-    assertThat(catalog.namespaceExists(Namespace.of("special_name%spa_e")))
-        .as("Should false to namespace doesn't exist")
-        .isFalse();
-
-    underscore2 =
-        Namespace.of("special_nested_name%space", "names\\pace1", "names_pace1", "name%s_pace1");
-    assertThat(catalog.namespaceExists(underscore2)).isFalse();
-    catalog.createNamespace(underscore2);
-    assertThat(catalog.namespaceExists(underscore2)).isTrue();
-    assertThat(catalog.namespaceExists(Namespace.of("special_nested_name%space", "names\\pace1")))
-        .isTrue();
-    assertThat(
-            catalog.namespaceExists(
-                Namespace.of("special_nested_name%space", "names\\pace1", "names_pace1")))
-        .isTrue();
-    assertThat(catalog.namespaceExists(Namespace.of("special_nested_name%space"))).isTrue();
+    assertThat(catalog.namespaceExists(Namespace.of("test.Db", "ns1"))).isTrue();
+    // TODO FIX handle Dot (its namespace SEPERATOR)? in the namespace levels. currently its
+    // accepted and threaded as a level
+    // Related to https://github.com/google/guava/issues/412
+    assertThat(catalog.namespaceExists(Namespace.of("test", "Db"))).isTrue();
+    assertThat(catalog.namespaceExists(Namespace.of("test", "Db", "ns1", "ns2"))).isTrue();
   }
 
   @Test

From ec812f9d8750f35e1dbec5c5d7b7db91e93d1465 Mon Sep 17 00:00:00 2001
From: ismail simsek <6005685+ismailsimsek@users.noreply.github.com>
Date: Sat, 28 Oct 2023 22:27:21 +0200
Subject: [PATCH 10/11] Temoved test with `.`, and extended
 testCreateNamespaceWithBackslashCharacter

---
 .../apache/iceberg/jdbc/TestJdbcCatalog.java  | 21 +++++++------------
 1 file changed, 8 insertions(+), 13 deletions(-)

diff --git a/core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java b/core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java
index 9e6c7e944733..e5a45eb7dacb 100644
--- a/core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java
+++ b/core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java
@@ -628,6 +628,14 @@ public void testCreateNamespaceWithBackslashCharacter() {
     assertThat(catalog.namespaceExists(Namespace.of("test%"))).isFalse();
     assertThat(catalog.namespaceExists(Namespace.of("test\\%"))).isFalse();
     assertThat(catalog.namespaceExists(Namespace.of("test_Db", "ns\\.1"))).isFalse();
+    //
+    testNamespace = Namespace.of("test\\%Db2", "ns1");
+    assertThat(catalog.namespaceExists(testNamespace)).isFalse();
+    catalog.createNamespace(testNamespace);
+    assertThat(catalog.namespaceExists(testNamespace)).isTrue();
+    assertThat(catalog.namespaceExists(Namespace.of("test\\%Db2"))).isTrue();
+    assertThat(catalog.namespaceExists(Namespace.of("test%Db2"))).isFalse();
+    assertThat(catalog.namespaceExists(Namespace.of("test\\_Db2"))).isFalse();
   }
 
   @Test
@@ -659,19 +667,6 @@ public void testCreateNamespaceWithUnderscoreCharacter() {
     assertThat(catalog.namespaceExists(Namespace.of("test_Db", "ns_%"))).isFalse();
   }
 
-  @Test
-  public void testCreateNamespaceWithDotCharacter() {
-    Namespace testNamespace = Namespace.of("test.Db", "ns1", "ns2");
-    catalog.createNamespace(testNamespace);
-    assertThat(catalog.namespaceExists(testNamespace)).isTrue();
-    assertThat(catalog.namespaceExists(Namespace.of("test.Db", "ns1"))).isTrue();
-    // TODO FIX handle Dot (its namespace SEPERATOR)? in the namespace levels. currently its
-    // accepted and threaded as a level
-    // Related to https://github.com/google/guava/issues/412
-    assertThat(catalog.namespaceExists(Namespace.of("test", "Db"))).isTrue();
-    assertThat(catalog.namespaceExists(Namespace.of("test", "Db", "ns1", "ns2"))).isTrue();
-  }
-
   @Test
   public void testCreateTableInNonExistingNamespace() {
     try (JdbcCatalog jdbcCatalog = initCatalog("non_strict_jdbc_catalog", ImmutableMap.of())) {

From 61dcb2ecaf199f11ba96f6fa565f2c0f214f115b Mon Sep 17 00:00:00 2001
From: Ismail Simsek <6005685+ismailsimsek@users.noreply.github.com>
Date: Thu, 7 Dec 2023 23:41:56 +0100
Subject: [PATCH 11/11] Add review improvements

---
 .../org/apache/iceberg/jdbc/TestJdbcCatalog.java     | 12 +++++-------
 1 file changed, 5 insertions(+), 7 deletions(-)

diff --git a/core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java b/core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java
index e5a45eb7dacb..43f0349781c3 100644
--- a/core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java
+++ b/core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java
@@ -606,29 +606,28 @@ public void testCreateNamespace() {
     assertThat(catalog.namespaceExists(testNamespace)).isTrue();
     assertThat(catalog.namespaceExists(Namespace.of("testDb"))).isTrue();
     assertThat(catalog.namespaceExists(Namespace.of("testDb", "ns1"))).isTrue();
+    assertThat(catalog.namespaceExists(Namespace.of("testDb", "ns1", "ns2"))).isTrue();
     assertThat(catalog.namespaceExists(Namespace.of("ns1", "ns2"))).isFalse();
     assertThat(catalog.namespaceExists(Namespace.of("testDb", "ns%"))).isFalse();
     assertThat(catalog.namespaceExists(Namespace.of("testDb", "ns_"))).isFalse();
-    assertThat(catalog.namespaceExists(Namespace.of("testDb", "ns1", "ns2"))).isTrue();
     assertThat(catalog.namespaceExists(Namespace.of("testDb", "ns1", "ns2", "ns3"))).isFalse();
   }
 
   @Test
   public void testCreateNamespaceWithBackslashCharacter() {
-
     Namespace testNamespace = Namespace.of("test\\Db", "ns\\1", "ns3");
     assertThat(catalog.namespaceExists(testNamespace)).isFalse();
     catalog.createNamespace(testNamespace);
     assertThat(catalog.namespaceExists(testNamespace)).isTrue();
     assertThat(catalog.namespaceExists(Namespace.of("test\\Db", "ns\\1"))).isTrue();
-    //
+    // test that SQL special characters `%`,`.`,`_` are escaped and returns false
     assertThat(catalog.namespaceExists(Namespace.of("test\\%Db", "ns\\.1"))).isFalse();
     assertThat(catalog.namespaceExists(Namespace.of("test%Db", "ns\\.1"))).isFalse();
     assertThat(catalog.namespaceExists(Namespace.of("test%Db"))).isFalse();
     assertThat(catalog.namespaceExists(Namespace.of("test%"))).isFalse();
     assertThat(catalog.namespaceExists(Namespace.of("test\\%"))).isFalse();
     assertThat(catalog.namespaceExists(Namespace.of("test_Db", "ns\\.1"))).isFalse();
-    //
+    // test that backslash with `%` is escaped and treated correctly
     testNamespace = Namespace.of("test\\%Db2", "ns1");
     assertThat(catalog.namespaceExists(testNamespace)).isFalse();
     catalog.createNamespace(testNamespace);
@@ -640,13 +639,12 @@ public void testCreateNamespaceWithBackslashCharacter() {
 
   @Test
   public void testCreateNamespaceWithPercentCharacter() {
-
     Namespace testNamespace = Namespace.of("testDb%", "ns%1");
     assertThat(catalog.namespaceExists(testNamespace)).isFalse();
     catalog.createNamespace(testNamespace);
     assertThat(catalog.namespaceExists(testNamespace)).isTrue();
     assertThat(catalog.namespaceExists(Namespace.of("testDb%"))).isTrue();
-    //
+    // test that searching with SQL special characters `\`,`%` are escaped and returns false
     assertThat(catalog.namespaceExists(Namespace.of("testDb\\%"))).isFalse();
     assertThat(catalog.namespaceExists(Namespace.of("testDb"))).isFalse();
     assertThat(catalog.namespaceExists(Namespace.of("tes%Db%"))).isFalse();
@@ -659,7 +657,7 @@ public void testCreateNamespaceWithUnderscoreCharacter() {
     catalog.createNamespace(testNamespace);
     assertThat(catalog.namespaceExists(testNamespace)).isTrue();
     assertThat(catalog.namespaceExists(Namespace.of("test_Db", "ns_1"))).isTrue();
-    //
+    // test that searching with SQL special characters `_`,`%` are escaped and returns false
     assertThat(catalog.namespaceExists(Namespace.of("test_Db"))).isTrue();
     assertThat(catalog.namespaceExists(Namespace.of("test_D_"))).isFalse();
     assertThat(catalog.namespaceExists(Namespace.of("test_D%"))).isFalse();