fix(metadata): scope schemas to current catalog

Closes #8461

agents/common/src/main/java/com/dbx/agent/ConfiguredJdbcAgent.java

@@ -47,7 +47,7 @@ public abstract class ConfiguredJdbcAgent extends AbstractJdbcAgent {
 
     @Override
     public List<String> listSchemas() {
-        return StandardJdbcMetadata.INSTANCE.listSchemas(requireConnection(), profile);
+        return StandardJdbcMetadata.INSTANCE.listSchemas(requireConnection(), profile, configuredDatabase);
     }
 
     @Override

agents/common/src/main/java/com/dbx/agent/StandardJdbcMetadata.java

@@ -42,11 +42,23 @@ public final class StandardJdbcMetadata {
     }
 
     public List<String> listSchemas(Connection conn, JdbcAgentProfile profile) {
+        return listSchemas(conn, profile, null);
+    }
+
+    public List<String> listSchemas(Connection conn, JdbcAgentProfile profile, String configuredCatalog) {
         return unchecked(() -> {
             Set<String> names = new LinkedHashSet<>();
             DatabaseMetaData meta = conn.getMetaData();
+            String catalog = blankToNull(configuredCatalog);
             try {
-                appendSchemas(names, meta.getSchemas(null, null));
+                String currentCatalog = conn.getCatalog();
+                if (currentCatalog != null && !currentCatalog.trim().isEmpty()) {
+                    catalog = currentCatalog.trim();
+                }
+            } catch (Exception | AbstractMethodError ignored) {
+            }
+            try {
+                appendSchemas(names, meta.getSchemas(catalog, null));
             } catch (Exception | AbstractMethodError first) {
                 try {
                     appendSchemas(names, meta.getSchemas());

agents/common/src/test/java/com/dbx/agent/StandardJdbcMetadataTest.java

@@ -43,6 +43,37 @@ class StandardJdbcMetadataTest {
         assertEquals(Arrays.asList("APP", "PUBLIC"), StandardJdbcMetadata.INSTANCE.listSchemas(conn, profile));
     }
 
+    @Test
+    void scopesSchemasToTheConnectionCatalog() {
+        AtomicReference<Object[]> capturedArgs = new AtomicReference<>();
+        DatabaseMetaData meta = proxy(DatabaseMetaData.class, new MethodHandler() {
+            @Override
+            public Object handle(Method method, Object[] args) {
+                if ("getSchemas".equals(method.getName())) {
+                    capturedArgs.set(args);
+                    return rows(row("TABLE_SCHEM", "APP"));
+                }
+                return defaultValue(method.getReturnType());
+            }
+        });
+        Connection conn = proxy(Connection.class, new MethodHandler() {
+            @Override
+            public Object handle(Method method, Object[] args) {
+                if ("getMetaData".equals(method.getName())) {
+                    return meta;
+                }
+                if ("getCatalog".equals(method.getName())) {
+                    return "regular_catalog";
+                }
+                return defaultValue(method.getReturnType());
+            }
+        });
+
+        assertEquals(List.of("APP"), StandardJdbcMetadata.INSTANCE.listSchemas(conn, profile, "initial_catalog"));
+        assertEquals("regular_catalog", capturedArgs.get()[0]);
+        assertEquals(null, capturedArgs.get()[1]);
+    }
+
     @Test
     void listsSchemasWhenConnectionGetSchemaIsUnsupported() {
         Connection conn = connection(