fix(postgres): schema-qualify the primary-key subquery in getTableColumns

The PK subquery joined tc.constraint_name = ku.constraint_name and filtered on
ku.table_name alone, with no schema predicate anywhere. Postgres auto-names
primary keys "<table>_pkey", so two schemas holding a same-named table collide
by construction — the join cross-matches them.

Measured against two schemas each with an `orders` table (s_a's PK is `other`,
s_b's is `name`), asking for s_b.orders:

    column_name | is_primary_key
    name        | t
    name        | t      <- duplicated row
    other       | t      <- false positive: `other` is s_a's PK, not s_b's
    other       | t      <- duplicated row

Ground truth from pg_index: s_b.orders has exactly one PK column, `name`. So the
provider reported every column as a primary key and returned each one twice —
duplicate ColumnInfo entries, and a table with no real single PK. Adding
tc.table_schema = ku.table_schema and ku.table_schema = ? returns the correct
`name | t`, `other | f`.

This predates the search_path change and was mostly latent while the provider
only ever read `public`: a collision needed some other schema to hold a
same-named table. Now that it targets whatever the session resolves to, that is
no longer the exception — staging/marts/public each holding `orders` or
`customers` is the normal shape of the dbt warehouse this change was made for.
So the earlier commit does not introduce this bug, but it does move it from
unlikely to expected, which is why it is fixed on the same branch.

Also hoists resolveSchema() out of the argument list: this method runs once per
table, and each call costs a round-trip.

Tests: 12/12. A mocked ResultSet cannot exercise SQL semantics, so the new test
asserts the two schema predicates are present and that every placeholder is
bound — enough to stop the qualification being dropped again. The correctness
evidence is the measurement above, run against Postgres 18.

Author: geekypunk

backend/src/main/java/com/dbaagent/provider/postgres/PostgresIntrospectionProvider.java

@@ -156,24 +156,52 @@ SELECT p.proname as name, pg_get_functiondef(p.oid) as definition
     public List<ColumnInfo> getTableColumns(Connection connection, String database, String tableName) throws SQLException {
         List<ColumnInfo> columns = new ArrayList<>();
 
+        // The PK subquery must be schema-qualified on BOTH the constraint join and
+        // the table lookup. Without it, two schemas holding a same-named table
+        // collide by construction: Postgres auto-names primary keys
+        // "<table>_pkey", so `tc.constraint_name = ku.constraint_name` alone
+        // cross-joins the schemas.
+        //
+        // Measured against two schemas each holding an `orders` table, asking for
+        // s_b.orders whose only PK is `name`:
+        //     name  | t
+        //     name  | t     <- duplicated row
+        //     other | t     <- false positive; `other` is s_a's PK, not s_b's
+        //     other | t     <- duplicated row
+        // i.e. every column reported as a primary key, and each one twice. With the
+        // predicates below the same query returns `name | t`, `other | f`.
+        //
+        // The bug predates the search_path change but was mostly latent while this
+        // provider only ever read `public`. Now that it targets whatever the session
+        // resolves to, same-named tables across schemas — the normal shape of a dbt
+        // warehouse (staging/marts/public each with `orders`) — are the expected
+        // case rather than the exception.
         String query = """
             SELECT c.column_name, c.data_type, c.is_nullable, c.column_default,
                 CASE WHEN pk.column_name IS NOT NULL THEN true ELSE false END as is_primary_key
             FROM information_schema.columns c
             LEFT JOIN (
                 SELECT ku.column_name
                 FROM information_schema.table_constraints tc
-                JOIN information_schema.key_column_usage ku ON tc.constraint_name = ku.constraint_name
-                WHERE tc.constraint_type = 'PRIMARY KEY' AND ku.table_name = ?
+                JOIN information_schema.key_column_usage ku
+                    ON tc.constraint_name = ku.constraint_name
+                    AND tc.table_schema = ku.table_schema
+                WHERE tc.constraint_type = 'PRIMARY KEY'
+                    AND ku.table_name = ? AND ku.table_schema = ?
             ) pk ON c.column_name = pk.column_name
             WHERE c.table_name = ? AND c.table_schema = ?
             ORDER BY c.ordinal_position
             """;
 
+        // Resolved once: this method runs per table, and resolveSchema() costs a
+        // round-trip each call.
+        String schema = resolveSchema(connection);
+
         try (PreparedStatement stmt = connection.prepareStatement(query)) {
             stmt.setString(1, tableName);
-            stmt.setString(2, tableName);
-            stmt.setString(3, resolveSchema(connection));
+            stmt.setString(2, schema);
+            stmt.setString(3, tableName);
+            stmt.setString(4, schema);
             try (ResultSet rs = stmt.executeQuery()) {
                 while (rs.next()) {
                     ColumnInfo col = new ColumnInfo();

backend/src/test/java/com/dbaagent/provider/postgres/PostgresIntrospectionProviderTest.java

@@ -198,6 +198,38 @@ void getTableStats_returnsStats() throws SQLException {
         assertEquals(122880L, stats.getSizeBytes());
     }
 
+    @Test
+    void getTableColumns_qualifiesThePrimaryKeySubqueryBySchema() throws SQLException {
+        // Postgres auto-names primary keys "<table>_pkey", so joining
+        // tc.constraint_name = ku.constraint_name WITHOUT a schema predicate
+        // cross-joins any two schemas holding a same-named table. Measured against
+        // two `orders` tables (s_a PK `other`, s_b PK `name`), asking for s_b:
+        //     name|t  name|t  other|t  other|t
+        // — every column a primary key, and each row duplicated. Schema-qualified
+        // it returns name|t, other|f.
+        //
+        // A mocked ResultSet cannot exercise SQL semantics, so this asserts the
+        // predicates are present and every placeholder is bound — enough to stop
+        // the qualification being dropped again.
+        ArgumentCaptor<String> sqlCaptor = ArgumentCaptor.forClass(String.class);
+        when(connection.prepareStatement(anyString())).thenReturn(preparedStatement);
+        when(preparedStatement.executeQuery()).thenReturn(resultSet);
+        when(resultSet.next()).thenReturn(false);
+
+        provider.getTableColumns(connection, "public", "orders");
+
+        verify(connection).prepareStatement(sqlCaptor.capture());
+        String sql = sqlCaptor.getValue();
+
+        assertTrue(sql.contains("tc.table_schema = ku.table_schema"),
+            "the constraint join must be schema-qualified, or <table>_pkey collides across schemas");
+        assertTrue(sql.contains("ku.table_schema = ?"),
+            "the PK lookup must be restricted to the target schema");
+
+        int placeholders = (int) sql.chars().filter(c -> c == '?').count();
+        verify(preparedStatement, times(placeholders)).setString(anyInt(), anyString());
+    }
+
     @Test
     void getTableStats_bindsEveryPlaceholderInTheStatsQuery() throws SQLException {
         // The stats query carried NINE `?` placeholders (the two size subtractions use