JDBC Catalog: Fix namespaceExists check with special characters (#8340)…

… (#9291)
apache · Dec 19, 2023 · 793f815 · 793f815
1 parent 5e84984
commit 793f815
Show file tree

Hide file tree

Showing 2 changed files with 82 additions and 6 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
@@ -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,19 +354,27 @@ 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;
     }
 
     if (exists(
         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
@@ -610,15 +610,74 @@ public void testDropNamespace() {
   public void testCreateNamespace() {
     Namespace testNamespace = Namespace.of("testDb", "ns1", "ns2");
     assertThat(catalog.namespaceExists(testNamespace)).isFalse();
-    // Test with no metadata
+    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("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", "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);
+    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
+  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();
+    assertThat(catalog.namespaceExists(Namespace.of("testDb%", "ns%"))).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();
+    // 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();
+    assertThat(catalog.namespaceExists(Namespace.of("test_Db", "ns_"))).isFalse();
+    assertThat(catalog.namespaceExists(Namespace.of("test_Db", "ns_%"))).isFalse();
   }
 
   @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();