Fix metadata identifiers containing backticks

NEXT_CHANGELOG.md

@@ -8,6 +8,7 @@
 
 ### Fixed
 - Fixed `setCatalog()` and `setSchema()` producing invalid SQL (e.g. `SET CATALOG ``name``) when the catalog or schema name was passed already wrapped in backticks. Backticks are now stripped before wrapping, and `getCatalog()`/`getSchema()` return the bare identifier name.
+- Fixed metadata SQL generation for catalog, schema, and table identifiers containing backticks.
 
 ---
 *Note: When making changes, please add your change under the appropriate section

src/main/java/com/databricks/jdbc/dbclient/impl/common/CommandConstants.java

@@ -132,7 +132,11 @@ private static ImmutableSqlParameter buildStringParam(int index, String value) {
         .build();
   }
 
+  public static String escapeSqlIdentifier(String identifier) {
+    return identifier == null ? null : identifier.replace("`", "``");
+  }
+
   private static String getCatalogPrefix(String catalog) {
-    return (catalog == null) ? "system" : "`" + catalog + "`";
+    return (catalog == null) ? "system" : "`" + escapeSqlIdentifier(catalog) + "`";
   }
 }

src/main/java/com/databricks/jdbc/dbclient/impl/sqlexec/CommandBuilder.java

