Fix enable multiple catalog support to match existing driver behaviour (databricks#1181)

## Description
<!-- Provide a brief summary of the changes made and the issue they aim
to address.-->
Modified the logic for enableMultipleCatalogSupport parameter to only
return results when the catalog provided in metadata calls is null or is
equal to the current catalog when the param is disabled. This matches
the behaviour with existing driver.
## Testing
<!-- Describe how the changes have been tested-->
Tested locally 
## Additional Notes to the Reviewer
<!-- Share any additional context or insights that may help the reviewer
understand the changes better. This could include challenges faced,
limitations, or compromises made during the development process.
Also, mention any areas of the code that you would like the reviewer to
focus on specifically. -->

Author: msrathore-db

NEXT_CHANGELOG.md

@@ -12,6 +12,7 @@
 - Geospatial column type names now include SRID information (e.g., `GEOMETRY(4326)` instead of `GEOMETRY`).
 - Changed default value of `IgnoreTransactions` from `0` to `1` to disable multi-statement transactions by default. Preview participants can opt-in by setting `IgnoreTransactions=0`. Also updated `supportsTransactions()` to respect this flag.
 - Implemented lazy loading for inline Arrow results, fetching arrow batches on demand instead of all at once. This improves memory usage and initial response time for large result sets when using the Thrift protocol with Arrow format.
+- **Enhanced `enableMultipleCatalogSupport` behavior**: When this parameter is disabled (`enableMultipleCatalogSupport=0`), metadata operations (such as `getSchemas()`, `getTables()`, `getColumns()`, etc.) now return results only when the catalog parameter is either `null` or matches the current catalog. For any other catalog name, an empty result set is returned. This ensures metadata queries are restricted to the current catalog context. When enabled (`enableMultipleCatalogSupport=1`), metadata operations continue to work across all accessible catalogs.
 
 ### Fixed
 - Fixed complex data type metadata support when retrieving 0 rows in Arrow format

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

@@ -11,10 +11,13 @@
 
 import com.databricks.jdbc.api.impl.DatabricksResultSet;
 import com.databricks.jdbc.api.internal.IDatabricksConnectionContext;
+import com.databricks.jdbc.api.internal.IDatabricksSession;
 import com.databricks.jdbc.common.CommandName;
 import com.databricks.jdbc.common.Nullable;
 import com.databricks.jdbc.common.StatementType;
 import com.databricks.jdbc.exception.DatabricksSQLException;
+import com.databricks.jdbc.log.JdbcLogger;
+import com.databricks.jdbc.log.JdbcLoggerFactory;
 import com.databricks.jdbc.model.core.ColumnMetadata;
 import com.databricks.jdbc.model.core.ResultColumn;
 import com.databricks.jdbc.model.core.StatementStatus;
@@ -29,6 +32,8 @@
 import java.util.stream.Collectors;
 
 public class MetadataResultSetBuilder {
+  private static final JdbcLogger LOGGER =
+      JdbcLoggerFactory.getLogger(MetadataResultSetBuilder.class);
   private static final IDatabricksResultSetAdapter defaultAdapter =
       new DefaultDatabricksResultSetAdapter();
   private static final IDatabricksResultSetAdapter importedKeysAdapter =
@@ -39,6 +44,31 @@ public MetadataResultSetBuilder(IDatabricksConnectionContext ctx) {
     this.ctx = ctx;
   }
 
+  public boolean shouldAllowCatalogAccess(
+      String catalog, String currentCatalog, IDatabricksSession session) throws SQLException {
+    if (ctx == null || ctx.getEnableMultipleCatalogSupport()) {
+      return true;
+    }
+
+    if (catalog == null) {
+      return true;
+    }
+
+    if (currentCatalog == null) {
+      currentCatalog = session.getCurrentCatalog();
+    }
+
+    if (currentCatalog != null && currentCatalog.equals(catalog)) {
+      return true;
+    }
+
+    LOGGER.debug(
+        "Catalog access denied for catalog '{}' when enableMultipleCatalogSupport=false. Current catalog is '{}'",
+        catalog,
+        currentCatalog);
+    return false;
+  }
+
   public DatabricksResultSet getFunctionsResult(DatabricksResultSet resultSet, String catalog)
       throws SQLException {
     List<List<Object>> rows = getRowsForFunctions(resultSet, FUNCTION_COLUMNS, catalog);

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

@@ -1,13 +1,11 @@
 package com.databricks.jdbc.dbclient.impl.sqlexec;
 
 import static com.databricks.jdbc.common.MetadataResultConstants.*;
-import static com.databricks.jdbc.dbclient.impl.common.CommandConstants.GET_TABLES_STATEMENT_ID;
 import static com.databricks.jdbc.dbclient.impl.common.CommandConstants.METADATA_STATEMENT_ID;
 import static com.databricks.jdbc.dbclient.impl.sqlexec.ResultConstants.TYPE_INFO_RESULT;
 
 import com.databricks.jdbc.api.impl.DatabricksResultSet;
 import com.databricks.jdbc.api.internal.IDatabricksSession;
-import com.databricks.jdbc.common.MetadataResultConstants;
 import com.databricks.jdbc.common.StatementType;
 import com.databricks.jdbc.common.util.JdbcThreadUtils;
 import com.databricks.jdbc.common.util.WildcardUtil;
@@ -54,8 +52,11 @@ public DatabricksResultSet listCatalogs(IDatabricksSession session) throws SQLEx
     // If multiple catalog support is disabled, return only the current catalog
     if (isMultipleCatalogSupportDisabled()) {
       String currentCatalog = session.getCurrentCatalog();
-      if (currentCatalog == null) {
-        currentCatalog = "";
+      if (currentCatalog == null || currentCatalog.isEmpty()) {
+        currentCatalog = "spark";
+        LOGGER.debug(
+            "Current catalog is null or empty when multiple catalog support is disabled. Using default catalog: {}",
+            currentCatalog);
       }
       String SQL = String.format("SELECT '%s' AS catalog", currentCatalog);
       LOGGER.debug("SQL command to fetch catalogs: {}", SQL);
@@ -71,7 +72,13 @@ public DatabricksResultSet listCatalogs(IDatabricksSession session) throws SQLEx
   @Override
   public DatabricksResultSet listSchemas(
       IDatabricksSession session, String catalog, String schemaNamePattern) throws SQLException {
-    catalog = autoFillCatalog(catalog, session);
+    // Only fetch currentCatalog if multiple catalog support is disabled
+    String currentCatalog = isMultipleCatalogSupportDisabled() ? session.getCurrentCatalog() : null;
+    if (!metadataResultSetBuilder.shouldAllowCatalogAccess(catalog, currentCatalog, session)) {
+      return metadataResultSetBuilder.getSchemasResult(new ArrayList<>());
+    }
+
+    catalog = autoFillCatalog(catalog, currentCatalog);
 
     // Return empty result set if catalog is an empty string
     if (catalog != null && catalog.isEmpty()) {
@@ -111,11 +118,19 @@ public DatabricksResultSet listTables(
       String tableNamePattern,
       String[] tableTypes)
       throws SQLException {
-    catalog = autoFillCatalog(catalog, session);
     String[] validatedTableTypes =
         Optional.ofNullable(tableTypes)
             .filter(types -> types.length > 0)
             .orElse(DEFAULT_TABLE_TYPES);
+
+    // Only fetch currentCatalog if multiple catalog support is disabled
+    String currentCatalog = isMultipleCatalogSupportDisabled() ? session.getCurrentCatalog() : null;
+    if (!metadataResultSetBuilder.shouldAllowCatalogAccess(catalog, currentCatalog, session)) {
+      return metadataResultSetBuilder.getTablesResult(
+          catalog, validatedTableTypes, new ArrayList<>());
+    }
+
+    catalog = autoFillCatalog(catalog, currentCatalog);
     CommandBuilder commandBuilder =
         new CommandBuilder(catalog, session)
             .setSchemaPattern(schemaNamePattern)
@@ -132,11 +147,8 @@ public DatabricksResultSet listTables(
         // Gracefully handles the case where an older DBSQL version doesn't support all catalogs in
         // the SHOW TABLES command.
         LOGGER.debug("SQL command failed with syntax error. Returning empty result set.");
-        return metadataResultSetBuilder.getResultSetWithGivenRowsAndColumns(
-            MetadataResultConstants.TABLE_COLUMNS,
-            new ArrayList<>(),
-            GET_TABLES_STATEMENT_ID,
-            com.databricks.jdbc.common.CommandName.LIST_TABLES);
+        return metadataResultSetBuilder.getTablesResult(
+            catalog, validatedTableTypes, new ArrayList<>());
       } else {
         throw e;
       }
@@ -157,7 +169,13 @@ public DatabricksResultSet listColumns(
       String tableNamePattern,
       String columnNamePattern)
       throws SQLException {
-    catalog = autoFillCatalog(catalog, session);
+    // Only fetch currentCatalog if multiple catalog support is disabled
+    String currentCatalog = isMultipleCatalogSupportDisabled() ? session.getCurrentCatalog() : null;
+    if (!metadataResultSetBuilder.shouldAllowCatalogAccess(catalog, currentCatalog, session)) {
+      return metadataResultSetBuilder.getColumnsResult(new ArrayList<>());
+    }
+
+    catalog = autoFillCatalog(catalog, currentCatalog);
 
     // Fetch columns from all catalogs if catalog is null
     if (catalog == null) {
@@ -183,7 +201,13 @@ public DatabricksResultSet listFunctions(
       String schemaNamePattern,
       String functionNamePattern)
       throws SQLException {
-    catalog = autoFillCatalog(catalog, session);
+    // Only fetch currentCatalog if multiple catalog support is disabled
+    String currentCatalog = isMultipleCatalogSupportDisabled() ? session.getCurrentCatalog() : null;
+    if (!metadataResultSetBuilder.shouldAllowCatalogAccess(catalog, currentCatalog, session)) {
+      return metadataResultSetBuilder.getFunctionsResult(catalog, new ArrayList<>());
+    }
+
+    catalog = autoFillCatalog(catalog, currentCatalog);
 
     // Fetch current catalog if catalog is null
     if (catalog == null) {
@@ -212,7 +236,13 @@ public DatabricksResultSet listFunctions(
   @Override
   public DatabricksResultSet listPrimaryKeys(
       IDatabricksSession session, String catalog, String schema, String table) throws SQLException {
-    catalog = autoFillCatalog(catalog, session);
+    // Only fetch currentCatalog if multiple catalog support is disabled
+    String currentCatalog = isMultipleCatalogSupportDisabled() ? session.getCurrentCatalog() : null;
+    if (!metadataResultSetBuilder.shouldAllowCatalogAccess(catalog, currentCatalog, session)) {
+      return metadataResultSetBuilder.getPrimaryKeysResult(new ArrayList<>());
+    }
+
+    catalog = autoFillCatalog(catalog, currentCatalog);
 
     // Return empty result set if catalog, schema, or table is null
     if (catalog == null || schema == null || table == null) {
@@ -222,7 +252,7 @@ public DatabricksResultSet listPrimaryKeys(
           schema,
           table);
       return metadataResultSetBuilder.getResultSetWithGivenRowsAndColumns(
-          MetadataResultConstants.PRIMARY_KEYS_COLUMNS,
+          PRIMARY_KEYS_COLUMNS,
           new ArrayList<>(),
           METADATA_STATEMENT_ID,
           com.databricks.jdbc.common.CommandName.LIST_PRIMARY_KEYS);
@@ -239,7 +269,14 @@ public DatabricksResultSet listPrimaryKeys(
   public DatabricksResultSet listImportedKeys(
       IDatabricksSession session, String catalog, String schema, String table) throws SQLException {
     LOGGER.debug("public ResultSet listImportedKeys() using SDK");
-    catalog = autoFillCatalog(catalog, session);
+
+    // Only fetch currentCatalog if multiple catalog support is disabled
+    String currentCatalog = isMultipleCatalogSupportDisabled() ? session.getCurrentCatalog() : null;
+    if (!metadataResultSetBuilder.shouldAllowCatalogAccess(catalog, currentCatalog, session)) {
+      return metadataResultSetBuilder.getImportedKeys(new ArrayList<>());
+    }
+
+    catalog = autoFillCatalog(catalog, currentCatalog);
 
     // Return empty result set if catalog, schema, or table is null
     if (catalog == null || schema == null || table == null) {
@@ -249,7 +286,7 @@ public DatabricksResultSet listImportedKeys(
           schema,
           table);
       return metadataResultSetBuilder.getResultSetWithGivenRowsAndColumns(
-          MetadataResultConstants.IMPORTED_KEYS_COLUMNS,
+          IMPORTED_KEYS_COLUMNS,
           new ArrayList<>(),
           METADATA_STATEMENT_ID,
           com.databricks.jdbc.common.CommandName.GET_IMPORTED_KEYS);
@@ -265,11 +302,7 @@ public DatabricksResultSet listImportedKeys(
         // This is a workaround for the issue where the SQL command fails with "syntax error at or
         // near "foreign""
         LOGGER.debug("SQL command failed with syntax error. Returning empty result set.");
-        return metadataResultSetBuilder.getResultSetWithGivenRowsAndColumns(
-            MetadataResultConstants.IMPORTED_KEYS_COLUMNS,
-            new ArrayList<>(),
-            METADATA_STATEMENT_ID,
-            com.databricks.jdbc.common.CommandName.GET_IMPORTED_KEYS);
+        return metadataResultSetBuilder.getImportedKeys(new ArrayList<>());
       } else {
         throw e;
       }
@@ -280,13 +313,17 @@ public DatabricksResultSet listImportedKeys(
   public DatabricksResultSet listExportedKeys(
       IDatabricksSession session, String catalog, String schema, String table) throws SQLException {
     LOGGER.debug("public ResultSet listExportedKeys() using SDK");
-    catalog = autoFillCatalog(catalog, session);
+
+    // Only fetch currentCatalog if multiple catalog support is disabled
+    String currentCatalog = isMultipleCatalogSupportDisabled() ? session.getCurrentCatalog() : null;
+    if (!metadataResultSetBuilder.shouldAllowCatalogAccess(catalog, currentCatalog, session)) {
+      return metadataResultSetBuilder.getExportedKeys(new ArrayList<>());
+    }
+
+    catalog = autoFillCatalog(catalog, currentCatalog);
+
     // Exported keys not tracked in DBSQL. Returning empty result set
-    return metadataResultSetBuilder.getResultSetWithGivenRowsAndColumns(
-        MetadataResultConstants.EXPORTED_KEYS_COLUMNS,
-        new ArrayList<>(),
-        METADATA_STATEMENT_ID,
-        com.databricks.jdbc.common.CommandName.GET_EXPORTED_KEYS);
+    return metadataResultSetBuilder.getExportedKeys(new ArrayList<>());
   }
 
   @Override
@@ -300,6 +337,15 @@ public DatabricksResultSet listCrossReferences(
       String foreignTable)
       throws SQLException {
     LOGGER.debug("public ResultSet listCrossReferences() using SDK");
+
+    // Only fetch currentCatalog if multiple catalog support is disabled
+    String currentCatalog = isMultipleCatalogSupportDisabled() ? session.getCurrentCatalog() : null;
+    if (!metadataResultSetBuilder.shouldAllowCatalogAccess(parentCatalog, currentCatalog, session)
+        || !metadataResultSetBuilder.shouldAllowCatalogAccess(
+            foreignCatalog, currentCatalog, session)) {
+      return metadataResultSetBuilder.getCrossRefsResult(new ArrayList<>());
+    }
+
     CommandBuilder commandBuilder =
         new CommandBuilder(foreignCatalog, session).setSchema(foreignSchema).setTable(foreignTable);
     String SQL = commandBuilder.getSQLString(CommandName.LIST_FOREIGN_KEYS);
@@ -312,11 +358,7 @@ public DatabricksResultSet listCrossReferences(
         // near "foreign""
         // This is a known issue in Databricks for older DBSQL versions
         LOGGER.debug("SQL command failed with syntax error. Returning empty result set.");
-        return metadataResultSetBuilder.getResultSetWithGivenRowsAndColumns(
-            MetadataResultConstants.CROSS_REFERENCE_COLUMNS,
-            new ArrayList<>(),
-            METADATA_STATEMENT_ID,
-            com.databricks.jdbc.common.CommandName.GET_CROSS_REFERENCE);
+        return metadataResultSetBuilder.getCrossRefsResult(new ArrayList<>());
       } else {
         LOGGER.error(
             e,
@@ -333,10 +375,20 @@ private boolean isMultipleCatalogSupportDisabled() {
         && !sdkClient.getConnectionContext().getEnableMultipleCatalogSupport();
   }
 
-  private String autoFillCatalog(String catalog, IDatabricksSession session) throws SQLException {
-    if (isMultipleCatalogSupportDisabled()) {
-      String currentCatalog = session.getCurrentCatalog();
-      return (currentCatalog != null && !currentCatalog.isEmpty()) ? currentCatalog : "";
+  /**
+   * Auto-fills the catalog parameter if multiple catalog support is disabled and catalog is null.
+   *
+   * @param catalog the catalog parameter to auto-fill
+   * @param currentCatalog the current catalog from the session
+   * @return the auto-filled catalog or the original catalog if no auto-fill is needed
+   */
+  private String autoFillCatalog(String catalog, String currentCatalog) {
+    if (isMultipleCatalogSupportDisabled() && catalog == null) {
+      String result =
+          (currentCatalog != null && !currentCatalog.isEmpty()) ? currentCatalog : "spark";
+      LOGGER.debug(
+          "Auto-filling null catalog with '{}' when multiple catalog support is disabled", result);
+      return result;
     }
     return catalog;
   }
@@ -446,7 +498,7 @@ private DatabricksResultSet fetchColumnsAcrossCatalogs(
 
     // Convert combined data into a result set
     return metadataResultSetBuilder.getResultSetWithGivenRowsAndColumns(
-        MetadataResultConstants.COLUMN_COLUMNS,
+        COLUMN_COLUMNS,
         columnRows,
         METADATA_STATEMENT_ID,
         com.databricks.jdbc.common.CommandName.LIST_COLUMNS);