Fix getSchemas returning JDBC-escaped catalog name in SEA mode

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

@@ -9,6 +9,7 @@
 import com.databricks.jdbc.common.MetadataOperationType;
 import com.databricks.jdbc.common.StatementType;
 import com.databricks.jdbc.common.util.JdbcThreadUtils;
+import com.databricks.jdbc.common.util.WildcardUtil;
 import com.databricks.jdbc.dbclient.IDatabricksClient;
 import com.databricks.jdbc.dbclient.IDatabricksMetadataClient;
 import com.databricks.jdbc.dbclient.impl.common.CommandConstants;
@@ -98,9 +99,16 @@ public DatabricksResultSet listSchemas(
         new CommandBuilder(catalog, session).setSchemaPattern(schemaNamePattern);
     String SQL = commandBuilder.getSQLString(CommandName.LIST_SCHEMAS);
     LOGGER.debug("SQL command to fetch schemas: {}", SQL);
+    // Strip JDBC escape sequences from catalog for the result set TABLE_CATALOG column.
+    // SHOW SCHEMAS IN `catalog` doesn't return a catalog column from the server,
+    // so the client populates it from this parameter. Without stripping, JDBC-escaped
+    // underscores (\_) would appear in the result (e.g., "comparator\_tests" instead
+    // of "comparator_tests").
+    String resultCatalog =
+        catalog != null ? WildcardUtil.stripJdbcEscapes(catalog).toLowerCase() : null;
     try {
       return metadataResultSetBuilder.getSchemasResult(
-          getResultSet(SQL, session, MetadataOperationType.GET_SCHEMAS), catalog);
+          getResultSet(SQL, session, MetadataOperationType.GET_SCHEMAS), resultCatalog);
     } catch (SQLException e) {
       if (catalog == null && PARSE_SYNTAX_ERROR_SQL_STATE.equals(e.getSQLState())) {
         // This is a fallback for the case where the SQL command fails with "syntax error at or near
@@ -426,6 +434,16 @@ public DatabricksResultSet listCrossReferences(
       return metadataResultSetBuilder.getCrossRefsResult(new ArrayList<>());
     }
 
+    // When all three foreign-side parameters are null, SHOW FOREIGN KEYS cannot be constructed.
+    // Match Thrift server behavior which delegates to getExportedKeys in this case
+    // (returns 0 rows since exported keys are not tracked in DBSQL).
+    if (foreignCatalog == null && foreignSchema == null && foreignTable == null) {
+      LOGGER.debug(
+          "All foreign key parameters are null for getCrossReference, "
+              + "returning empty result set to match Thrift behavior.");
+      return metadataResultSetBuilder.getCrossRefsResult(new ArrayList<>());
+    }
+
     CommandBuilder commandBuilder =
         new CommandBuilder(foreignCatalog, session).setSchema(foreignSchema).setTable(foreignTable);
     String SQL = commandBuilder.getSQLString(CommandName.LIST_FOREIGN_KEYS);

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

@@ -458,6 +458,50 @@ void testListSchemas(String sqlStatement, String schema, String description) thr
         ((DatabricksResultSetMetaData) actualResult.getMetaData()).getTotalRows(), 1, description);
   }
 
+  /**
+   * Tests that getSchemas with a JDBC-escaped, mixed-case catalog name returns the unescaped,
+   * lowercased catalog name in the TABLE_CATALOG column. This reproduces the SEA/Thrift parity
+   * issue where SHOW SCHEMAS IN `catalog` doesn't return a catalog column from the server, so the
+   * client populates it from the parameter — which must be unescaped and lowercased.
+   */
+  @Test
+  void testListSchemasWithEscapedUnderscoreCatalog() throws SQLException {
+    String escapedCatalog = "Comparator\\_Tests";
+    String expectedCatalog = "comparator_tests";
+    // CommandBuilder strips escapes for SQL: SHOW SCHEMAS IN `Comparator_Tests`
+    String expectedSQL = "SHOW SCHEMAS IN `Comparator_Tests`";
+
+    when(session.getComputeResource()).thenReturn(mockedComputeResource);
+    DatabricksMetadataQueryClient metadataClient = new DatabricksMetadataQueryClient(mockClient);
+    when(mockClient.executeStatement(
+            eq(expectedSQL),
+            eq(mockedComputeResource),
+            any(),
+            eq(StatementType.METADATA),
+            eq(session),
+            any(),
+            any(MetadataOperationType.class)))
+        .thenReturn(mockedResultSet);
+    when(mockedResultSet.next()).thenReturn(true, false);
+    when(mockedResultSet.getObject("databaseName")).thenReturn("default");
+    doReturn(2).when(mockedMetaData).getColumnCount();
+    doReturn(SCHEMA_COLUMN.getResultSetColumnName()).when(mockedMetaData).getColumnName(1);
+    doReturn(CATALOG_COLUMN.getResultSetColumnName()).when(mockedMetaData).getColumnName(2);
+    when(mockedResultSet.getMetaData()).thenReturn(mockedMetaData);
+    // SHOW SCHEMAS IN `catalog` doesn't return a catalog column — the client must populate it
+    when(mockedResultSet.findColumn(CATALOG_RESULT_COLUMN.getResultSetColumnName()))
+        .thenThrow(DatabricksSQLException.class);
+
+    DatabricksResultSet actualResult = metadataClient.listSchemas(session, escapedCatalog, null);
+
+    assertTrue(actualResult.next());
+    // TABLE_CATALOG (column 2) should be unescaped and lowercased
+    assertEquals(
+        expectedCatalog,
+        actualResult.getObject(2),
+        "TABLE_CATALOG should be unescaped and lowercased to match Thrift behavior");
+  }
+
   @Test
   void testListSchemasNullCatalog() throws SQLException {
     when(session.getComputeResource()).thenReturn(mockedComputeResource);
@@ -717,6 +761,21 @@ void testListCrossReferences() throws Exception {
     }
   }
 
+  /**
+   * Tests that getCrossReference returns empty result set (not an exception) when all three
+   * foreign-side parameters are null. Matches Thrift server behavior where null foreign table
+   * delegates to getExportedKeys which returns empty in DBSQL.
+   */
+  @Test
+  void testListCrossReferences_allForeignParamsNull_returnsEmpty() throws Exception {
+    DatabricksMetadataQueryClient metadataClient = new DatabricksMetadataQueryClient(mockClient);
+
+    DatabricksResultSet result =
+        metadataClient.listCrossReferences(
+            session, TEST_CATALOG, TEST_SCHEMA, TEST_TABLE, null, null, null);
+    assertFalse(result.next(), "Should return empty when all foreign params are null, not throw");
+  }
+
   @Test
   void testListCrossReferences_throwsParseSyntaxError() throws Exception {
     DatabricksSQLException exception =