Resolve null catalog/schema in SEA key-based metadata operations (#1370)

## Summary

- Adds null-resolution logic for SEA key-based metadata operations
(`getPrimaryKeys`, `getImportedKeys`, `getCrossReference`) to match
Thrift server behavior
- Adds `getCurrentCatalogAndSchema()` to `IDatabricksSession` — fetches
both current catalog and schema in a single `SELECT CURRENT_CATALOG(),
CURRENT_SCHEMA()` query
- Refactors `getCurrentCatalog()` to delegate to
`getCurrentCatalogAndSchema()[0]`, eliminating duplicate query logic
- Adds `resolveKeyBasedParams()` as a shared helper in
`DatabricksMetadataQueryClient` for consistent null-replacement across
all key-based operations
- Adds 15 integration tests with WireMock record/replay stubs

## Problem

SEA metadata operations build SQL commands like `SHOW KEYS IN CATALOG x
IN SCHEMA y IN TABLE z` which require explicit values. When users pass
`null` for catalog/schema (standard JDBC behavior), SEA previously
returned empty results or threw errors — unlike Thrift where the server
resolves nulls to current catalog/schema.

## Null Resolution Rules (matching Thrift server behavior)

Verified via live Thrift testing against Databricks workspace:

| Parameter | Rule |
|---|---|
| `catalog=null` | Replace with `CURRENT_CATALOG()` |
| `schema=null` + catalog was also null | Replace with
`CURRENT_SCHEMA()` |
| `schema=null` + catalog explicitly provided | Return empty result |
| `table=null` | Return empty result |

The key invariant: **schema is only auto-filled when catalog is also
null**. If catalog is explicitly provided (even if it matches the
current catalog), null schema returns empty — matching Thrift server
behavior exactly.

## Thrift Behavior Evidence

With current catalog=`main`, current schema=`msr_testing`:

**getPrimaryKeys:**
- `(null, null, 'test_parent')` → **1 row found** (resolved to current
catalog+schema)
- `(null, 'msr_testing', 'test_parent')` → **1 row found** (null catalog
resolved, explicit schema used)
- `('main', null, 'test_parent')` → **EXCEPTION** (explicit catalog +
null schema)
- `('main', 'msr_testing', 'test_parent')` → **1 row found** (fully
specified)

**getCrossReference:**
- `(null, null, test_parent, null, null, test_child)` → **1 row** (both
sides resolved)
- `(null, msr_testing, test_parent, null, msr_testing, test_child)` →
**1 row**
- `(main, null, test_parent, main, msr_testing, test_child)` →
**EXCEPTION** (explicit catalog + null schema)

## Files Changed

| File | Change |
|---|---|
| `IDatabricksSession.java` | Added `getCurrentCatalogAndSchema()`
interface method |
| `DatabricksSession.java` | Implemented `getCurrentCatalogAndSchema()`,
refactored `getCurrentCatalog()` to delegate |
| `DatabricksMetadataQueryClient.java` | Added
`resolveKeyBasedParams()`, updated `listPrimaryKeys`,
`listImportedKeys`, `listCrossReferences` |
| `DatabricksMetadataQueryClientTest.java` | Added
`testKeyBasedOpsReturnEmptyForNullTable`,
`testKeyBasedOpsReturnEmptyForNullSchemaWithExplicitCatalog`, removed
stale stubs |
| `MetadataNullResolutionTests.java` | New: 15 integration tests with
WireMock stubs for all null-resolution scenarios |

## Not Changed

- **Thrift implementation** — server handles nulls, no client changes
needed
- **Pattern-based operations** (getSchemas, getTables, getColumns,
getFunctions) — these use LIKE patterns, not exact keys
- **listExportedKeys** — already returns empty (not supported in DBSQL)
- **CommandBuilder** — receives resolved non-null values

## Test plan

- [x] Unit tests pass: `DatabricksMetadataQueryClientTest` (52 tests)
- [x] Unit tests pass: `DatabricksDatabaseMetaDataTest` (248 tests)
- [x] Integration tests pass: `MetadataNullResolutionTests` (15 tests,
REPLAY mode)
- [x] Integration tests recorded and verified against e2-dogfood staging
(RECORD mode)
- [x] Live verification against Databricks workspace with Thrift to
confirm behavior parity

NO_CHANGELOG=true

This pull request was AI-assisted by Isaac.

---------

Signed-off-by: Madhavendra Rathore <[email protected]>

Author: msrathore-db

src/main/java/com/databricks/jdbc/api/impl/DatabricksSession.java

@@ -363,8 +363,7 @@ public String getCurrentCatalog() throws DatabricksSQLException {
               null /* metadataOperationType */);
 
       if (resultSet.next()) {
-        String currentCatalog = resultSet.getString(1);
-        return currentCatalog;
+        return resultSet.getString(1);
       }
     } catch (Exception e) {
       LOGGER.warn(
@@ -374,6 +373,30 @@ public String getCurrentCatalog() throws DatabricksSQLException {
     return this.catalog;
   }
 
+  @Override
+  public String[] getCurrentCatalogAndSchema() throws DatabricksSQLException {
+    try {
+      DatabricksResultSet resultSet =
+          databricksClient.executeStatement(
+              "SELECT CURRENT_CATALOG(), CURRENT_SCHEMA()",
+              this.computeResource,
+              new HashMap<>(),
+              StatementType.METADATA,
+              this,
+              null,
+              null /* metadataOperationType */);
+
+      if (resultSet.next()) {
+        return new String[] {resultSet.getString(1), resultSet.getString(2)};
+      }
+    } catch (Exception e) {
+      LOGGER.warn(
+          "Failed to get current catalog and schema from database, falling back to session values: {}",
+          e.getMessage());
+    }
+    return new String[] {this.catalog, this.schema};
+  }
+
   @Override
   public void setEmptyMetadataClient() {
     databricksMetadataClient = new DatabricksEmptyMetadataClient(connectionContext);

src/main/java/com/databricks/jdbc/api/internal/IDatabricksSession.java

@@ -125,6 +125,13 @@ public interface IDatabricksSession {
   /** Gets the current catalog from the database */
   String getCurrentCatalog() throws DatabricksSQLException;
 
+  /**
+   * Gets the current catalog and schema from the database in a single query.
+   *
+   * @return String array of length 2: [currentCatalog, currentSchema]
+   */
+  String[] getCurrentCatalogAndSchema() throws DatabricksSQLException;
+
   void setEmptyMetadataClient();
 
   void forceClose();

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

@@ -323,10 +323,10 @@ public DatabricksResultSet listPrimaryKeys(
 
     catalog = autoFillCatalog(catalog, currentCatalog);
 
-    // Return empty result set if catalog, schema, or table is null
-    if (catalog == null || schema == null || table == null) {
+    String[] resolvedParams = resolveKeyBasedParams(catalog, schema, table, session);
+    if (resolvedParams == null) {
       LOGGER.debug(
-          "Catalog, schema, or table is null (catalog={}, schema={}, table={}), returning empty result set for listPrimaryKeys",
+          "Could not resolve key-based params (catalog={}, schema={}, table={}), returning empty result set for listPrimaryKeys",
           catalog,
           schema,
           table);
@@ -337,8 +337,14 @@ public DatabricksResultSet listPrimaryKeys(
           com.databricks.jdbc.common.CommandName.LIST_PRIMARY_KEYS);
     }
 
+    String resolvedCatalog = resolvedParams[0];
+    String resolvedSchema = resolvedParams[1];
+    String resolvedTable = resolvedParams[2];
+
     CommandBuilder commandBuilder =
-        new CommandBuilder(catalog, session).setSchema(schema).setTable(table);
+        new CommandBuilder(resolvedCatalog, session)
+            .setSchema(resolvedSchema)
+            .setTable(resolvedTable);
     String SQL = commandBuilder.getSQLString(CommandName.LIST_PRIMARY_KEYS);
     LOGGER.debug("SQL command to fetch primary keys: {}", SQL);
     try {
@@ -366,10 +372,10 @@ public DatabricksResultSet listImportedKeys(
 
     catalog = autoFillCatalog(catalog, currentCatalog);
 
-    // Return empty result set if catalog, schema, or table is null
-    if (catalog == null || schema == null || table == null) {
+    String[] resolvedParams = resolveKeyBasedParams(catalog, schema, table, session);
+    if (resolvedParams == null) {
       LOGGER.debug(
-          "Catalog, schema, or table is null (catalog={}, schema={}, table={}), returning empty result set for listImportedKeys",
+          "Could not resolve key-based params (catalog={}, schema={}, table={}), returning empty result set for listImportedKeys",
           catalog,
           schema,
           table);
@@ -380,8 +386,14 @@ public DatabricksResultSet listImportedKeys(
           com.databricks.jdbc.common.CommandName.GET_IMPORTED_KEYS);
     }
 
+    String resolvedCatalog = resolvedParams[0];
+    String resolvedSchema = resolvedParams[1];
+    String resolvedTable = resolvedParams[2];
+
     CommandBuilder commandBuilder =
-        new CommandBuilder(catalog, session).setSchema(schema).setTable(table);
+        new CommandBuilder(resolvedCatalog, session)
+            .setSchema(resolvedSchema)
+            .setTable(resolvedTable);
     String SQL = commandBuilder.getSQLString(CommandName.LIST_FOREIGN_KEYS);
     try {
       return metadataResultSetBuilder.getImportedKeysResult(
@@ -434,25 +446,48 @@ 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) {
+    // Resolve null params for the foreign side (used to build the SQL query)
+    String[] resolvedForeignParams =
+        resolveKeyBasedParams(foreignCatalog, foreignSchema, foreignTable, session);
+    if (resolvedForeignParams == null) {
       LOGGER.debug(
-          "All foreign key parameters are null for getCrossReference, "
-              + "returning empty result set to match Thrift behavior.");
+          "Could not resolve foreign key-based params (catalog={}, schema={}, table={}), returning empty result set",
+          foreignCatalog,
+          foreignSchema,
+          foreignTable);
       return metadataResultSetBuilder.getCrossRefsResult(new ArrayList<>());
     }
 
+    // Resolve null params for the parent side (used for filtering results)
+    String[] resolvedParentParams =
+        resolveKeyBasedParams(parentCatalog, parentSchema, parentTable, session);
+    if (resolvedParentParams == null) {
+      LOGGER.debug(
+          "Could not resolve parent key-based params (catalog={}, schema={}, table={}), returning empty result set",
+          parentCatalog,
+          parentSchema,
+          parentTable);
+      return metadataResultSetBuilder.getCrossRefsResult(new ArrayList<>());
+    }
+
+    String resolvedForeignCatalog = resolvedForeignParams[0];
+    String resolvedForeignSchema = resolvedForeignParams[1];
+    String resolvedForeignTable = resolvedForeignParams[2];
+    String resolvedParentCatalog = resolvedParentParams[0];
+    String resolvedParentSchema = resolvedParentParams[1];
+    String resolvedParentTable = resolvedParentParams[2];
+
     CommandBuilder commandBuilder =
-        new CommandBuilder(foreignCatalog, session).setSchema(foreignSchema).setTable(foreignTable);
+        new CommandBuilder(resolvedForeignCatalog, session)
+            .setSchema(resolvedForeignSchema)
+            .setTable(resolvedForeignTable);
     String SQL = commandBuilder.getSQLString(CommandName.LIST_FOREIGN_KEYS);
     try {
       return metadataResultSetBuilder.getCrossReferenceKeysResult(
           getResultSet(SQL, session, MetadataOperationType.GET_CROSS_REFERENCE),
-          parentCatalog,
-          parentSchema,
-          parentTable);
+          resolvedParentCatalog,
+          resolvedParentSchema,
+          resolvedParentTable);
     } catch (SQLException e) {
       if (PARSE_SYNTAX_ERROR_SQL_STATE.equals(e.getSQLState()) || isObjectNotFoundException(e)) {
         LOGGER.debug(
@@ -506,6 +541,35 @@ private String autoFillCatalog(String catalog, String currentCatalog) {
     return catalog;
   }
 
+  /**
+   * Resolves null catalog/schema for key-based metadata operations to match Thrift server behavior.
+   * When catalog is null, it is replaced with current_catalog and (if schema is also null) schema
+   * is replaced with current_schema. Returns null if the caller should return an empty result set
+   * (table is null, schema is null without catalog also being null, or any resolved value is null).
+   */
+  private String[] resolveKeyBasedParams(
+      String catalog, String schema, String table, IDatabricksSession session) throws SQLException {
+    if (table == null) {
+      return null;
+    }
+
+    if (catalog == null) {
+      String[] currentCatalogAndSchema = session.getCurrentCatalogAndSchema();
+      catalog = currentCatalogAndSchema[0];
+      if (schema == null) {
+        schema = currentCatalogAndSchema[1];
+      }
+    } else if (schema == null) {
+      return null;
+    }
+
+    if (catalog == null || schema == null) {
+      return null;
+    }
+
+    return new String[] {catalog, schema, table};
+  }
+
   private DatabricksResultSet getResultSet(
       String SQL, IDatabricksSession session, MetadataOperationType metadataOperationType)
       throws SQLException {

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

@@ -624,9 +624,6 @@ void testImportedKeys_throwsParseSyntaxError() throws Exception {
         new DatabricksSQLException(
             "syntax error at or near \"foreign\"", PARSE_SYNTAX_ERROR_SQL_STATE);
     when(session.getComputeResource()).thenReturn(WAREHOUSE_COMPUTE);
-    IDatabricksConnectionContext mockContext = mock(IDatabricksConnectionContext.class);
-    when(mockContext.getEnableMultipleCatalogSupport()).thenReturn(true);
-    when(mockClient.getConnectionContext()).thenReturn(mockContext);
     DatabricksMetadataQueryClient metadataClient = new DatabricksMetadataQueryClient(mockClient);
     when(mockClient.executeStatement(
             eq(
@@ -946,27 +943,40 @@ void testListFunctionsWithNullCatalog() throws SQLException {
   }
 
   @Test
-  void testReturnsEmptyResultSetInCaseOfNullCatalog() throws SQLException {
-    IDatabricksConnectionContext mockContext = mock(IDatabricksConnectionContext.class);
-    when(mockContext.getEnableMultipleCatalogSupport()).thenReturn(true);
-    when(mockClient.getConnectionContext()).thenReturn(mockContext);
+  void testKeyBasedOpsReturnEmptyForNullTable() throws SQLException {
+    DatabricksMetadataQueryClient metadataClient = new DatabricksMetadataQueryClient(mockClient);
+
+    // null table should return empty for listPrimaryKeys
+    DatabricksResultSet pkResult =
+        metadataClient.listPrimaryKeys(session, TEST_CATALOG, TEST_SCHEMA, null);
+    assertNotNull(pkResult);
+    assertFalse(pkResult.next(), "Expected empty result set for listPrimaryKeys with null table");
+
+    // null table should return empty for listImportedKeys
+    DatabricksResultSet ikResult =
+        metadataClient.listImportedKeys(session, TEST_CATALOG, TEST_SCHEMA, null);
+    assertNotNull(ikResult);
+    assertFalse(ikResult.next(), "Expected empty result set for listImportedKeys with null table");
+  }
+
+  @Test
+  void testKeyBasedOpsReturnEmptyForNullSchemaWithExplicitCatalog() throws SQLException {
     DatabricksMetadataQueryClient metadataClient = new DatabricksMetadataQueryClient(mockClient);
 
-    // listPrimaryKeys with null catalog should return empty ResultSet
-    DatabricksResultSet primaryKeysResult =
-        metadataClient.listPrimaryKeys(session, null, TEST_SCHEMA, TEST_TABLE);
-    assertNotNull(primaryKeysResult);
+    // schema=null with explicit catalog should return empty (matching Thrift behavior)
+    DatabricksResultSet pkResult =
+        metadataClient.listPrimaryKeys(session, "any_catalog", null, TEST_TABLE);
+    assertNotNull(pkResult);
     assertFalse(
-        primaryKeysResult.next(),
-        "Expected empty result set for listPrimaryKeys with null catalog");
+        pkResult.next(),
+        "Expected empty result set for listPrimaryKeys with null schema and explicit catalog");
 
-    // listImportedKeys with null catalog should return empty ResultSet
-    DatabricksResultSet importedKeysResult =
-        metadataClient.listImportedKeys(session, null, TEST_SCHEMA, TEST_TABLE);
-    assertNotNull(importedKeysResult);
+    DatabricksResultSet ikResult =
+        metadataClient.listImportedKeys(session, "any_catalog", null, TEST_TABLE);
+    assertNotNull(ikResult);
     assertFalse(
-        importedKeysResult.next(),
-        "Expected empty result set for listImportedKeys with null catalog");
+        ikResult.next(),
+        "Expected empty result set for listImportedKeys with null schema and explicit catalog");
   }
 
   @Test
@@ -1176,9 +1186,6 @@ void testListImportedKeys_handlesNullSqlStateWithoutNPE() throws Exception {
             "syntax error at or near \"foreign\"", (String) null); // null SQL state
 
     when(session.getComputeResource()).thenReturn(WAREHOUSE_COMPUTE);
-    IDatabricksConnectionContext mockContext = mock(IDatabricksConnectionContext.class);
-    when(mockContext.getEnableMultipleCatalogSupport()).thenReturn(true);
-    when(mockClient.getConnectionContext()).thenReturn(mockContext);
 
     DatabricksMetadataQueryClient metadataClient = new DatabricksMetadataQueryClient(mockClient);
     when(mockClient.executeStatement(
@@ -1205,10 +1212,6 @@ void testListCrossReferences_handlesNullSqlStateWithoutNPE() throws Exception {
             "syntax error at or near \"foreign\"", (String) null); // null SQL state
 
     when(session.getComputeResource()).thenReturn(WAREHOUSE_COMPUTE);
-    IDatabricksConnectionContext mockContext = mock(IDatabricksConnectionContext.class);
-    when(mockContext.getEnableMultipleCatalogSupport()).thenReturn(true);
-    when(mockClient.getConnectionContext()).thenReturn(mockContext);
-
     DatabricksMetadataQueryClient metadataClient = new DatabricksMetadataQueryClient(mockClient);
     when(mockClient.executeStatement(
             eq(