fix(oceanbase): resolve synonym columns with target identity (#11416)

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

@@ -15,6 +15,9 @@ public final class ColumnInfo {
     private Integer character_maximum_length;
     private String character_set;
     private String collation;
+    private String resolved_schema;
+    private String resolved_table;
+    private String resolved_object_type;
 
     public ColumnInfo() {
         this("", "", true, null, false);
@@ -128,6 +131,18 @@ public final class ColumnInfo {
         return collation;
     }
 
+    public String getResolved_schema() {
+        return resolved_schema;
+    }
+
+    public String getResolved_table() {
+        return resolved_table;
+    }
+
+    public String getResolved_object_type() {
+        return resolved_object_type;
+    }
+
     public void setName(String name) {
         this.name = name;
     }
@@ -176,6 +191,18 @@ public final class ColumnInfo {
         this.collation = collation;
     }
 
+    public void setResolved_schema(String resolved_schema) {
+        this.resolved_schema = resolved_schema;
+    }
+
+    public void setResolved_table(String resolved_table) {
+        this.resolved_table = resolved_table;
+    }
+
+    public void setResolved_object_type(String resolved_object_type) {
+        this.resolved_object_type = resolved_object_type;
+    }
+
     @Override
     public boolean equals(Object other) {
         if (this == other) return true;
@@ -192,7 +219,10 @@ public final class ColumnInfo {
             && Objects.equals(numeric_scale, that.numeric_scale)
             && Objects.equals(character_maximum_length, that.character_maximum_length)
             && Objects.equals(character_set, that.character_set)
-            && Objects.equals(collation, that.collation);
+            && Objects.equals(collation, that.collation)
+            && Objects.equals(resolved_schema, that.resolved_schema)
+            && Objects.equals(resolved_table, that.resolved_table)
+            && Objects.equals(resolved_object_type, that.resolved_object_type);
     }
 
     @Override
@@ -209,7 +239,10 @@ public final class ColumnInfo {
             numeric_scale,
             character_maximum_length,
             character_set,
-            collation
+            collation,
+            resolved_schema,
+            resolved_table,
+            resolved_object_type
         );
     }
 
@@ -227,6 +260,9 @@ public final class ColumnInfo {
             + ", character_maximum_length=" + character_maximum_length
             + ", character_set=" + character_set
             + ", collation=" + collation
+            + ", resolved_schema=" + resolved_schema
+            + ", resolved_table=" + resolved_table
+            + ", resolved_object_type=" + resolved_object_type
             + ")";
     }
 }

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

@@ -86,6 +86,10 @@ public interface DatabaseAgent {
 
     List<ColumnInfo> getColumns(String schema, String table);
 
+    default List<ColumnInfo> getColumnsInContext(String schema, String table, String currentSchema) {
+        return getColumns(schema, table);
+    }
+
     default ObjectSource getObjectSource(String schema, String name, String objectType) {
         throw new UnsupportedOperationException("Object source is not supported");
     }

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

@@ -265,7 +265,9 @@ public final class JsonRpcServer {
         }
         if (AgentProtocol.METHOD_GET_COLUMNS.equals(method)) {
             switchCatalog(params);
-            return agent.getColumns(params.get("schema").getAsString(), params.get("table").getAsString());
+            String currentSchema = params.has("current_schema") && !params.get("current_schema").isJsonNull()
+                ? params.get("current_schema").getAsString() : null;
+            return agent.getColumnsInContext(params.get("schema").getAsString(), params.get("table").getAsString(), currentSchema);
         }
         if (AgentProtocol.METHOD_LIST_INDEXES.equals(method)) {
             switchCatalog(params);

agents/drivers/oceanbase-oracle/src/main/java/com/dbx/agent/oceanbaseoracle/OceanBaseOracleAgent.java

@@ -907,6 +907,142 @@ public final class OceanBaseOracleAgent extends ConfiguredJdbcAgent {
 
     @Override
     public List<ColumnInfo> getColumns(String schema, String table) {
+        return getColumnsInContext(schema, table, null);
+    }
+
+    @Override
+    public List<ColumnInfo> getColumnsInContext(String schema, String table, String currentSchema) {
+        return unchecked(() -> {
+            String owner = normalizeSchema(schema == null || schema.isBlank() ? currentSchema : schema);
+            String tableName = normalizeObjectName(table);
+            String initialOwner = owner;
+            String initialName = tableName;
+            boolean publicFallback = false;
+            Set<List<String>> visited = new HashSet<>();
+            for (int depth = 0; depth <= 32; depth++) {
+                if (!visited.add(List.of(owner, tableName))) {
+                    throw new SQLException("Synonym cycle detected at " + owner + "." + tableName);
+                }
+                List<ColumnInfo> columns = getObjectColumns(owner, tableName);
+                if (!columns.isEmpty()) {
+                    if (depth > 0) {
+                        if (publicFallback) verifyPublicColumnTarget(initialOwner, initialName, owner, tableName);
+                        String objectType = queryColumnObjectType(owner, tableName);
+                        for (ColumnInfo column : columns) {
+                            column.setResolved_schema(owner);
+                            column.setResolved_table(tableName);
+                            column.setResolved_object_type(objectType);
+                        }
+                    }
+                    return columns;
+                }
+                String objectType = queryColumnObjectType(owner, tableName);
+                if (objectType != null && !"SYNONYM".equals(objectType)) {
+                    throw new SQLException("Column metadata for " + owner + "." + tableName + " (" + objectType + ") is not accessible or not available");
+                }
+                SynonymTarget target = queryColumnSynonym(owner, tableName);
+                if (target == null && objectType == null && depth == 0 && (schema == null || schema.isBlank())) {
+                    target = queryColumnSynonym("PUBLIC", tableName);
+                    // OceanBase 4.2.5 exposes public synonyms under its internal dictionary owner.
+                    if (target == null) target = queryColumnSynonym("__public", tableName);
+                    publicFallback = target != null;
+                }
+                if (target == null) {
+                    throw new SQLException("Object " + owner + "." + tableName + " not found or not accessible for column metadata");
+                }
+                if (target.databaseLink() != null && !target.databaseLink().isBlank()) {
+                    throw new SQLException("Column metadata for DBLink synonyms is not supported: " + owner + "." + tableName);
+                }
+                if (target.owner() == null || target.owner().isBlank() || target.name() == null || target.name().isBlank()) {
+                    throw new SQLException("Incomplete synonym target metadata for " + owner + "." + tableName);
+                }
+                owner = target.owner();
+                tableName = target.name();
+            }
+            throw new SQLException("Synonym resolution exceeded 32 links");
+        });
+    }
+
+    private record SynonymTarget(String owner, String name, String databaseLink) {}
+
+    private void verifyPublicColumnTarget(String owner, String name, String targetOwner, String targetName) throws SQLException {
+        // ALL_OBJECTS can hide a foreign-schema object that still shadows a public synonym.
+        // NAME_RESOLVE stops at DBLinks; no remote resolution is performed here.
+        String originalSchema = currentSchema();
+        if (originalSchema.isBlank()) throw new SQLException("Cannot determine current schema for public synonym resolution");
+        boolean switchSchema = !owner.equals(originalSchema);
+        Exception primaryError = null;
+        try {
+            if (switchSchema) {
+                try (var stmt = requireConnection().createStatement()) {
+                    stmt.execute(setSchemaSQL(owner));
+                }
+            }
+            try (var stmt = requireConnection().prepareCall("BEGIN DBMS_UTILITY.NAME_RESOLVE(?, ?, ?, ?, ?, ?, ?, ?); END;")) {
+                stmt.setString(1, quoteIdentifier(name));
+                stmt.setInt(2, 0);
+                for (int index = 3; index <= 6; index++) stmt.registerOutParameter(index, Types.VARCHAR);
+                stmt.registerOutParameter(7, Types.NUMERIC);
+                stmt.registerOutParameter(8, Types.NUMERIC);
+                stmt.execute();
+                String databaseLink = stmt.getString(6);
+                if (databaseLink != null && !databaseLink.isBlank()) {
+                    throw new SQLException("Column metadata for DBLink synonyms is not supported: " + owner + "." + name);
+                }
+                if (!targetOwner.equals(stmt.getString(3)) || !targetName.equals(stmt.getString(4))) {
+                    throw new SQLException("Public synonym target does not match database name resolution for " + owner + "." + name
+                        + "; a local object may be inaccessible");
+                }
+            }
+        } catch (SQLException | RuntimeException error) {
+            primaryError = error;
+            throw error;
+        } finally {
+            if (switchSchema) {
+                try (var stmt = requireConnection().createStatement()) {
+                    stmt.execute(setSchemaSQL(originalSchema));
+                } catch (SQLException restoreError) {
+                    if (primaryError == null) throw restoreError;
+                    primaryError.addSuppressed(restoreError);
+                }
+            }
+        }
+    }
+
+    private SynonymTarget queryColumnSynonym(String owner, String name) throws SQLException {
+        String sql = "SELECT TABLE_OWNER, TABLE_NAME, DB_LINK FROM ALL_SYNONYMS WHERE OWNER = ? AND SYNONYM_NAME = ?";
+        try (var stmt = requireConnection().prepareStatement(sql)) {
+            stmt.setString(1, owner);
+            stmt.setString(2, name);
+            try (ResultSet rs = stmt.executeQuery()) {
+                return rs.next() ? new SynonymTarget(rs.getString("TABLE_OWNER"), rs.getString("TABLE_NAME"), rs.getString("DB_LINK")) : null;
+            }
+        }
+    }
+
+    private String queryColumnObjectType(String owner, String name) throws SQLException {
+        String sql = """
+            SELECT OBJECT_TYPE FROM ALL_OBJECTS
+            WHERE OWNER = ? AND OBJECT_NAME = ?
+              AND OBJECT_TYPE IN ('TABLE', 'VIEW', 'MATERIALIZED VIEW', 'SYNONYM',
+                'SEQUENCE', 'PROCEDURE', 'FUNCTION', 'PACKAGE', 'TYPE')
+            """;
+        try (var stmt = requireConnection().prepareStatement(sql)) {
+            stmt.setString(1, owner);
+            stmt.setString(2, name);
+            try (ResultSet rs = stmt.executeQuery()) {
+                String type = null;
+                while (rs.next()) {
+                    String candidate = rs.getString("OBJECT_TYPE");
+                    if ("MATERIALIZED VIEW".equals(candidate)) return candidate;
+                    if (type == null || "VIEW".equals(candidate)) type = candidate;
+                }
+                return type;
+            }
+        }
+    }
+
+    private List<ColumnInfo> getObjectColumns(String schema, String table) {
         return unchecked(() -> {
             String owner = normalizeSchema(schema);
             String tableName = normalizeObjectName(table);

agents/drivers/oceanbase-oracle/src/test/java/com/dbx/agent/oceanbaseoracle/OceanBaseOracleSynonymColumnsTest.java

@@ -0,0 +1,489 @@
+package com.dbx.agent.oceanbaseoracle;
+
+import com.dbx.agent.ColumnInfo;
+import com.dbx.agent.test.TestSupport;
+import org.junit.jupiter.api.Test;
+
+import java.lang.reflect.InvocationHandler;
+import java.lang.reflect.Proxy;
+import java.sql.Connection;
+import java.sql.CallableStatement;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+class OceanBaseOracleSynonymColumnsTest {
+    @Test
+    void privateSynonymReturnsTargetColumnMetadataAcrossSchemas() {
+        var catalog = new Catalog();
+        catalog.synonym("APP", "ORDER_ALIAS", "SALES", "ORDERS", null);
+        catalog.object("SALES", "ORDERS", "TABLE");
+        catalog.columns("SALES", "ORDERS", amountColumn());
+
+        List<ColumnInfo> columns = catalog.agent.getColumns("APP", "ORDER_ALIAS");
+
+        assertEquals(1, columns.size());
+        ColumnInfo column = columns.getFirst();
+        assertEquals("AMOUNT", column.getName());
+        assertEquals("NUMBER(19,4)", column.getData_type());
+        assertEquals(19, column.getNumeric_precision());
+        assertEquals(4, column.getNumeric_scale());
+        assertFalse(column.getIs_nullable());
+        assertTrue(column.getIs_primary_key());
+        assertEquals("0", column.getColumn_default());
+        assertEquals("Order amount", column.getComment());
+    }
+
+    private static Map<String, Object> amountColumn() {
+        return Map.of("COLUMN_NAME", "AMOUNT", "DATA_TYPE", "NUMBER", "NULLABLE", "N",
+            "DATA_PRECISION", 19, "DATA_SCALE", 4, "DATA_LENGTH", 22,
+            "DATA_DEFAULT", "0", "COMMENTS", "Order amount", "IS_PK", 1);
+    }
+
+    @Test
+    void synonymChainRetainsQuotedTargetNamesAndVerifiedViewIdentity() {
+        var catalog = new Catalog();
+        catalog.synonym("APP", "ORDER_ALIAS", "Bridge.Owner", "Next.Alias", null);
+        catalog.synonym("Bridge.Owner", "Next.Alias", "Sales.Owner", "Order\"View", null);
+        catalog.object("Sales.Owner", "Order\"View", "VIEW");
+        catalog.columns("Sales.Owner", "Order\"View", amountColumn());
+
+        ColumnInfo column = catalog.agent.getColumns("APP", "ORDER_ALIAS").getFirst();
+
+        assertEquals("Sales.Owner", column.getResolved_schema());
+        assertEquals("Order\"View", column.getResolved_table());
+        assertEquals("VIEW", column.getResolved_object_type());
+        assertEquals("NUMBER(19,4)", column.getData_type());
+        assertFalse(column.getIs_nullable());
+        assertTrue(column.getIs_primary_key());
+    }
+
+    private record Name(String owner, String name) {}
+
+    @Test
+    void publicSynonymResolvesOnlyWhenSchemaIsNotExplicit() {
+        var catalog = new Catalog();
+        catalog.synonym("PUBLIC", "PUBLIC_ORDERS", "SALES", "ORDERS", null);
+        catalog.object("SALES", "ORDERS", "TABLE");
+        catalog.columns("SALES", "ORDERS", amountColumn());
+        catalog.nativeResult("APP", "PUBLIC_ORDERS", "SALES", "ORDERS", null);
+
+        ColumnInfo column = catalog.agent.getColumns("", "PUBLIC_ORDERS").getFirst();
+        assertEquals("SALES", column.getResolved_schema());
+        assertEquals("ORDERS", column.getResolved_table());
+        assertEquals("TABLE", column.getResolved_object_type());
+        assertTrue(catalog.sql.stream().noneMatch(sql -> sql.startsWith("ALTER SESSION")));
+        RuntimeException error = assertThrows(RuntimeException.class,
+            () -> catalog.agent.getColumns("APP", "PUBLIC_ORDERS"));
+        assertTrue(error.getMessage().contains("not found or not accessible"), error.getMessage());
+    }
+
+    @Test
+    void publicSynonymSupportsTheOceanBaseDictionaryPublicOwner() {
+        var catalog = new Catalog();
+        catalog.synonym("__public", "PUBLIC_ORDERS", "SALES", "ORDERS", null);
+        catalog.object("SALES", "ORDERS", "TABLE");
+        catalog.columns("SALES", "ORDERS", amountColumn());
+        catalog.nativeResult("APP", "PUBLIC_ORDERS", "SALES", "ORDERS", null);
+
+        ColumnInfo column = catalog.agent.getColumns("", "PUBLIC_ORDERS").getFirst();
+
+        assertEquals("SALES", column.getResolved_schema());
+        assertEquals("ORDERS", column.getResolved_table());
+        assertEquals("TABLE", column.getResolved_object_type());
+        assertThrows(RuntimeException.class, () -> catalog.agent.getColumns("APP", "PUBLIC_ORDERS"));
+    }
+
+    @Test
+    void selectedSchemaContextResolvesLocalNamesAndStillAllowsPublicFallback() {
+        var catalog = new Catalog();
+        catalog.columns("Reporting", "LOCAL_ORDERS", amountColumn());
+        catalog.synonym("PUBLIC", "PUBLIC_ORDERS", "SALES", "ORDERS", null);
+        catalog.object("SALES", "ORDERS", "TABLE");
+        catalog.columns("SALES", "ORDERS", amountColumn());
+        catalog.nativeResult("Reporting", "PUBLIC_ORDERS", "SALES", "ORDERS", null);
+
+        assertEquals("AMOUNT", catalog.agent.getColumnsInContext("", "LOCAL_ORDERS", "Reporting").getFirst().getName());
+        assertEquals("SALES", catalog.agent.getColumnsInContext("", "PUBLIC_ORDERS", "Reporting").getFirst().getResolved_schema());
+        assertEquals("APP", catalog.currentSchema);
+        assertThrows(RuntimeException.class, () -> catalog.agent.getColumnsInContext("Reporting", "PUBLIC_ORDERS", "APP"));
+    }
+
+    @Test
+    void localNamespaceObjectsShadowPublicSynonymsEvenWhenTheyHaveNoColumns() {
+        for (String type : List.of("TABLE", "VIEW", "SEQUENCE", "PROCEDURE", "FUNCTION", "PACKAGE", "TYPE")) {
+            var catalog = new Catalog();
+            catalog.object("APP", "ORDERS", type);
+            catalog.synonym("PUBLIC", "ORDERS", "SALES", "ORDERS", null);
+            catalog.object("SALES", "ORDERS", "TABLE");
+            catalog.columns("SALES", "ORDERS", amountColumn());
+
+            assertThrows(RuntimeException.class, () -> catalog.agent.getColumns("", "ORDERS"), type);
+            assertFalse(catalog.columnReads.contains(new Name("SALES", "ORDERS")), type);
+        }
+    }
+
+    @Test
+    void remoteSynonymIsRejectedBeforeAnyTargetQuery() {
+        var catalog = new Catalog();
+        catalog.synonym("APP", "REMOTE_ORDERS", "SALES", "ORDERS", "REMOTE.EXAMPLE");
+        catalog.columns("SALES", "ORDERS", amountColumn());
+
+        RuntimeException error = assertThrows(RuntimeException.class,
+            () -> catalog.agent.getColumns("APP", "REMOTE_ORDERS"));
+
+        assertTrue(error.getMessage().contains("DBLink"), error.getMessage());
+        assertTrue(error.getMessage().contains("not supported"), error.getMessage());
+        assertEquals(List.of(new Name("APP", "REMOTE_ORDERS")), catalog.columnReads);
+        assertTrue(catalog.sql.stream().noneMatch(sql -> sql.contains("@")));
+    }
+
+    @Test
+    void cyclicSynonymsFailWithAnExplicitCycleError() {
+        var catalog = new Catalog();
+        catalog.synonym("APP", "A", "OTHER", "B", null);
+        catalog.synonym("OTHER", "B", "APP", "A", null);
+
+        RuntimeException error = assertThrows(RuntimeException.class, () -> catalog.agent.getColumns("APP", "A"));
+
+        assertTrue(error.getMessage().contains("cycle"), error.getMessage());
+        assertEquals(List.of(new Name("APP", "A"), new Name("OTHER", "B")), catalog.columnReads);
+    }
+
+    @Test
+    void incompleteSynonymTargetMetadataNeverFallsBackToTheSessionOwner() {
+        for (String[] target : new String[][]{{null, "ORDERS"}, {"SALES", null}, {"", "ORDERS"}, {"SALES", ""}}) {
+            var catalog = new Catalog();
+            catalog.synonym("APP", "BROKEN", target[0], target[1], null);
+            catalog.columns("APP", "ORDERS", amountColumn());
+
+            RuntimeException error = assertThrows(RuntimeException.class, () -> catalog.agent.getColumns("APP", "BROKEN"));
+
+            assertTrue(error.getMessage() != null && error.getMessage().contains("Incomplete synonym target metadata"), String.valueOf(error));
+            assertEquals(List.of(new Name("APP", "BROKEN")), catalog.columnReads);
+        }
+    }
+
+    @Test
+    void materializedViewIdentityTakesPrecedenceOverItsTableEntry() {
+        var catalog = new Catalog();
+        catalog.synonym("APP", "MV_ALIAS", "SALES", "ORDER_TOTALS", null);
+        catalog.objects.put(new Name("SALES", "ORDER_TOTALS"), List.of(
+            Map.of("OBJECT_TYPE", "TABLE"), Map.of("OBJECT_TYPE", "MATERIALIZED VIEW")));
+        catalog.columns("SALES", "ORDER_TOTALS", amountColumn());
+
+        assertEquals("MATERIALIZED VIEW", catalog.agent.getColumns("APP", "MV_ALIAS").getFirst().getResolved_object_type());
+    }
+
+    @Test
+    void unknownTargetTypeIsNotPromotedToTable() {
+        var catalog = new Catalog();
+        catalog.synonym("APP", "ORDER_ALIAS", "SALES", "ORDERS", null);
+        catalog.columns("SALES", "ORDERS", amountColumn());
+
+        ColumnInfo column = catalog.agent.getColumns("APP", "ORDER_ALIAS").getFirst();
+
+        assertEquals("SALES", column.getResolved_schema());
+        assertEquals("ORDERS", column.getResolved_table());
+        assertNull(column.getResolved_object_type());
+    }
+
+    @Test
+    void localTableAndPrivateSynonymTakePrecedenceOverPublicSynonym() {
+        var catalog = new Catalog();
+        catalog.synonym("PUBLIC", "ORDERS", "OTHER", "ORDERS", null);
+        catalog.synonym("PUBLIC", "ORDER_ALIAS", "OTHER", "ORDERS", null);
+        catalog.synonym("APP", "ORDER_ALIAS", "APP", "ORDERS", null);
+        catalog.object("APP", "ORDERS", "TABLE");
+        catalog.columns("APP", "ORDERS", amountColumn());
+
+        assertEquals("AMOUNT", catalog.agent.getColumns("", "ORDERS").getFirst().getName());
+        assertEquals("APP", catalog.agent.getColumns("", "ORDER_ALIAS").getFirst().getResolved_schema());
+        assertFalse(catalog.columnReads.contains(new Name("OTHER", "ORDERS")));
+    }
+
+    @Test
+    void brokenPrivateTargetDoesNotFallThroughToPublicSynonyms() {
+        var catalog = new Catalog();
+        catalog.synonym("APP", "ORDERS", "SALES", "MISSING", null);
+        catalog.synonym("PUBLIC", "ORDERS", "OTHER", "ORDERS", null);
+        catalog.synonym("PUBLIC", "MISSING", "OTHER", "ORDERS", null);
+        catalog.columns("OTHER", "ORDERS", amountColumn());
+
+        RuntimeException error = assertThrows(RuntimeException.class, () -> catalog.agent.getColumns("", "ORDERS"));
+
+        assertTrue(error.getMessage().contains("SALES.MISSING not found or not accessible"), error.getMessage());
+        assertFalse(catalog.columnReads.contains(new Name("OTHER", "ORDERS")));
+    }
+
+    @Test
+    void missingAndInaccessibleTargetsAreNotSuccessfulEmptyColumns() {
+        var catalog = new Catalog();
+        catalog.synonym("APP", "INACCESSIBLE", "SALES", "HIDDEN", null);
+
+        for (String name : List.of("MISSING", "INACCESSIBLE")) {
+            RuntimeException error = assertThrows(RuntimeException.class, () -> catalog.agent.getColumns("APP", name));
+            assertTrue(error.getMessage().contains("not found or not accessible"), error.getMessage());
+        }
+    }
+
+    @Test
+    void dictionaryFailuresPreserveTheOriginalSqlException() {
+        for (String view : List.of("ALL_TAB_COLUMNS", "ALL_OBJECTS", "ALL_SYNONYMS")) {
+            var catalog = new Catalog();
+            catalog.failure = new SQLException("insufficient privileges", "42000", 1031);
+            catalog.failureView = view;
+
+            RuntimeException error = assertThrows(RuntimeException.class, () -> catalog.agent.getColumns("APP", "ORDERS"));
+
+            assertSame(catalog.failure, error.getCause(), view);
+        }
+    }
+
+    @Test
+    void longAcyclicChainHasABoundedResolutionLimit() {
+        var catalog = new Catalog();
+        for (int index = 0; index < 40; index++) {
+            catalog.synonym("APP", "S" + index, "APP", "S" + (index + 1), null);
+        }
+        catalog.columns("APP", "S40", amountColumn());
+
+        RuntimeException error = assertThrows(RuntimeException.class, () -> catalog.agent.getColumns("APP", "S0"));
+
+        assertTrue(error.getMessage().contains("exceeded 32 links"), error.getMessage());
+        assertTrue(catalog.columnReads.size() <= 33);
+        assertFalse(catalog.columnReads.contains(new Name("APP", "S40")));
+    }
+
+    @Test
+    void invisibleForeignLocalObjectCannotBeReplacedByAPublicTarget() {
+        var catalog = new Catalog();
+        catalog.synonym("__public", "PUBLIC_ORDERS", "SALES", "ORDERS", null);
+        catalog.object("SALES", "ORDERS", "TABLE");
+        catalog.columns("SALES", "ORDERS", amountColumn());
+        catalog.nativeResult("Reporting", "PUBLIC_ORDERS", "Reporting", "PUBLIC_ORDERS", null);
+
+        RuntimeException error = assertThrows(RuntimeException.class,
+            () -> catalog.agent.getColumnsInContext("", "PUBLIC_ORDERS", "Reporting"));
+
+        assertTrue(error.getMessage().contains("does not match"), error.getMessage());
+        assertEquals("APP", catalog.currentSchema);
+    }
+
+    private record NativeTarget(String owner, String name, String databaseLink) {}
+
+    @Test
+    void nativePublicResolutionFailureRestoresTheExactOriginalSchema() {
+        var catalog = publicCatalog();
+        catalog.currentSchema = "Original.Owner";
+        catalog.nativeFailure = new SQLException("NAME_RESOLVE not available", "42000", 6550);
+
+        RuntimeException error = assertThrows(RuntimeException.class,
+            () -> catalog.agent.getColumnsInContext("", "PUBLIC_ORDERS", "Reporting"));
+
+        assertSame(catalog.nativeFailure, error.getCause());
+        assertEquals("Original.Owner", catalog.currentSchema);
+    }
+
+    @Test
+    void nativePublicResolutionRejectsRemoteTargetsWithoutRemoteQueries() {
+        var catalog = publicCatalog();
+        catalog.nativeResult("Reporting", "PUBLIC_ORDERS", "SALES", "ORDERS", "REMOTE.EXAMPLE");
+
+        RuntimeException error = assertThrows(RuntimeException.class,
+            () -> catalog.agent.getColumnsInContext("", "PUBLIC_ORDERS", "Reporting"));
+
+        assertTrue(error.getMessage().contains("DBLink"), error.getMessage());
+        assertEquals("APP", catalog.currentSchema);
+        assertTrue(catalog.sql.stream().noneMatch(sql -> sql.contains("@")));
+    }
+
+    @Test
+    void schemaRestoreFailureCannotReturnSuccessfulColumns() {
+        var catalog = publicCatalog();
+        catalog.restoreFailure = new SQLException("restore failed");
+
+        RuntimeException error = assertThrows(RuntimeException.class,
+            () -> catalog.agent.getColumnsInContext("", "PUBLIC_ORDERS", "Reporting"));
+
+        assertSame(catalog.restoreFailure, error.getCause());
+    }
+
+    @Test
+    void schemaRestoreFailureIsSuppressedOnTheOriginalResolutionError() {
+        var catalog = publicCatalog();
+        catalog.nativeFailure = new SQLException("resolution failed");
+        catalog.restoreFailure = new SQLException("restore failed");
+
+        RuntimeException error = assertThrows(RuntimeException.class,
+            () -> catalog.agent.getColumnsInContext("", "PUBLIC_ORDERS", "Reporting"));
+
+        assertSame(catalog.nativeFailure, error.getCause());
+        assertArrayEquals(new Throwable[]{catalog.restoreFailure}, error.getCause().getSuppressed());
+    }
+
+    private static Catalog publicCatalog() {
+        var catalog = new Catalog();
+        catalog.synonym("__public", "PUBLIC_ORDERS", "SALES", "ORDERS", null);
+        catalog.object("SALES", "ORDERS", "TABLE");
+        catalog.columns("SALES", "ORDERS", amountColumn());
+        catalog.nativeResult("Reporting", "PUBLIC_ORDERS", "SALES", "ORDERS", null);
+        return catalog;
+    }
+
+    private static final class Catalog {
+        final OceanBaseOracleAgent agent = new OceanBaseOracleAgent();
+        final Map<Name, List<Map<String, Object>>> columns = new HashMap<>();
+        final Map<Name, List<Map<String, Object>>> synonyms = new HashMap<>();
+        final Map<Name, List<Map<String, Object>>> objects = new HashMap<>();
+        final Map<Name, NativeTarget> nativeTargets = new HashMap<>();
+        final List<String> sql = new ArrayList<>();
+        final List<Name> columnReads = new ArrayList<>();
+        String currentSchema = "APP";
+        SQLException failure;
+        SQLException nativeFailure;
+        SQLException restoreFailure;
+        String failureView;
+
+        Catalog() {
+            Connection connection = proxy(Connection.class, (ignored, method, args) -> {
+                if (method.getName().equals("prepareStatement")) return statement((String) args[0]);
+                if (method.getName().equals("prepareCall")) return nativeStatement((String) args[0]);
+                if (method.getName().equals("createStatement")) {
+                    return proxy(Statement.class, (stmt, call, params) -> {
+                        if (call.getName().equals("executeQuery")) {
+                            sql.add((String) params[0]);
+                            return rows(List.of(Map.of("CURRENT_SCHEMA", currentSchema)));
+                        }
+                        if (call.getName().equals("execute")) {
+                            String query = (String) params[0];
+                            sql.add(query);
+                            String prefix = "ALTER SESSION SET CURRENT_SCHEMA = ";
+                            assertTrue(query.startsWith(prefix), query);
+                            String identifier = query.substring(prefix.length());
+                            if (restoreFailure != null && identifier.equals("\"APP\"")) throw restoreFailure;
+                            currentSchema = identifier.substring(1, identifier.length() - 1).replace("\"\"", "\"");
+                            return false;
+                        }
+                        return defaultValue(call.getReturnType());
+                    });
+                }
+                return defaultValue(method.getReturnType());
+            });
+            TestSupport.setPrivateConnection(agent, connection);
+        }
+
+        void synonym(String owner, String name, String targetOwner, String targetName, String link) {
+            var row = new HashMap<String, Object>();
+            row.put("TABLE_OWNER", targetOwner);
+            row.put("TABLE_NAME", targetName);
+            row.put("DB_LINK", link);
+            synonyms.put(new Name(owner, name), List.of(row));
+            object(owner, name, "SYNONYM");
+        }
+
+        void object(String owner, String name, String type) {
+            objects.put(new Name(owner, name), List.of(Map.of("OBJECT_TYPE", type)));
+        }
+
+        void columns(String owner, String name, Map<String, Object> column) {
+            columns.put(new Name(owner, name), List.of(column));
+        }
+
+        void nativeResult(String owner, String name, String targetOwner, String targetName, String link) {
+            nativeTargets.put(new Name(owner, name), new NativeTarget(targetOwner, targetName, link));
+        }
+
+        private CallableStatement nativeStatement(String query) {
+            sql.add(query);
+            Map<Integer, Object> parameters = new HashMap<>();
+            NativeTarget[] result = {null};
+            return proxy(CallableStatement.class, (ignored, method, args) -> {
+                if (method.getName().equals("setString") || method.getName().equals("setInt")) {
+                    parameters.put((Integer) args[0], args[1]);
+                    return null;
+                }
+                if (method.getName().equals("execute")) {
+                    assertEquals(0, parameters.get(2), "OceanBase table resolution requires context 0");
+                    if (nativeFailure != null) throw nativeFailure;
+                    String identifier = (String) parameters.get(1);
+                    assertTrue(identifier.startsWith("\"") && identifier.endsWith("\""), identifier);
+                    String name = identifier.substring(1, identifier.length() - 1).replace("\"\"", "\"");
+                    result[0] = nativeTargets.get(new Name(currentSchema, name));
+                    assertNotNull(result[0], "Unexpected native resolution for " + currentSchema + "." + name);
+                    return false;
+                }
+                if (method.getName().equals("getString")) {
+                    return switch ((Integer) args[0]) {
+                        case 3 -> result[0].owner();
+                        case 4 -> result[0].name();
+                        case 6 -> result[0].databaseLink();
+                        default -> null;
+                    };
+                }
+                return defaultValue(method.getReturnType());
+            });
+        }
+
+        private PreparedStatement statement(String query) {
+            sql.add(query);
+            Map<Integer, String> parameters = new HashMap<>();
+            return proxy(PreparedStatement.class, (ignored, method, args) -> {
+                if (method.getName().equals("setString")) {
+                    parameters.put((Integer) args[0], (String) args[1]);
+                    return null;
+                }
+                if (method.getName().equals("executeQuery")) {
+                    if (failure != null && query.contains(failureView)) throw failure;
+                    Name name = new Name(parameters.get(1), parameters.get(2));
+                    if (query.contains("FROM ALL_TAB_COLUMNS")) {
+                        assertEquals(name, new Name(parameters.get(3), parameters.get(4)), "PK and column owners must agree");
+                        columnReads.add(name);
+                        return rows(columns.getOrDefault(name, List.of()));
+                    }
+                    if (query.contains("FROM ALL_SYNONYMS")) return rows(synonyms.getOrDefault(name, List.of()));
+                    if (query.contains("FROM ALL_OBJECTS")) return rows(objects.getOrDefault(name, List.of()));
+                    throw new AssertionError("Unexpected metadata query: " + query);
+                }
+                return defaultValue(method.getReturnType());
+            });
+        }
+    }
+
+    private static ResultSet rows(List<Map<String, Object>> data) {
+        int[] index = {-1};
+        boolean[] wasNull = {false};
+        return proxy(ResultSet.class, (ignored, method, args) -> {
+            if (method.getName().equals("next")) return ++index[0] < data.size();
+            if (method.getName().equals("wasNull")) return wasNull[0];
+            if (method.getName().equals("getString") || method.getName().equals("getInt") || method.getName().equals("getObject")) {
+                Object value = args[0] instanceof Integer
+                    ? data.get(index[0]).values().iterator().next() : data.get(index[0]).get(args[0]);
+                wasNull[0] = value == null;
+                if (method.getName().equals("getObject")) return value;
+                if (method.getName().equals("getInt")) return value == null ? 0 : ((Number) value).intValue();
+                return value == null ? null : value.toString();
+            }
+            return defaultValue(method.getReturnType());
+        });
+    }
+
+    @SuppressWarnings("unchecked")
+    private static <T> T proxy(Class<T> type, InvocationHandler handler) {
+        return (T) Proxy.newProxyInstance(type.getClassLoader(), new Class<?>[]{type}, handler);
+    }
+
+    private static Object defaultValue(Class<?> type) {
+        if (type == boolean.class) return false;
+        if (type == int.class) return 0;
+        if (type == long.class) return 0L;
+        return null;
+    }
+}

apps/desktop/src/components/editor/__tests__/useQueryEditorCompletionMetadata.spec.ts

@@ -42,6 +42,17 @@ function createHarness(overrides: Partial<QueryEditorProps> = {}) {
 }
 
 describe("QueryEditor completion metadata ownership", () => {
+  it("keeps an OceanBase selected schema separate from an explicitly qualified synonym", async () => {
+    const { metadata, store } = createHarness({ databaseType: "oceanbase-oracle", dialect: "oracle", schema: "MixedOwner" });
+    store.listCompletionColumns.mockResolvedValue(columns);
+    await metadata.ensureColumnsForTable({ name: "Alias" });
+    expect(store.listCompletionColumns).toHaveBeenLastCalledWith("connection", "demo", "Alias", undefined,
+      expect.objectContaining({ currentSchema: "MixedOwner" }), undefined);
+    await metadata.ensureColumnsForTable({ name: "Alias", schema: "MixedOwner" });
+    expect(store.listCompletionColumns).toHaveBeenLastCalledWith("connection", "demo", "Alias", "MixedOwner",
+      expect.objectContaining({ currentSchema: "MixedOwner" }), undefined);
+  });
+
   it.each(["getEditorSqlCompletionContext", "getEditorSemanticModel"] as const)("keys %s by document, editor state, position and dialect", (method) => {
     const { metadata, props } = createHarness();
     const state = EditorState.create({ doc: props.modelValue });

apps/desktop/src/components/editor/useQueryEditorCompletionMetadata.ts

@@ -216,6 +216,7 @@ export function useQueryEditorCompletionMetadata(options: QueryEditorCompletionM
       version: props.completionContextVersion,
       tableQuoted: reference?.nameQuoted,
       schemaQuoted: reference?.schemaQuoted,
+      ...(props.databaseType === "oceanbase-oracle" ? { currentSchema: props.schema } : {}),
     };
   }
 
@@ -249,6 +250,7 @@ export function useQueryEditorCompletionMetadata(options: QueryEditorCompletionM
     const scopedDatabase = scope && scope.database !== props.database ? scope.database : undefined;
     const database = supportsDatabaseSchemaQualifierCompletion() ? (table.database ?? scopedDatabase) : undefined;
     const baseKey = schema ? `${database ? `${database}.` : ""}${schema}.${table.name}` : table.name;
+    if (props.databaseType === "oceanbase-oracle") return `${baseKey}:${table.schema ? "qualified" : "unqualified"}`;
     if (props.databaseType !== "postgres" || (!table.nameQuoted && !table.schemaQuoted)) return baseKey;
     return `${baseKey}:quoted:s=${table.schemaQuoted ? "1" : "0"}:t=${table.nameQuoted ? "1" : "0"}`;
   }
@@ -361,7 +363,7 @@ export function useQueryEditorCompletionMetadata(options: QueryEditorCompletionM
     // unqualified table name. The query editor commonly has no schema selected
     // when the user is working from a database-level tab, so use the same
     // default as the table/DDL metadata paths instead of returning no columns.
-    const selectedSchema = table.schema ?? scope?.schema ?? props.schema;
+    const selectedSchema = props.databaseType === "oceanbase-oracle" ? table.schema : table.schema ?? scope?.schema ?? props.schema;
     const effectiveSchema = selectedSchema ?? (props.databaseType === "sqlserver" ? metadataSchemaForConnection(connectionStore.getConfig(props.connectionId ?? ""), currentDatabase, undefined) : undefined);
     if (supportsDatabaseSchemaQualifierCompletion() && table.database) {
       return { database: table.database, schema: effectiveSchema, catalog: table.catalog ?? props.catalog };
@@ -420,7 +422,7 @@ export function useQueryEditorCompletionMetadata(options: QueryEditorCompletionM
     if (!props.connectionId || props.database == null) return false;
     const target = completionMetadataTarget(table, scope);
     if (!target) return false;
-    const localColumns = connectionStore.lookupLocalCompletionColumns(props.connectionId, target.database, table.name, target.schema, target.catalog, completionColumnRequestContext(reference));
+    const localColumns = props.databaseType === "oceanbase-oracle" && !table.schema ? [] : connectionStore.lookupLocalCompletionColumns(props.connectionId, target.database, table.name, target.schema, target.catalog, completionColumnRequestContext(reference));
     if (localColumns.length > 0) {
       cachedColumnsByTable.set(cacheKey, localColumns);
       loadedColumnsByTable.add(cacheKey.toLowerCase());
@@ -433,7 +435,7 @@ export function useQueryEditorCompletionMetadata(options: QueryEditorCompletionM
     // local/remote table cache and retry with its schema before reporting that
     // star expansion is unavailable. This is especially important for aliased
     // sources because the alias itself must never be sent as the table name.
-    if (columns.length === 0 && !table.schema && !target.schema && !supportsDatabaseQualifierCompletion()) {
+    if (columns.length === 0 && props.databaseType !== "oceanbase-oracle" && !table.schema && !target.schema && !supportsDatabaseQualifierCompletion()) {
       const schemaCandidates: string[] = [];
       const seenSchemas = new Set<string>();
       const addSchema = (schema?: string | null) => {

apps/desktop/src/lib/__tests__/sql/oracleCompletionSession.spec.ts

@@ -8,7 +8,7 @@ describe("Oracle session-scoped completion", () => {
     expect(usesOracleSessionCompletionColumns({ databaseType: "oracle", clientSessionId: "tab-a", referenceSchema: "REPORTING" })).toBe(false);
     expect(usesOracleSessionCompletionColumns({ databaseType: "oceanbase-oracle", clientSessionId: "tab-a", referenceSchema: "REPORTING" })).toBe(false);
     expect(usesOracleSessionCompletionColumns({ databaseType: "oracle", clientSessionId: "tab-a", selectedSchema: "REPORTING" })).toBe(false);
-    expect(usesOracleSessionCompletionColumns({ databaseType: "oceanbase-oracle", clientSessionId: "tab-a", selectedSchema: "REPORTING" })).toBe(false);
+    expect(usesOracleSessionCompletionColumns({ databaseType: "oceanbase-oracle", clientSessionId: "tab-a", selectedSchema: "REPORTING" })).toBe(true);
     expect(usesOracleSessionCompletionColumns({ databaseType: "postgres", clientSessionId: "tab-a" })).toBe(false);
     expect(usesOracleSessionCompletionColumns({ databaseType: "oracle" })).toBe(false);
   });

apps/desktop/src/lib/backend/http.ts

@@ -1241,8 +1241,8 @@ export async function getCustomTypeDetails(connectionId: string, database: strin
   return get(`/api/schema/custom-type-details?${qs({ connection_id: connectionId, database, schema, table: name })}`);
 }
 
-export async function getColumns(connectionId: string, database: string, schema: string, table: string, catalog?: string, clientSessionId?: string): Promise<ColumnInfo[]> {
-  return get(`/api/schema/columns?${qs({ connection_id: connectionId, database, schema, table, catalog, client_session_id: clientSessionId })}`);
+export async function getColumns(connectionId: string, database: string, schema: string, table: string, catalog?: string, clientSessionId?: string, currentSchema?: string): Promise<ColumnInfo[]> {
+  return get(`/api/schema/columns?${qs({ connection_id: connectionId, database, schema, table, catalog, client_session_id: clientSessionId, current_schema: currentSchema })}`);
 }
 
 export async function getPluginTableMetadata(request: PluginTableMetadataRequest): Promise<PluginTableMetadata> {

apps/desktop/src/lib/backend/tauri.ts

@@ -1791,7 +1791,7 @@ export async function listSchemaInfos(connectionId: string, database: string): P
 export async function getCustomTypeDetails(connectionId: string, database: string, schema: string, name: string): Promise<CustomTypeDetails> {
   return invoke("get_custom_type_details", { connectionId, database, schema, name });
 }
-export async function getColumns(connectionId: string, database: string, schema: string, table: string, catalog?: string, clientSessionId?: string): Promise<ColumnInfo[]> {
+export async function getColumns(connectionId: string, database: string, schema: string, table: string, catalog?: string, clientSessionId?: string, currentSchema?: string): Promise<ColumnInfo[]> {
   return invoke("get_columns", {
     connectionId,
     database,
@@ -1799,6 +1799,7 @@ export async function getColumns(connectionId: string, database: string, schema:
     table,
     catalog,
     clientSessionId,
+    currentSchema,
   });
 }
 

apps/desktop/src/lib/metadata/__tests__/tableMetadataCache.columns.spec.ts

@@ -69,6 +69,53 @@ describe("tableMetadataCache columns facet request counts", () => {
     expect(mocks.getColumns).toHaveBeenCalledOnce();
   });
 
+  it("uses a synonym's exact resolved target for indexes while retaining its source name", async () => {
+    mocks.getColumns.mockResolvedValue([
+      { ...column("ItemId"), is_primary_key: false, resolved_schema: "MixedOwner", resolved_table: "ActualItems", resolved_object_type: "TABLE" },
+    ]);
+    mocks.listIndexes.mockResolvedValue([{ name: "ITEMS_PK", columns: ["ItemId"], is_primary: true, is_unique: true }]);
+    const request = { connectionId: "ob-1", database: "service", schema: "AppOwner", tableName: "ItemsAlias", tableType: "SYNONYM", databaseType: "oceanbase-oracle" };
+
+    const result = await loadTableMetadata(request);
+
+    expect(mocks.listIndexes).toHaveBeenCalledExactlyOnceWith("ob-1", "service", "MixedOwner", "ActualItems", undefined);
+    expect(result.metadata).toMatchObject({ schema: "AppOwner", tableName: "ItemsAlias", tableType: "TABLE", primaryKeys: ["ItemId"] });
+    expect((await loadTableColumns(request)).tableType).toBe("TABLE");
+  });
+
+  it("keeps unqualified synonym column metadata separate for different current schemas", async () => {
+    const request = { connectionId: "ob-1", database: "service", tableName: "ITEMS", databaseType: "oceanbase-oracle", currentSchema: "FirstOwner" };
+    mocks.getColumns.mockImplementation(async (_c, _d, _s, _t, _catalog, _session, currentSchema) => [{ ...column(currentSchema === "SecondOwner" ? "SecondId" : "FirstId"), resolved_schema: currentSchema, resolved_table: "Items", resolved_object_type: "TABLE" }]);
+
+    const first = await loadTableColumns(request);
+    const second = await loadTableColumns({ ...request, currentSchema: "SecondOwner" });
+    const cached = await loadTableColumns(request);
+
+    expect(first.columns[0]?.name).toBe("FirstId");
+    expect(second.columns[0]?.name).toBe("SecondId");
+    expect(cached.columns[0]?.name).toBe("FirstId");
+    expect(mocks.getColumns).toHaveBeenNthCalledWith(1, "ob-1", "service", "", "ITEMS", undefined, undefined, "FirstOwner");
+    expect(mocks.getColumns).toHaveBeenNthCalledWith(2, "ob-1", "service", "", "ITEMS", undefined, undefined, "SecondOwner");
+    expect(mocks.getColumns).toHaveBeenCalledTimes(2);
+
+    invalidateTableMetadataCache({ connectionId: "ob-1", database: "service", schema: "FirstOwner", tableName: "ITEMS" });
+    await loadTableColumns(request);
+    await loadTableColumns({ ...request, currentSchema: "SecondOwner" });
+    expect(mocks.getColumns).toHaveBeenCalledTimes(3);
+  });
+
+  it("uses the current schema for parallel index discovery on a known direct table", async () => {
+    let releaseColumns: (value: ColumnInfo[]) => void = () => {};
+    mocks.getColumns.mockReturnValueOnce(new Promise<ColumnInfo[]>((resolve) => { releaseColumns = resolve; }));
+    const load = loadTableMetadata({ connectionId: "ob-1", database: "service", schema: "", currentSchema: "SelectedOwner", tableName: "Items", tableType: "TABLE", databaseType: "oceanbase-oracle" });
+
+    await flush();
+
+    expect(mocks.listIndexes).toHaveBeenCalledExactlyOnceWith("ob-1", "service", "SelectedOwner", "Items", undefined);
+    releaseColumns([column("ID")]);
+    expect((await load).metadata.primaryKeys).toEqual(["ID"]);
+  });
+
   it("R1 — cold single-source columns load: 1 getColumns, 0 listIndexes", async () => {
     const result = await loadTableColumns({ ...usersRequest });
     expect(result.columns[0]?.name).toBe("users");

apps/desktop/src/lib/metadata/tableMetadataCache.ts

@@ -30,6 +30,7 @@ export interface TableMetadataRequest {
   tableName: string;
   tableType?: string;
   databaseType: DatabaseType | string;
+  currentSchema?: string;
   driverProfile?: string;
   catalog?: string;
   force?: boolean;
@@ -156,16 +157,16 @@ function bumpTableMetadataInvalidationStamp(scopeKey: string): void {
   tableMetadataInvalidationStamps.set(scopeKey, (tableMetadataInvalidationStamps.get(scopeKey) ?? 0) + 1);
 }
 
-export function tableMetadataScope(request: Pick<TableMetadataRequest, "connectionId" | "database" | "schema" | "tableName" | "tableType" | "driverProfile" | "databaseType" | "catalog">): MetadataScopeInput {
+export function tableMetadataScope(request: Pick<TableMetadataRequest, "connectionId" | "database" | "schema" | "tableName" | "tableType" | "driverProfile" | "databaseType" | "catalog" | "currentSchema">): MetadataScopeInput {
   return {
     kind: "table-metadata",
     connectionId: request.connectionId,
     database: request.database,
-    schema: request.schema ?? "",
+    schema: request.schema || request.currentSchema || "",
     tableName: request.tableName,
     tableType: request.tableType,
     driverProfile: request.driverProfile || request.databaseType,
-    extra: request.catalog ? { catalog: request.catalog } : undefined,
+    extra: request.catalog || request.currentSchema ? { catalog: request.catalog, currentSchema: request.currentSchema } : undefined,
   };
 }
 
@@ -176,13 +177,13 @@ function withVirtualRowIdentifier(metadata: TableMetadata, request: Pick<TableMe
   return { ...metadata, primaryKeys: virtualPrimaryKeys, virtualPrimaryKeys, rowIdentityResolved: true };
 }
 
-export function getCachedTableMetadata(request: Pick<TableMetadataRequest, "connectionId" | "database" | "schema" | "tableName" | "tableType" | "driverProfile" | "databaseType" | "catalog">): TableMetadataLoadResult | undefined {
+export function getCachedTableMetadata(request: Pick<TableMetadataRequest, "connectionId" | "database" | "schema" | "tableName" | "tableType" | "driverProfile" | "databaseType" | "catalog" | "currentSchema">): TableMetadataLoadResult | undefined {
   const hit = tableMetadataCache.get(tableMetadataScope(request));
   if (!hit) return undefined;
   return { metadata: withVirtualRowIdentifier(hit.value, request), cacheStatus: hit.stale ? "stale" : "hit", ageMs: hit.ageMs };
 }
 
-export function updateCachedTableMetadataType(request: Pick<TableMetadataRequest, "connectionId" | "database" | "schema" | "tableName" | "tableType" | "driverProfile" | "databaseType" | "catalog">, tableType: string): boolean {
+export function updateCachedTableMetadataType(request: Pick<TableMetadataRequest, "connectionId" | "database" | "schema" | "tableName" | "tableType" | "driverProfile" | "databaseType" | "catalog" | "currentSchema">, tableType: string): boolean {
   const scope = tableMetadataScope(request);
   const hit = tableMetadataCache.get(scope);
   if (!hit) return false;
@@ -190,7 +191,7 @@ export function updateCachedTableMetadataType(request: Pick<TableMetadataRequest
   return true;
 }
 
-export function getCachedTableColumns(request: Pick<TableMetadataRequest, "connectionId" | "database" | "schema" | "tableName" | "tableType" | "driverProfile" | "databaseType" | "catalog">): TableColumnsLoadResult | undefined {
+export function getCachedTableColumns(request: Pick<TableMetadataRequest, "connectionId" | "database" | "schema" | "tableName" | "tableType" | "driverProfile" | "databaseType" | "catalog" | "currentSchema">): TableColumnsLoadResult | undefined {
   const hit = tableColumnsCache.get(tableMetadataScope(request));
   if (!hit) return undefined;
   return { columns: hit.value.columns, tableType: hit.value.tableType, cacheStatus: hit.stale ? "stale" : "hit", ageMs: hit.ageMs, cachedAt: hit.cachedAt };
@@ -247,11 +248,14 @@ export async function loadTableColumns(request: TableMetadataRequest): Promise<T
       scope,
       async () => {
         // Display-only loader: columns only, never index discovery.
-        const columns = await api.getColumns(request.connectionId, request.database, request.schema ?? "", request.tableName, request.catalog);
+        const columns = request.currentSchema
+          ? await api.getColumns(request.connectionId, request.database, request.schema ?? "", request.tableName, request.catalog, undefined, request.currentSchema)
+          : await api.getColumns(request.connectionId, request.database, request.schema ?? "", request.tableName, request.catalog);
+        const resolvedTarget = columns.find((column) => column.resolved_table);
         return {
           schema: request.schema || undefined,
           tableName: request.tableName,
-          tableType: request.tableType,
+          tableType: resolvedTarget ? resolvedTarget.resolved_object_type?.trim() || "UNKNOWN" : request.tableType,
           catalog: request.catalog,
           database: request.database,
           columns,
@@ -300,7 +304,7 @@ export async function loadTableIndexes(request: TableMetadataRequest): Promise<I
     if (!oldestKey) break;
     tableIndexesLoads.delete(oldestKey);
   }
-  const promise = api.listIndexes(request.connectionId, request.database, request.schema ?? "", request.tableName, request.catalog);
+  const promise = api.listIndexes(request.connectionId, request.database, request.schema || request.currentSchema || "", request.tableName, request.catalog);
   const entry = { parts: metadataScopeParts(scope), promise, expiresAt: Date.now() + TABLE_METADATA_CACHE_TTL_MS };
   tableIndexesLoads.set(scopeKey, entry);
   void promise.catch(() => {
@@ -339,25 +343,28 @@ export async function loadTableMetadata(request: TableMetadataRequest): Promise<
     metadata = await tableMetadataCoordinator.run(
       scope,
       async () => {
-        // Column discovery can be especially slow on Oracle. Start row-identity
-        // discovery independently unless an agent-backed PostgreSQL-family
-        // relation must first report its visible schema for the index lookup.
+        // Synonyms can resolve to a different owner and object name. Wait for
+        // their columns, while known base tables retain concurrent discovery.
         const resolveReportedSchema = (request.databaseType === "vastbase" || request.databaseType === "kingbase") && !request.schema;
+        const resolveSynonymTarget = (request.databaseType === "oracle" || request.databaseType === "oceanbase-oracle" || request.driverProfile === "oracle" || request.driverProfile === "oceanbase-oracle") && request.tableType?.trim().toUpperCase() !== "TABLE";
         let rowIdentityResolved = true;
         const indexFailure = (): IndexInfo[] => {
           rowIdentityResolved = false;
           return [];
         };
-        const indexesPromise = resolveReportedSchema ? undefined : loadTableIndexes(request).catch(indexFailure);
+        const indexesPromise = resolveReportedSchema || resolveSynonymTarget ? undefined : loadTableIndexes(request).catch(() => undefined);
         const columnsResult = await columnsPromise;
         const columns = columnsResult.columns;
+        const resolvedTarget = columns.find((column) => column.resolved_table);
         const resolvedSchema = resolveReportedSchema ? columns.find((column) => column.resolved_schema)?.resolved_schema : request.schema;
-        const indexes = columns.length > 0 ? await (indexesPromise ?? loadTableIndexes({ ...request, schema: resolvedSchema }).catch(indexFailure)) : [];
-        const primaryKeys = editableRowIdentifierColumns(request.databaseType as DatabaseType, columns, indexes, request.tableType);
+        const indexRequest = { ...request, schema: resolvedTarget?.resolved_schema ?? (resolvedSchema || request.currentSchema), tableName: resolvedTarget?.resolved_table ?? request.tableName };
+        const loadedIndexes = columns.length > 0 ? await ((!resolvedTarget && indexesPromise) || loadTableIndexes(indexRequest).catch(() => undefined)) : [];
+        const indexes = loadedIndexes ?? indexFailure();
+        const primaryKeys = editableRowIdentifierColumns(request.databaseType as DatabaseType, columns, indexes, columnsResult.tableType);
         return {
           schema: resolvedSchema || undefined,
           tableName: request.tableName,
-          tableType: request.tableType,
+          tableType: columnsResult.tableType,
           catalog: request.catalog,
           database: request.database,
           columns,

apps/desktop/src/lib/sql/oracleCompletionSession.ts

@@ -10,5 +10,6 @@ export function usesOracleCurrentSchemaCompletion(databaseType?: DatabaseType, s
 }
 
 export function usesOracleSessionCompletionColumns(options: { databaseType?: DatabaseType; selectedSchema?: string; referenceSchema?: string | null; clientSessionId?: string }): boolean {
-  return usesOracleCurrentSchemaCompletion(options.databaseType, options.referenceSchema || options.selectedSchema) && !!options.clientSessionId;
+  const schema = options.databaseType === "oceanbase-oracle" ? options.referenceSchema : options.referenceSchema || options.selectedSchema;
+  return usesOracleCurrentSchemaCompletion(options.databaseType, schema) && !!options.clientSessionId;
 }

apps/desktop/src/stores/__tests__/connectionStore.completion.spec.ts

@@ -784,6 +784,21 @@ describe("connectionStore completion assistant", () => {
     expect(store.lookupLocalCompletionColumns("oracle-1", "ORCL", "ORDERS")).toEqual([]);
   });
 
+  it("passes OceanBase synonym lookup context separately and isolates selected schemas", async () => {
+    const getColumns = vi.fn().mockResolvedValue([{ name: "ID", data_type: "NUMBER", is_nullable: false }]);
+    vi.doMock("@/lib/backend/tauriRuntime", () => ({ isTauriRuntime: () => false }));
+    vi.doMock("@/lib/backend/api", () => ({ getColumns, checkConnectionHealth: vi.fn().mockResolvedValue(undefined) }));
+    const { useConnectionStore } = await import("@/stores/connectionStore");
+    const store = useConnectionStore();
+    store.connections = [oceanBaseOracleConnection()];
+    store.connectedIds.add("oceanbase-oracle-1");
+    await store.listCompletionColumns("oceanbase-oracle-1", "OBORCL", "Alias", undefined, { currentSchema: "MixedOwner" });
+    expect(getColumns).toHaveBeenLastCalledWith("oceanbase-oracle-1", "OBORCL", "", "Alias", undefined, undefined, "MixedOwner");
+    await store.listCompletionColumns("oceanbase-oracle-1", "OBORCL", "Alias", undefined, { currentSchema: "OtherOwner" });
+    expect(getColumns).toHaveBeenLastCalledWith("oceanbase-oracle-1", "OBORCL", "", "Alias", undefined, undefined, "OtherOwner");
+    expect(getColumns).toHaveBeenCalledTimes(2);
+  });
+
   it("lets a JDBC connection to Oracle resolve CURRENT_SCHEMA for unqualified column completion, same as the native Oracle driver", async () => {
     const completionAssistantSearch = vi.fn().mockResolvedValue({
       candidates: [],

apps/desktop/src/stores/__tests__/queryStore.hiddenPrimaryKey.spec.ts

@@ -1341,6 +1341,61 @@ describe("queryStore hidden primary key editing", () => {
     expect(tab.result?.hidden_column_indexes).toBeUndefined();
   });
 
+  it.each([
+    { sourceSchema: undefined, synonym: true },
+    { sourceSchema: "ExplicitOwner", synonym: true },
+    { sourceSchema: undefined, synonym: false },
+  ])("keeps OceanBase current schema separate from explicit source schema $sourceSchema with synonym=$synonym", async ({ sourceSchema, synonym }) => {
+    getConnectionConfig.mockReturnValue({ id: "ob-1", name: "OceanBase", db_type: "oceanbase-oracle", database: "service", query_timeout_secs: 30 });
+    getColumns.mockResolvedValue([{ name: "ItemId", data_type: "NUMBER", is_nullable: false, column_default: null, is_primary_key: true, extra: null, ...(synonym ? { resolved_schema: "TargetOwner", resolved_table: "Items", resolved_object_type: "TABLE" } : {}) }]);
+    analyzeEditableQueryEditability.mockResolvedValue({
+      editable: true,
+      analysis: {
+        schema: sourceSchema,
+        schemaQuoted: true,
+        tableName: "ItemsAlias",
+        tableNameQuoted: true,
+        selectStar: false,
+        columns: [{ sourceName: "ItemId", sourceNameQuoted: true, resultName: "ItemId", expression: '"ItemId"' }],
+      },
+    });
+    executeMulti.mockResolvedValue([{ columns: ["ItemId"], rows: [[1]], affected_rows: 0, execution_time_ms: 1 }]);
+    const { useQueryStore } = await import("@/stores/queryStore");
+    const store = useQueryStore();
+    const tabId = store.createTab("ob-1", "service", "Query", "query", "SelectedOwner");
+    store.setAutoCommit(tabId, true);
+
+    await store.executeTabSql(tabId, sourceSchema ? 'SELECT "ItemId" FROM "ExplicitOwner"."ItemsAlias"' : 'SELECT "ItemId" FROM "ItemsAlias"');
+
+    if (sourceSchema) expect(getColumns).toHaveBeenCalledWith("ob-1", "service", "ExplicitOwner", "ItemsAlias", undefined);
+    else expect(getColumns).toHaveBeenCalledWith("ob-1", "service", "", "ItemsAlias", undefined, undefined, "SelectedOwner");
+    expect(listIndexes).toHaveBeenCalledWith("ob-1", "service", synonym ? "TargetOwner" : "SelectedOwner", synonym ? "Items" : "ItemsAlias", undefined);
+    const tab = store.tabs.find((item) => item.id === tabId)!;
+    await vi.waitFor(() => expect(tab.tableMeta).toMatchObject({ schema: synonym ? "TargetOwner" : "SelectedOwner", tableName: synonym ? "Items" : "ItemsAlias" }));
+    if (synonym) expect(tab.tableMeta?.tableType).toBe("TABLE");
+    expect(tab.queryAnalysis?.tableName).toBe("ItemsAlias");
+  });
+
+  it.each(["VIEW", undefined])("does not treat a synonym target of type %s as an Oracle base table", async (objectType) => {
+    getConnectionConfig.mockReturnValue({ id: "oracle-1", name: "Oracle", db_type: "oracle", database: "ORCL", query_timeout_secs: 30 });
+    getColumns.mockResolvedValue([{ name: "ID", data_type: "NUMBER", is_nullable: false, column_default: null, is_primary_key: false, extra: null, resolved_schema: "TargetOwner", resolved_table: "ActualItems", resolved_object_type: objectType }]);
+    lookupLocalCompletionTables.mockReturnValue([{ name: "ITEMS_ALIAS", type: "table", schema: "APP" }]);
+    analyzeEditableQueryEditability.mockResolvedValue({ editable: true, analysis: { schema: "APP", tableName: "ITEMS_ALIAS", selectStar: true, columns: [] } });
+    executeMulti.mockResolvedValue([{ columns: ["ID"], rows: [[1]], affected_rows: 0, execution_time_ms: 1 }]);
+    const { useQueryStore } = await import("@/stores/queryStore");
+    const store = useQueryStore();
+    const tabId = store.createTab("oracle-1", "ORCL", "Query");
+    store.setAutoCommit(tabId, true);
+
+    await store.executeTabSql(tabId, "SELECT * FROM APP.ITEMS_ALIAS");
+
+    expect(executeMulti).toHaveBeenCalledWith("oracle-1", "ORCL", "SELECT * FROM APP.ITEMS_ALIAS", undefined, expect.any(String), expect.any(Object));
+    const tab = store.tabs.find((item) => item.id === tabId)!;
+    await vi.waitFor(() => expect(tab.result?.columns).toEqual(["ID"]));
+    expect(tab.result?.hidden_column_indexes).toBeUndefined();
+    expect(tab.queryAnalysis).toBeUndefined();
+  });
+
   it.each(["CLOB", "XMLTYPE", "SYS.XMLTYPE"])("enables deferred Oracle %s values only when a base-table query has a stable key", async (dataType) => {
     getConnectionConfig.mockReturnValue({ id: "oracle-1", name: "Oracle", db_type: "oracle", database: "ORCL", query_timeout_secs: 30 });
     getColumns.mockResolvedValue([

apps/desktop/src/stores/connectionStore.ts

@@ -9260,7 +9260,7 @@ export const useConnectionStore = defineStore("connection", () => {
   /// the same fallback.
   const LOGIN_SCHEMA_COMPLETION_TYPES = new Set(["dameng", "db2"]);
 
-  async function listCompletionColumns(connectionId: string, database: string, table: string, schema?: string, context?: { clientSessionId?: string; version?: number; tableQuoted?: boolean; schemaQuoted?: boolean }, catalog?: string): Promise<SqlCompletionColumn[]> {
+  async function listCompletionColumns(connectionId: string, database: string, table: string, schema?: string, context?: { clientSessionId?: string; version?: number; tableQuoted?: boolean; schemaQuoted?: boolean; currentSchema?: string }, catalog?: string): Promise<SqlCompletionColumn[]> {
     const config = getConfig(connectionId);
     // Use the effective database type (e.g. a JDBC connection whose URL is
     // `jdbc:oracle:...` resolves to "oracle") rather than the raw db_type.
@@ -9282,7 +9282,9 @@ export const useConnectionStore = defineStore("connection", () => {
       return [];
     }
     const sessionCacheScope = usesCurrentSchema && context?.clientSessionId ? `:${context.clientSessionId}:${context.version ?? 0}` : "";
-    const cacheKey = `${completionColumnsKey(connectionId, database, completionTable, completionSchema, catalog, context)}${sessionCacheScope}`;
+    const currentSchema = effectiveDbType === "oceanbase-oracle" && usesCurrentSchema ? context?.currentSchema : undefined;
+    const contextCacheScope = currentSchema ? `:current=${JSON.stringify(currentSchema)}` : "";
+    const cacheKey = `${completionColumnsKey(connectionId, database, completionTable, completionSchema, catalog, context)}${sessionCacheScope}${contextCacheScope}`;
     if (!completionColumnsCache.value[cacheKey]) {
       const requestRevision = completionCacheRevision(connectionId, database);
       await withCompletionInFlight(
@@ -9316,7 +9318,9 @@ export const useConnectionStore = defineStore("connection", () => {
             }
           }
           const querySchema = usesCurrentSchema ? "" : metadataQuerySchema(connectionId, database, completionSchema);
-          const columns = await api.getColumns(connectionId, database, querySchema, completionTable, catalog, usesCurrentSchema ? context?.clientSessionId : undefined);
+          const columns = currentSchema
+            ? await api.getColumns(connectionId, database, querySchema, completionTable, catalog, context?.clientSessionId, currentSchema)
+            : await api.getColumns(connectionId, database, querySchema, completionTable, catalog, usesCurrentSchema ? context?.clientSessionId : undefined);
           if (requestRevision !== completionCacheRevision(connectionId, database)) return;
           completionColumnsCache.value[cacheKey] = columns;
           evictOldestCacheEntries(completionColumnsCache.value, COMPLETION_CACHE_MAX);
@@ -9387,7 +9391,7 @@ export const useConnectionStore = defineStore("connection", () => {
     return listCompletionDatabases(connectionId);
   }
 
-  function refreshCompletionColumns(connectionId: string, database: string, table: string, schema?: string, context?: { clientSessionId?: string; version?: number; tableQuoted?: boolean; schemaQuoted?: boolean }, catalog?: string): Promise<SqlCompletionColumn[]> {
+  function refreshCompletionColumns(connectionId: string, database: string, table: string, schema?: string, context?: { clientSessionId?: string; version?: number; tableQuoted?: boolean; schemaQuoted?: boolean; currentSchema?: string }, catalog?: string): Promise<SqlCompletionColumn[]> {
     return listCompletionColumns(connectionId, database, table, schema, context, catalog);
   }
 

apps/desktop/src/stores/queryStore.ts

@@ -6143,6 +6143,7 @@ export const useQueryStore = defineStore("query", () => {
   }
 
   function canUseQueryKeylessRowPredicate(databaseType: DatabaseType, loaded: LoadedEditableSource): boolean {
+    if (loaded.tableMeta.columns.some((column) => column.resolved_table) && loaded.tableMeta.tableType?.trim().toUpperCase() !== "TABLE") return false;
     if (!canUseKeylessRowPredicate(databaseType, loaded.tableMeta.primaryKeys)) return false;
     // An unknown Oracle object may be a view whose query shape rejects ROWID
     // and whose rows cannot be mapped safely for writes. Keep the result
@@ -6200,7 +6201,10 @@ export const useQueryStore = defineStore("query", () => {
     // (issue #10567). Fold the SQL-text schema, not a tab-selected one — the
     // object tree already reports the stored spelling.
     const foldedSourceSchema = foldUnquotedPostgresMetadataIdentifier(metadataDbType, source.schema, source.schemaQuoted);
-    const schema = foldedSourceSchema || (dbType === "sqlserver" ? "" : tab.schema) || "";
+    // An OceanBase public synonym is eligible only for an unqualified name.
+    // Keep the selected CURRENT_SCHEMA separate from an explicit SQL owner.
+    const resolveInCurrentOceanbaseSchema = metadataDbType === "oceanbase-oracle" && !source.schema;
+    const schema = foldedSourceSchema || (dbType === "sqlserver" || resolveInCurrentOceanbaseSchema ? "" : tab.schema) || "";
     // Oracle-family connection databases are service names, not schemas. When
     // the query does not qualify a schema, let the driver resolve the current
     // login user's schema instead of looking up metadata under the service name.
@@ -6232,7 +6236,7 @@ export const useQueryStore = defineStore("query", () => {
     // Keep SQL Server writes unqualified unless the SELECT source explicitly
     // named a schema, so SELECT and UPDATE resolve the same object.
     const writeSchema = dbType === "sqlserver" && !source.schema ? undefined : metadataSchema || undefined;
-    const localTableType = oracleCompletionTableType(tab, metadataDbType, metadataDatabase, metadataSchema || conn?.default_schema || "", metadataTableName, metadataCatalog);
+    const localTableType = oracleCompletionTableType(tab, metadataDbType, metadataDatabase, metadataSchema || (resolveInCurrentOceanbaseSchema ? tab.schema : undefined) || conn?.default_schema || "", metadataTableName, metadataCatalog);
     const knownTableType = localTableType ?? (tab.tableMeta?.tableName.toLowerCase() === metadataTableName.toLowerCase() && normalizeOptionalSchema(tab.tableMeta.schema) === normalizeOptionalSchema(metadataSchema) ? tab.tableMeta.tableType : undefined);
     return {
       source: metadataSource,
@@ -6242,6 +6246,7 @@ export const useQueryStore = defineStore("query", () => {
         connectionId: tab.connectionId!,
         database: metadataDatabase,
         schema: metadataSchema,
+        ...(resolveInCurrentOceanbaseSchema && tab.schema ? { currentSchema: tab.schema } : {}),
         tableName: metadataTableName,
         tableType: knownTableType,
         databaseType: dbType,
@@ -6253,7 +6258,8 @@ export const useQueryStore = defineStore("query", () => {
 
   function loadedEditableSourceFromMetadata(target: EditableSourceMetadataTarget, metadata: Awaited<ReturnType<typeof loadTableMetadata>>["metadata"]): LoadedEditableSource {
     const usesReportedSchema = target.request.databaseType === "vastbase" || target.request.databaseType === "kingbase";
-    const writeSchema = usesReportedSchema && !target.writeSchema ? metadata.schema : target.writeSchema;
+    const resolvedTarget = metadata.columns.find((column) => column.resolved_table);
+    const writeSchema = resolvedTarget?.resolved_schema ?? (usesReportedSchema && !target.writeSchema ? metadata.schema : target.writeSchema || target.request.currentSchema);
     return {
       source: target.source,
       analysis: target.analysis,
@@ -6261,7 +6267,7 @@ export const useQueryStore = defineStore("query", () => {
         catalog: target.request.catalog,
         database: target.request.database,
         schema: writeSchema,
-        tableName: target.request.tableName,
+        tableName: resolvedTarget?.resolved_table ?? target.request.tableName,
         tableType: metadata.tableType,
         columns: metadata.columns,
         primaryKeys: metadata.primaryKeys,

apps/desktop/src/types/database.ts

@@ -1136,6 +1136,8 @@ export interface ColumnInfo {
   name: string;
   data_type: string;
   resolved_schema?: string;
+  resolved_table?: string;
+  resolved_object_type?: string;
   is_nullable: boolean;
   column_default: string | null;
   is_primary_key: boolean;

crates/dbx-core/src/data/script_generator.rs

@@ -1434,6 +1434,8 @@ mod tests {
             name: name.to_string(),
             data_type: data_type.to_string(),
             resolved_schema: None,
+            resolved_table: None,
+            resolved_object_type: None,
             is_nullable: true,
             column_default: None,
             is_primary_key: false,
@@ -2388,6 +2390,8 @@ mod tests {
                 name: "large_val".to_string(),
                 data_type: "BIGINT".to_string(),
                 resolved_schema: None,
+                resolved_table: None,
+                resolved_object_type: None,
                 is_nullable: true,
                 column_default: None,
                 is_primary_key: false,
@@ -2406,6 +2410,8 @@ mod tests {
                 name: "large_val".to_string(),
                 data_type: "INT".to_string(),
                 resolved_schema: None,
+                resolved_table: None,
+                resolved_object_type: None,
                 is_nullable: true,
                 column_default: None,
                 is_primary_key: false,

crates/dbx-core/src/schema/mod.rs

@@ -7753,6 +7753,20 @@ pub async fn get_columns_core_for_session(
     schema: &str,
     table: &str,
     client_session_id: Option<&str>,
+) -> Result<Vec<db::ColumnInfo>, String> {
+    get_columns_core_for_session_in_context(state, connection_id, database, schema, table, client_session_id, None)
+        .await
+}
+
+/// Keep the selected OceanBase schema separate from an explicitly qualified object schema.
+pub async fn get_columns_core_for_session_in_context(
+    state: &AppState,
+    connection_id: &str,
+    database: &str,
+    schema: &str,
+    table: &str,
+    client_session_id: Option<&str>,
+    current_schema: Option<&str>,
 ) -> Result<Vec<db::ColumnInfo>, String> {
     if connection_config(state, connection_id).await.is_some_and(|config| config.db_type == DatabaseType::MongoDb) {
         return Box::pin(mongodb_columns::get_columns(state, connection_id, database, table)).await;
@@ -7761,7 +7775,7 @@ pub async fn get_columns_core_for_session(
         let metadata_session =
             EphemeralAgentMetadataSession::open(state, connection_id, Some(database), "columns").await;
         if metadata_session.client_session_id().is_some() {
-            let result = get_columns_core_for_session_inner(
+            let result = get_columns_core_for_session_inner_in_context(
                 state,
                 connection_id,
                 database,
@@ -7769,13 +7783,36 @@ pub async fn get_columns_core_for_session(
                 table,
                 metadata_session.client_session_id(),
                 false,
+                current_schema,
             )
             .await;
             metadata_session.finish(state, connection_id, Some(database)).await;
             return result;
         }
     }
-    get_columns_core_for_session_inner(state, connection_id, database, schema, table, client_session_id, true).await
+    if current_schema.is_none() {
+        return get_columns_core_for_session_inner(
+            state,
+            connection_id,
+            database,
+            schema,
+            table,
+            client_session_id,
+            true,
+        )
+        .await;
+    }
+    get_columns_core_for_session_inner_in_context(
+        state,
+        connection_id,
+        database,
+        schema,
+        table,
+        client_session_id,
+        true,
+        current_schema,
+    )
+    .await
 }
 
 async fn get_columns_core_for_session_inner(
@@ -7786,6 +7823,29 @@ async fn get_columns_core_for_session_inner(
     table: &str,
     client_session_id: Option<&str>,
     use_client_session_context: bool,
+) -> Result<Vec<db::ColumnInfo>, String> {
+    get_columns_core_for_session_inner_in_context(
+        state,
+        connection_id,
+        database,
+        schema,
+        table,
+        client_session_id,
+        use_client_session_context,
+        None,
+    )
+    .await
+}
+
+async fn get_columns_core_for_session_inner_in_context(
+    state: &AppState,
+    connection_id: &str,
+    database: &str,
+    schema: &str,
+    table: &str,
+    client_session_id: Option<&str>,
+    use_client_session_context: bool,
+    current_schema: Option<&str>,
 ) -> Result<Vec<db::ColumnInfo>, String> {
     get_columns_core_for_session_inner_with_pool(
         state,
@@ -7797,6 +7857,7 @@ async fn get_columns_core_for_session_inner(
         use_client_session_context,
         None,
         true,
+        current_schema,
     )
     .await
 }
@@ -7819,6 +7880,7 @@ async fn get_columns_core_for_existing_pool(
         true,
         Some(pool_key),
         false,
+        None,
     )
     .await
 }
@@ -7833,6 +7895,7 @@ async fn get_columns_core_for_session_inner_with_pool(
     use_client_session_context: bool,
     existing_pool_key: Option<&str>,
     allow_recovery: bool,
+    current_schema: Option<&str>,
 ) -> Result<Vec<db::ColumnInfo>, String> {
     let context_session_id = if use_client_session_context { client_session_id } else { None };
     let existing_pool_key = existing_pool_key.map(str::to_owned);
@@ -8001,10 +8064,13 @@ async fn get_columns_core_for_session_inner_with_pool(
                     }
                 }
                 match client
-                    .get_columns::<Vec<db::ColumnInfo>>(
+                    .get_columns_in_context::<Vec<db::ColumnInfo>>(
                         database,
                         schema,
                         table,
+                        current_schema.filter(|_| {
+                            db_config.as_ref().is_some_and(|config| config.db_type == DatabaseType::OceanbaseOracle)
+                        }),
                         agent_metadata_timeout(db_config.as_ref()),
                     )
                     .await

crates/dbx-core/src/schema/mongodb_columns.rs

@@ -77,6 +77,8 @@ fn columns_from_sample(result: &MongoDocumentResult) -> Result<Vec<ColumnInfo>,
             name,
             data_type: types.into_iter().collect::<Vec<_>>().join(" | "),
             resolved_schema: None,
+            resolved_table: None,
+            resolved_object_type: None,
             is_nullable: true,
             column_default: None,
             is_unique: false,

crates/dbx-core/tests/api_contract_verification.rs

@@ -283,6 +283,8 @@ fn column_info_serialization_roundtrip() {
         name: "id".to_string(),
         data_type: "int".to_string(),
         resolved_schema: None,
+        resolved_table: None,
+        resolved_object_type: None,
         is_nullable: false,
         column_default: None,
         is_primary_key: true,
@@ -331,6 +333,8 @@ fn table_columns_result_serialization_contract() {
             name: "id".to_string(),
             data_type: "int".to_string(),
             resolved_schema: None,
+            resolved_table: None,
+            resolved_object_type: None,
             is_nullable: false,
             column_default: None,
             is_primary_key: true,

crates/dbx-core/tests/bidirectional_diff_e2e.rs

@@ -18,6 +18,8 @@ fn col(name: &str, data_type: &str) -> ColumnInfo {
         name: name.to_string(),
         data_type: data_type.to_string(),
         resolved_schema: None,
+        resolved_table: None,
+        resolved_object_type: None,
         is_nullable: false,
         column_default: None,
         is_primary_key: false,

crates/dbx-core/tests/cross_dialect_integration.rs

@@ -21,6 +21,8 @@ fn col(name: &str, data_type: &str) -> ColumnInfo {
         name: name.to_string(),
         data_type: data_type.to_string(),
         resolved_schema: None,
+        resolved_table: None,
+        resolved_object_type: None,
         is_nullable: false,
         column_default: None,
         is_primary_key: false,

crates/dbx-core/tests/performance_benchmarks.rs

@@ -36,6 +36,8 @@ fn generate_details(tables: &[TableInfo], columns_per_table: usize) -> Vec<Table
                     name: format!("col_{}", j),
                     data_type: if j % 3 == 0 { "int".to_string() } else { "varchar(64)".to_string() },
                     resolved_schema: None,
+                    resolved_table: None,
+                    resolved_object_type: None,
                     is_nullable: j % 2 == 0,
                     column_default: if j == 0 { Some("0".to_string()) } else { None },
                     is_primary_key: j == 0,

crates/dbx-core/tests/regression_compatibility.rs

@@ -27,6 +27,8 @@ fn col(name: &str, data_type: &str) -> ColumnInfo {
         name: name.to_string(),
         data_type: data_type.to_string(),
         resolved_schema: None,
+        resolved_table: None,
+        resolved_object_type: None,
         is_nullable: false,
         column_default: None,
         is_primary_key: false,

crates/dbx-core/tests/schema_diff_table_mapping.rs

@@ -20,6 +20,8 @@ fn column(name: &str, data_type: &str) -> ColumnInfo {
         name: name.to_string(),
         data_type: data_type.to_string(),
         resolved_schema: None,
+        resolved_table: None,
+        resolved_object_type: None,
         is_nullable: false,
         column_default: None,
         is_primary_key: false,

crates/dbx-driver-agent/src/agent_driver.rs

@@ -2373,12 +2373,22 @@ impl AgentDriverClient {
         table: &str,
         timeout_duration: Option<Duration>,
     ) -> Result<T, String> {
-        self.call_method_with_timeout(
-            AgentMethod::GetColumns,
-            agent_schema_table_params(database, schema, table),
-            timeout_duration,
-        )
-        .await
+        self.get_columns_in_context(database, schema, table, None, timeout_duration).await
+    }
+
+    pub async fn get_columns_in_context<T: DeserializeOwned + Send + 'static>(
+        &mut self,
+        database: &str,
+        schema: &str,
+        table: &str,
+        current_schema: Option<&str>,
+        timeout_duration: Option<Duration>,
+    ) -> Result<T, String> {
+        let mut params = agent_schema_table_params(database, schema, table);
+        if let Some(current_schema) = current_schema {
+            params["current_schema"] = serde_json::json!(current_schema);
+        }
+        self.call_method_with_timeout(AgentMethod::GetColumns, params, timeout_duration).await
     }
 
     pub async fn get_custom_type_details<T: DeserializeOwned + Send + 'static>(
@@ -4617,6 +4627,9 @@ def respond(req):
 
     session_id = params.get('agentSessionId', '__legacy__')
     with session_lock(session_id):
+        if method == 'get_columns':
+            write_response(req, params)
+            return
         if method in ('execute_query', 'start_table_read'):
             sql = params.get('sql', '')
             with state_lock:
@@ -4664,6 +4677,43 @@ for line in sys.stdin:
         runtime.call(method, serde_json::json!({}), Some(Duration::from_secs(2)), None).await.unwrap()
     }
 
+    #[tokio::test]
+    async fn columns_rpc_preserves_explicit_schema_and_optional_current_schema() {
+        let (runtime, script_path) = spawn_stateful_test_runtime("columns-context-test").await;
+        let mut client = AgentDriverClient::shared_session(runtime.clone(), "columns-session".to_string());
+
+        let contextual: serde_json::Value = client
+            .get_columns_in_context("APP", "", "ORDERS_ALIAS", Some("SelectedOwner"), Some(Duration::from_secs(2)))
+            .await
+            .unwrap();
+        assert_eq!(contextual["schema"], "");
+        assert_eq!(contextual["current_schema"], "SelectedOwner");
+        assert_eq!(contextual["table"], "ORDERS_ALIAS");
+        assert_eq!(contextual["agentSessionId"], "columns-session");
+
+        let qualified: serde_json::Value = client
+            .get_columns_in_context(
+                "APP",
+                "ActualOwner",
+                "ORDERS_ALIAS",
+                Some("SelectedOwner"),
+                Some(Duration::from_secs(2)),
+            )
+            .await
+            .unwrap();
+        assert_eq!(qualified["schema"], "ActualOwner");
+        assert_eq!(qualified["current_schema"], "SelectedOwner");
+
+        let legacy: serde_json::Value =
+            client.get_columns("APP", "OWNER", "ORDERS_ALIAS", Some(Duration::from_secs(2))).await.unwrap();
+        assert_eq!(legacy["schema"], "OWNER");
+        assert!(legacy.get("current_schema").is_none());
+
+        runtime.kill();
+        wait_for_runtime_reap(&runtime).await;
+        let _ = std::fs::remove_file(script_path);
+    }
+
     async fn wait_for_runtime_reap(runtime: &AgentRuntimeClient) {
         tokio::time::timeout(Duration::from_secs(2), async {
             while !runtime.child_reaped.load(Ordering::Acquire) {

crates/dbx-driver-mysql/src/mysql.rs

@@ -4944,6 +4944,8 @@ where
                 name,
                 data_type: column_type,
                 resolved_schema: None,
+                resolved_table: None,
+                resolved_object_type: None,
                 is_nullable: get_str_by_name(row, "IS_NULLABLE") == "YES",
                 column_default: get_opt_str(row, "COLUMN_DEFAULT"),
                 extra: get_opt_str(row, "EXTRA"),
@@ -5017,6 +5019,8 @@ async fn fetch_columns_show(
                 name,
                 data_type: get_str_by_name(row, "Type"),
                 resolved_schema: None,
+                resolved_table: None,
+                resolved_object_type: None,
                 is_nullable: get_str_by_name(row, "Null").eq_ignore_ascii_case("YES"),
                 column_default: get_opt_str(row, "Default"),
                 is_primary_key: key.eq_ignore_ascii_case("PRI"),
@@ -8802,6 +8806,8 @@ mod tests {
             name: " state value".to_string(),
             data_type: "unknown".to_string(),
             resolved_schema: Some("analytics".to_string()),
+            resolved_table: None,
+            resolved_object_type: None,
             is_nullable: true,
             column_default: Some("seed".to_string()),
             is_primary_key: true,

crates/dbx-driver-mysql/src/mysql_compatible.rs

@@ -201,6 +201,8 @@ pub async fn get_columns_show_from(
                 name,
                 data_type: get_str_by_name(row, "Type"),
                 resolved_schema: None,
+                resolved_table: None,
+                resolved_object_type: None,
                 is_nullable: get_str_by_name(row, "Null").eq_ignore_ascii_case("YES"),
                 column_default: get_opt_str(row, "Default"),
                 is_primary_key: key.eq_ignore_ascii_case("PRI"),

crates/dbx-drivers/src/db/meilisearch_driver.rs

@@ -384,6 +384,8 @@ pub async fn get_columns(client: &MeilisearchClient, index: &str) -> Result<Vec<
             name,
             data_type: field_type.label().to_string(),
             resolved_schema: None,
+            resolved_table: None,
+            resolved_object_type: None,
             is_nullable: true,
             column_default: None,
             is_unique: false,

crates/dbx-drivers/src/db/salesforce_driver.rs

@@ -865,6 +865,8 @@ fn parse_describe_columns(describe: &Value) -> Vec<ColumnInfo> {
                         character_maximum_length: length.and_then(|l| i32::try_from(l).ok()),
                         enum_values: picklist_values,
                         resolved_schema: None,
+                        resolved_table: None,
+                        resolved_object_type: None,
                         // Plugin-provided column metadata capabilities do not apply:
                         // the describe payload is parsed here, not by a plugin driver.
                         metadata_capabilities: None,

crates/dbx-mcp/src/backend.rs

@@ -2807,6 +2807,8 @@ fn infer_document_columns(documents: &[Value]) -> Vec<ColumnInfo> {
             name,
             data_type,
             resolved_schema: None,
+            resolved_table: None,
+            resolved_object_type: None,
             is_nullable: true,
             column_default: None,
             is_primary_key: false,

crates/dbx-sql-schema/src/schema_diff.rs

@@ -6908,6 +6908,8 @@ mod tests {
             name: name.to_string(),
             data_type: data_type.to_string(),
             resolved_schema: None,
+            resolved_table: None,
+            resolved_object_type: None,
             is_nullable: false,
             column_default: None,
             is_primary_key: false,
@@ -9957,6 +9959,8 @@ mod tests {
                     name: "status".to_string(),
                     data_type: "text".to_string(),
                     resolved_schema: None,
+                    resolved_table: None,
+                    resolved_object_type: None,
                     is_nullable: true,
                     column_default: None,
                     is_primary_key: false,
@@ -13175,6 +13179,8 @@ mod tests {
             name: "c".into(),
             data_type: "varchar(100)".into(),
             resolved_schema: None,
+            resolved_table: None,
+            resolved_object_type: None,
             is_nullable: true,
             column_default: Some("'default'".into()),
             comment: Some("new".into()),
@@ -13193,6 +13199,8 @@ mod tests {
             name: "c".into(),
             data_type: "varchar(50)".into(),
             resolved_schema: None,
+            resolved_table: None,
+            resolved_object_type: None,
             is_nullable: false,
             column_default: None,
             comment: Some("old".into()),

crates/dbx-tauri-schema/src/commands.rs

@@ -398,18 +398,20 @@ pub async fn get_columns(
     table: String,
     catalog: Option<String>,
     client_session_id: Option<String>,
+    current_schema: Option<String>,
 ) -> Result<Vec<db::ColumnInfo>, String> {
     if let Some(catalog) = external_doris_catalog(&state, &connection_id, catalog.as_deref()).await {
         return dbx_core::schema::get_doris_catalog_columns_core(&state, &connection_id, &catalog, &database, &table)
             .await;
     }
-    dbx_core::schema::get_columns_core_for_session(
+    dbx_core::schema::get_columns_core_for_session_in_context(
         &state,
         &connection_id,
         &database,
         &schema,
         &table,
         client_session_id.as_deref(),
+        current_schema.as_deref(),
     )
     .await
 }

crates/dbx-types/src/types.rs

@@ -291,6 +291,10 @@ pub struct ColumnInfo {
     pub data_type: String,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub resolved_schema: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub resolved_table: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub resolved_object_type: Option<String>,
     pub is_nullable: bool,
     pub column_default: Option<String>,
     pub is_primary_key: bool,

crates/dbx-types/tests/column_target_identity.rs

@@ -0,0 +1,33 @@
+use dbx_types::types::ColumnInfo;
+use serde_json::json;
+
+#[test]
+fn column_metadata_preserves_synonym_target_identity_across_rpc() {
+    let payload = json!({
+        "name": "Amount", "data_type": "NUMBER(12,2)", "is_nullable": true,
+        "column_default": null, "is_primary_key": false, "extra": null,
+        "comment": null, "numeric_precision": 12, "numeric_scale": 2,
+        "character_maximum_length": null,
+        "resolved_schema": "MixedOwner", "resolved_table": "OrderView",
+        "resolved_object_type": "VIEW"
+    });
+    let column: ColumnInfo = serde_json::from_value(payload.clone()).unwrap();
+    let forwarded = serde_json::to_value(column).unwrap();
+    for key in ["resolved_schema", "resolved_table", "resolved_object_type"] {
+        assert_eq!(forwarded.get(key), payload.get(key), "lost {key} in Core metadata forwarding");
+    }
+}
+
+#[test]
+fn older_column_metadata_does_not_invent_a_target_type() {
+    let payload = json!({
+        "name": "ID", "data_type": "NUMBER", "is_nullable": false,
+        "column_default": null, "is_primary_key": true, "extra": null,
+        "comment": null, "numeric_precision": null, "numeric_scale": null,
+        "character_maximum_length": null
+    });
+    let column: ColumnInfo = serde_json::from_value(payload).unwrap();
+    let forwarded = serde_json::to_value(column).unwrap();
+    assert!(forwarded.get("resolved_table").is_none());
+    assert!(forwarded.get("resolved_object_type").is_none());
+}

crates/dbx-web/src/routes/schema.rs

@@ -27,6 +27,7 @@ pub struct SchemaQuery {
     pub table_name_filter: Option<String>,
     pub apply_visible_filter: Option<bool>,
     pub client_session_id: Option<String>,
+    pub current_schema: Option<String>,
     pub include_postgres_access: Option<bool>,
     pub portable: Option<bool>,
     pub execution_id: Option<String>,
@@ -484,8 +485,8 @@ where
     Ok(value)
 }
 
-fn should_cache_columns(client_session_id: Option<&str>) -> bool {
-    client_session_id.is_none()
+fn should_cache_columns(client_session_id: Option<&str>, current_schema: Option<&str>) -> bool {
+    client_session_id.is_none() && current_schema.is_none()
 }
 
 pub async fn get_plugin_table_metadata(
@@ -511,19 +512,20 @@ pub async fn list_columns(
                 .await
                 .map_err(AppError::from)
         } else {
-            dbx_core::schema::get_columns_core_for_session(
+            dbx_core::schema::get_columns_core_for_session_in_context(
                 &state.app,
                 &q.connection_id,
                 database,
                 schema,
                 table,
                 q.client_session_id.as_deref(),
+                q.current_schema.as_deref(),
             )
             .await
             .map_err(AppError::from)
         }
     };
-    let result = if should_cache_columns(q.client_session_id.as_deref()) {
+    let result = if should_cache_columns(q.client_session_id.as_deref(), q.current_schema.as_deref()) {
         cached_metadata(
             &state.app,
             &q.connection_id,
@@ -979,9 +981,21 @@ mod tests {
 
     #[test]
     fn session_scoped_columns_bypass_persistent_cache() {
-        assert!(should_cache_columns(None));
-        assert!(!should_cache_columns(Some("oracle-session")));
-        assert!(!should_cache_columns(Some("")));
+        assert!(should_cache_columns(None, None));
+        assert!(!should_cache_columns(Some("oracle-session"), None));
+        assert!(!should_cache_columns(Some(""), None));
+    }
+
+    #[test]
+    fn current_schema_columns_bypass_cache_without_changing_explicit_schema() {
+        let query: SchemaQuery = serde_json::from_value(serde_json::json!({
+            "connection_id": "ob", "database": "LOGIN", "schema": "", "table": "ORDERS_ALIAS",
+            "current_schema": "SelectedOwner"
+        }))
+        .unwrap();
+        assert_eq!(query.schema.as_deref(), Some(""));
+        assert_eq!(query.current_schema.as_deref(), Some("SelectedOwner"));
+        assert!(!should_cache_columns(query.client_session_id.as_deref(), query.current_schema.as_deref()));
     }
 
     #[test]