@@ -87,7 +87,7 @@ private String fetchSchemaSQL() {
       // Per JDBC spec, null catalog means "do not narrow the search" — list across all catalogs
       showSchemasSQL = SHOW_SCHEMAS_IN_ALL_CATALOGS_SQL;
     } else {
-      showSchemasSQL = String.format(SHOW_SCHEMAS_IN_CATALOG_SQL, catalogName);
+      showSchemasSQL = String.format(SHOW_SCHEMAS_IN_CATALOG_SQL, escapeSqlIdentifier(catalogName));
     }
     if (schemaPattern != null) {
       showSchemasSQL += String.format(LIKE_SQL, schemaPattern);
@@ -107,7 +107,7 @@ private String fetchTablesSQL() {
       // Per JDBC spec, null catalog means "do not narrow the search" — list across all catalogs
       showTablesSQL = SHOW_TABLES_IN_ALL_CATALOGS_SQL;
     } else {
-      showTablesSQL = String.format(SHOW_TABLES_SQL, catalogName);
+      showTablesSQL = String.format(SHOW_TABLES_SQL, escapeSqlIdentifier(catalogName));
     }
     if (schemaPattern != null) {
       showTablesSQL += String.format(SCHEMA_LIKE_SQL, schemaPattern);
@@ -125,7 +125,7 @@ private String fetchColumnsSQL() throws DatabricksSQLException {
             catalogName, schemaPattern, tablePattern, columnPattern, sessionContext);
     LOGGER.debug(contextString);
     throwErrorIfNull(Collections.singletonMap(CATALOG, catalogName), contextString);
-    String showColumnsSQL = String.format(SHOW_COLUMNS_SQL, catalogName);
+    String showColumnsSQL = String.format(SHOW_COLUMNS_SQL, escapeSqlIdentifier(catalogName));
 
     if (schemaPattern != null) {
       showColumnsSQL += String.format(SCHEMA_LIKE_SQL, schemaPattern);
@@ -149,7 +149,7 @@ private String fetchFunctionsSQL() throws DatabricksSQLException {
 
     LOGGER.debug(contextString);
     throwErrorIfNull(Collections.singletonMap(CATALOG, catalogName), contextString);
-    String showFunctionsSQL = String.format(SHOW_FUNCTIONS_SQL, catalogName);
+    String showFunctionsSQL = String.format(SHOW_FUNCTIONS_SQL, escapeSqlIdentifier(catalogName));
     if (schemaPattern != null) {
       showFunctionsSQL += String.format(SCHEMA_LIKE_SQL, schemaPattern);
     }
@@ -174,7 +174,11 @@ private String fetchPrimaryKeysSQL() throws DatabricksSQLException {
     hashMap.put(SCHEMA, schemaName);
     hashMap.put(TABLE, tableName);
     throwErrorIfNull(hashMap, contextString);
-    return String.format(SHOW_PRIMARY_KEYS_SQL, catalogName, schemaName, tableName);
+    return String.format(
+        SHOW_PRIMARY_KEYS_SQL,
+        escapeSqlIdentifier(catalogName),
+        escapeSqlIdentifier(schemaName),
+        escapeSqlIdentifier(tableName));
   }
 
   private String fetchForeignKeysSQL() throws DatabricksSQLException {
@@ -188,7 +192,11 @@ private String fetchForeignKeysSQL() throws DatabricksSQLException {
     hashMap.put(SCHEMA, schemaName);
     hashMap.put(TABLE, tableName);
     throwErrorIfNull(hashMap, contextString);
-    return String.format(SHOW_FOREIGN_KEYS_SQL, catalogName, schemaName, tableName);
+    return String.format(
+        SHOW_FOREIGN_KEYS_SQL,
+        escapeSqlIdentifier(catalogName),
+        escapeSqlIdentifier(schemaName),
+        escapeSqlIdentifier(tableName));
   }
 
   public String getSQLString(CommandName command) throws DatabricksSQLException {

src/test/java/com/databricks/jdbc/dbclient/impl/common/CommandConstantsTest.java

@@ -0,0 +1,23 @@
+package com.databricks.jdbc.dbclient.impl.common;
+
+import static com.databricks.jdbc.dbclient.impl.common.CommandConstants.buildProceduresSQL;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import com.databricks.jdbc.api.impl.ImmutableSqlParameter;
+import java.util.HashMap;
+import java.util.Map;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+class CommandConstantsTest {
+
+  @Test
+  @DisplayName("Should escape backticks in information schema catalog prefix")
+  void shouldEscapeBackticksInInformationSchemaCatalogPrefix() {
+    Map<Integer, ImmutableSqlParameter> params = new HashMap<>();
+
+    String sql = buildProceduresSQL("cat`alog", null, null, params);
+
+    assertTrue(sql.contains(" FROM `cat``alog`.information_schema.routines"));
+  }
+}

src/test/java/com/databricks/jdbc/dbclient/impl/sqlexec/CommandBuilderTest.java

@@ -48,6 +48,17 @@ void shouldGenerateCorrectSqlForPrimaryKeys() throws SQLException {
       assertEquals(expectedSql, sql);
     }
 
+    @Test
+    @DisplayName("Should escape backticks in identifiers for primary keys")
+    void shouldEscapeBackticksInIdentifiersForPrimaryKeys() throws SQLException {
+      CommandBuilder builder =
+          new CommandBuilder("cat`alog", mockSession).setSchema("sch`ema").setTable("tab`le");
+
+      String sql = builder.getSQLString(CommandName.LIST_PRIMARY_KEYS);
+
+      assertEquals("SHOW KEYS IN CATALOG `cat``alog` IN SCHEMA `sch``ema` IN TABLE `tab``le`", sql);
+    }
+
     @Test
     @DisplayName("Should throw SQLException when catalog is null for primary keys")
     void shouldThrowExceptionWhenCatalogIsNullForPrimaryKeys() {
@@ -150,6 +161,16 @@ void shouldTreatWildcardCatalogAsLiteral() throws SQLException {
       assertEquals(String.format(SHOW_TABLES_SQL, "*"), sql1);
     }
 
+    @Test
+    @DisplayName("Should escape backticks in catalog identifier for tables")
+    void shouldEscapeBackticksInCatalogIdentifierForTables() throws SQLException {
+      CommandBuilder builder = new CommandBuilder("cat`alog", mockSession);
+
+      String sql = builder.getSQLString(CommandName.LIST_TABLES);
+
+      assertEquals("SHOW TABLES IN CATALOG `cat``alog`", sql);
+    }
+
     @Test
     @DisplayName("Should generate SCHEMA LIKE clause for empty string schema pattern")
     void shouldGenerateSchemaLikeClauseForEmptyStringSchemaPattern() throws SQLException {
@@ -201,6 +222,18 @@ void shouldGenerateCorrectSqlForForeignKeys() throws SQLException {
       assertEquals(expectedSql, sql);
     }
 
+    @Test
+    @DisplayName("Should escape backticks in identifiers for foreign keys")
+    void shouldEscapeBackticksInIdentifiersForForeignKeys() throws SQLException {
+      CommandBuilder builder =
+          new CommandBuilder("cat`alog", mockSession).setSchema("sch`ema").setTable("tab`le");
+
+      String sql = builder.getSQLString(CommandName.LIST_FOREIGN_KEYS);
+
+      assertEquals(
+          "SHOW FOREIGN KEYS IN CATALOG `cat``alog` IN SCHEMA `sch``ema` IN TABLE `tab``le`", sql);
+    }
+
     @Test
     @DisplayName("Should throw SQLException when catalog is null for foreign keys")
     void shouldThrowExceptionWhenCatalogIsNullForForeignKeys() {
@@ -227,6 +260,18 @@ void shouldThrowExceptionWhenTableIsNullForForeignKeys() {
     }
   }
 
+  @Test
+  @DisplayName("Should escape backticks in catalog identifiers for other metadata commands")
+  void shouldEscapeBackticksInCatalogIdentifiersForOtherMetadataCommands() throws SQLException {
+    CommandBuilder builder = new CommandBuilder("cat`alog", mockSession);
+
+    assertEquals("SHOW SCHEMAS IN `cat``alog`", builder.getSQLString(CommandName.LIST_SCHEMAS));
+    assertEquals(
+        "SHOW COLUMNS IN CATALOG `cat``alog`", builder.getSQLString(CommandName.LIST_COLUMNS));
+    assertEquals(
+        "SHOW FUNCTIONS IN CATALOG `cat``alog`", builder.getSQLString(CommandName.LIST_FUNCTIONS));
+  }
+
   @Test
   @DisplayName("Should throw exception for unsupported command")
   void shouldThrowExceptionForUnsupportedCommand() {