diff --git a/apps/studio/components/interfaces/TableGridEditor/SidePanelEditor/SidePanelEditor.tsx b/apps/studio/components/interfaces/TableGridEditor/SidePanelEditor/SidePanelEditor.tsx
index 11118f700cb..6e5fa333474 100644
--- a/apps/studio/components/interfaces/TableGridEditor/SidePanelEditor/SidePanelEditor.tsx
+++ b/apps/studio/components/interfaces/TableGridEditor/SidePanelEditor/SidePanelEditor.tsx
@@ -443,6 +443,11 @@ export const SidePanelEditor = ({
         queryClient.invalidateQueries({
           queryKey: tableKeys.infiniteListPrefix(project?.ref, selectedTable?.schema),
         }),
+        // useTableQuery (FK selectors/formatters) has a 5min staleTime -- without this,
+        // an edited column can serve stale data to any FK target cell pointing at this table.
+        queryClient.invalidateQueries({
+          queryKey: tableKeys.retrieve(project?.ref, selectedTable?.name, selectedTable?.schema),
+        }),
       ])
 
       // We need to invalidate tableRowsAndCount after tableEditor
diff --git a/apps/studio/components/interfaces/TableGridEditor/SidePanelEditor/SidePanelEditor.utils.tsx b/apps/studio/components/interfaces/TableGridEditor/SidePanelEditor/SidePanelEditor.utils.tsx
index 6583e11a2c7..66b6b9a3243 100644
--- a/apps/studio/components/interfaces/TableGridEditor/SidePanelEditor/SidePanelEditor.utils.tsx
+++ b/apps/studio/components/interfaces/TableGridEditor/SidePanelEditor/SidePanelEditor.utils.tsx
@@ -870,6 +870,19 @@ export const updateTable = async ({
       queryKey: tableKeys.list(projectRef, table.schema, { includeColumns: true }),
     }),
     queryClient.invalidateQueries({ queryKey: lintKeys.lint(projectRef) }),
+    // useTableQuery (FK selectors/formatters) has a 5min staleTime, and the columns/FKs
+    // above were only just applied after `updatedTable` was fetched -- refresh its identity,
+    // plus the pre-rename one too, so nothing serves stale/deleted columns until then.
+    queryClient.invalidateQueries({
+      queryKey: tableKeys.retrieve(projectRef, updatedTable.name, updatedTable.schema),
+    }),
+    ...(updatedTable.name !== table.name || updatedTable.schema !== table.schema
+      ? [
+          queryClient.invalidateQueries({
+            queryKey: tableKeys.retrieve(projectRef, table.name, table.schema),
+          }),
+        ]
+      : []),
   ])
 
   // We need to invalidate tableRowsAndCount after tableEditor
diff --git a/apps/studio/components/layouts/DefaultLayout.tsx b/apps/studio/components/layouts/DefaultLayout.tsx
index 2f29bf4016a..2c35cca6353 100644
--- a/apps/studio/components/layouts/DefaultLayout.tsx
+++ b/apps/studio/components/layouts/DefaultLayout.tsx
@@ -16,6 +16,7 @@ import {
 import { ProjectContextProvider } from './ProjectLayout/ProjectContext'
 import { AppBannerWrapper } from '@/components/interfaces/App/AppBannerWrapper'
 import { Sidebar } from '@/components/interfaces/Sidebar'
+import { useSyncScopedIntrospection } from '@/data/scoped-introspection'
 import { useLastVisitedOrganization } from '@/hooks/misc/useLastVisitedOrganization'
 import { useCheckLatestDeploy } from '@/hooks/use-check-latest-deploy'
 import { IS_PLATFORM } from '@/lib/constants'
@@ -42,6 +43,7 @@ export const DefaultLayout = ({
   headerTitle,
   hideMobileMenu,
 }: PropsWithChildren<DefaultLayoutProps>) => {
+  useSyncScopedIntrospection()
   useCheckLatestDeploy()
 
   const { ref } = useParams()
diff --git a/apps/studio/data/database-columns/database-column-delete-mutation.ts b/apps/studio/data/database-columns/database-column-delete-mutation.ts
index dfe2745cd2f..2ad154eb8ba 100644
--- a/apps/studio/data/database-columns/database-column-delete-mutation.ts
+++ b/apps/studio/data/database-columns/database-column-delete-mutation.ts
@@ -8,6 +8,7 @@ import { entityTypeKeys } from '@/data/entity-types/keys'
 import { executeSql } from '@/data/sql/execute-sql-mutation'
 import { tableEditorKeys } from '@/data/table-editor/keys'
 import { tableRowKeys } from '@/data/table-rows/keys'
+import { tableKeys } from '@/data/tables/keys'
 import { viewKeys } from '@/data/views/keys'
 import type { ResponseError, UseCustomMutationOptions } from '@/types'
 
@@ -68,6 +69,11 @@ export const useDatabaseColumnDeleteMutation = ({
         queryClient.invalidateQueries({
           queryKey: viewKeys.listBySchema(projectRef, [column.schema]),
         }),
+        // useTableQuery (FK selectors/formatters) has a 5min staleTime -- without this,
+        // it can keep serving the deleted column to any FK target cell pointing at this table.
+        queryClient.invalidateQueries({
+          queryKey: tableKeys.retrieve(projectRef, column.table, column.schema),
+        }),
       ])
 
       // We need to invalidate tableRowsAndCount after tableEditor
diff --git a/apps/studio/data/enumerated-types/enumerated-types-query.ts b/apps/studio/data/enumerated-types/enumerated-types-query.ts
index b494ac83e3f..71d2b43e968 100644
--- a/apps/studio/data/enumerated-types/enumerated-types-query.ts
+++ b/apps/studio/data/enumerated-types/enumerated-types-query.ts
@@ -4,6 +4,7 @@ import { useQuery } from '@tanstack/react-query'
 import { executeSql } from '../sql/execute-sql-mutation'
 import { enumeratedTypesKeys } from './keys'
 import type { components } from '@/data/api'
+import { isScopedIntrospection, scopedIntrospectionReady } from '@/data/scoped-introspection'
 import type { ResponseError, UseCustomQueryOptions } from '@/types'
 
 export type EnumeratedTypesVariables = {
@@ -20,7 +21,9 @@ export async function getEnumeratedTypes(
 ) {
   if (!projectRef) throw new Error('projectRef is required')
 
-  const { sql } = pgMeta.types.list({ includedSchemas: schemas })
+  // Cold-load race guard -- see the module comment on scoped-introspection.ts.
+  await scopedIntrospectionReady()
+  const { sql } = pgMeta.types.list({ includedSchemas: schemas, scoped: isScopedIntrospection() })
   const { result } = await executeSql(
     {
       projectRef,
diff --git a/apps/studio/data/privileges/table-api-access-query.ts b/apps/studio/data/privileges/table-api-access-query.ts
index 43e289f68b9..84a24b72e5d 100644
--- a/apps/studio/data/privileges/table-api-access-query.ts
+++ b/apps/studio/data/privileges/table-api-access-query.ts
@@ -150,7 +150,7 @@ export const useTableApiAccessQuery = (
 
   const enablePrivilegesQuery = enabled && hasTables
   const privilegeStatus = useTablePrivilegesQuery(
-    { projectRef, connectionString },
+    { projectRef, connectionString, includedSchemas: [schemaName] },
     { enabled: enablePrivilegesQuery, ...options }
   )
 
diff --git a/apps/studio/data/privileges/table-privileges-query.ts b/apps/studio/data/privileges/table-privileges-query.ts
index 4c9c6bf7f14..a09611363b2 100644
--- a/apps/studio/data/privileges/table-privileges-query.ts
+++ b/apps/studio/data/privileges/table-privileges-query.ts
@@ -3,6 +3,7 @@ import { QueryClient, useQuery } from '@tanstack/react-query'
 import { z } from 'zod'
 
 import { privilegeKeys } from './keys'
+import { isScopedIntrospection, scopedIntrospectionReady } from '@/data/scoped-introspection'
 import { executeSql } from '@/data/sql/execute-sql-mutation'
 import { ResponseError, UseCustomQueryOptions } from '@/types'
 
@@ -22,7 +23,12 @@ async function getTablePrivileges(
   { projectRef, connectionString, includedSchemas }: TablePrivilegesVariables,
   signal?: AbortSignal
 ) {
-  const sql = pgMeta.tablePrivileges.list({ includedSchemas }).sql
+  // Cold-load race guard -- see the module comment on scoped-introspection.ts.
+  await scopedIntrospectionReady()
+  const sql = pgMeta.tablePrivileges.list({
+    includedSchemas,
+    scoped: isScopedIntrospection(),
+  }).sql
   const queryKey = ['table-privileges', includedSchemas?.join(',')]
 
   const { result } = await executeSql({ projectRef, connectionString, sql, queryKey }, signal)
diff --git a/apps/studio/data/scoped-introspection.test.tsx b/apps/studio/data/scoped-introspection.test.tsx
new file mode 100644
index 00000000000..89081efeb52
--- /dev/null
+++ b/apps/studio/data/scoped-introspection.test.tsx
@@ -0,0 +1,112 @@
+import { renderHook } from '@testing-library/react'
+import { FeatureFlagContext } from 'common'
+import { afterEach, beforeEach, describe, expect, test, vi } from 'vitest'
+
+import { PG_META_SCOPED_INTROSPECTION_FLAG } from '@/data/table-editor/table-editor-query'
+
+const { mockIsPlatform } = vi.hoisted(() => ({ mockIsPlatform: { value: true } }))
+
+vi.mock('@/lib/constants', async () => {
+  const actual = await vi.importActual<Record<string, unknown>>('@/lib/constants')
+  return {
+    ...actual,
+    get IS_PLATFORM() {
+      return mockIsPlatform.value
+    },
+  }
+})
+
+// Mirrors the real provider: `configcat` and `hasLoaded` flip together once
+// `processFlags()` resolves -- an empty store (the pre-load default) is what
+// makes `useFlag` read `false` before that.
+function flagProvider(hasLoaded: boolean, flagValue = true) {
+  const configcat: Record<string, boolean> = hasLoaded
+    ? { [PG_META_SCOPED_INTROSPECTION_FLAG]: flagValue }
+    : {}
+  return function FlagProviderWrapper({ children }: { children: React.ReactNode }) {
+    return (
+      <FeatureFlagContext.Provider value={{ configcat, posthog: {}, hasLoaded }}>
+        {children}
+      </FeatureFlagContext.Provider>
+    )
+  }
+}
+
+describe('scoped-introspection', () => {
+  // Module-level singleton state -- start every test from a fresh module
+  // instance so the ready-promise isn't already settled from a prior test.
+  let mod: typeof import('./scoped-introspection')
+
+  beforeEach(async () => {
+    mockIsPlatform.value = true
+    vi.resetModules()
+    mod = await import('./scoped-introspection')
+  })
+
+  afterEach(() => {
+    vi.useRealTimers()
+  })
+
+  test('isScopedIntrospection defaults to false', () => {
+    expect(mod.isScopedIntrospection()).toBe(false)
+  })
+
+  test('setScopedIntrospection updates the accessor', () => {
+    mod.setScopedIntrospection(true)
+    expect(mod.isScopedIntrospection()).toBe(true)
+  })
+
+  test('scopedIntrospectionReady stays pending until a loaded flag store hydrates it', async () => {
+    let settled = false
+    mod.scopedIntrospectionReady().then(() => {
+      settled = true
+    })
+
+    const { unmount } = renderHook(() => mod.useSyncScopedIntrospection(), {
+      wrapper: flagProvider(false),
+    })
+    await Promise.resolve()
+    expect(settled).toBe(false)
+    expect(mod.isScopedIntrospection()).toBe(false) // useFlag reads false pre-load
+    unmount()
+
+    renderHook(() => mod.useSyncScopedIntrospection(), { wrapper: flagProvider(true, true) })
+    await Promise.resolve()
+
+    expect(settled).toBe(true)
+    expect(mod.isScopedIntrospection()).toBe(true)
+  })
+
+  test('resolves immediately when self-hosted (flags disabled)', async () => {
+    mockIsPlatform.value = false
+
+    let settled = false
+    mod.scopedIntrospectionReady().then(() => {
+      settled = true
+    })
+
+    renderHook(() => mod.useSyncScopedIntrospection(), { wrapper: flagProvider(false, false) })
+    await Promise.resolve()
+
+    expect(settled).toBe(true)
+  })
+
+  test('falls back to resolving 5s after the first scopedIntrospectionReady() call if hydration never happens', async () => {
+    vi.useFakeTimers()
+
+    let settled = false
+    mod.scopedIntrospectionReady().then(() => {
+      settled = true
+    })
+
+    vi.advanceTimersByTime(4_999)
+    await Promise.resolve()
+    expect(settled).toBe(false)
+
+    vi.advanceTimersByTime(1)
+    await Promise.resolve()
+    expect(settled).toBe(true)
+    // Hydration never ran -- accessor stays at its default.
+    expect(mod.isScopedIntrospection()).toBe(false)
+  })
+})
diff --git a/apps/studio/data/scoped-introspection.ts b/apps/studio/data/scoped-introspection.ts
new file mode 100644
index 00000000000..00d801a089f
--- /dev/null
+++ b/apps/studio/data/scoped-introspection.ts
@@ -0,0 +1,72 @@
+import { useFeatureFlags, useFlag } from 'common'
+import { useEffect } from 'react'
+
+import { PG_META_SCOPED_INTROSPECTION_FLAG } from '@/data/table-editor/table-editor-query'
+import { IS_PLATFORM } from '@/lib/constants'
+
+/**
+ * Imperative accessor for the `pgMetaScopedIntrospection` flag, read by query
+ * fns that opt into scoped pg-meta SQL without threading `scoped` through
+ * their React Query key (accepted tradeoff: a mid-session flag flip can serve
+ * stale-keyed cached data until the query's next natural refetch). Hydrated
+ * by `useSyncScopedIntrospection`, called from `DefaultLayout` -- the first
+ * component both of Studio's root trees (pages/_app.tsx, routes/__root.tsx)
+ * render inside `FeatureFlagProviderWithOrgContext`.
+ */
+let scopedIntrospection = false
+
+export const setScopedIntrospection = (value: boolean) => {
+  scopedIntrospection = value
+}
+
+export const isScopedIntrospection = () => scopedIntrospection
+
+const READY_TIMEOUT_MS = 5_000
+
+let isReadySettled = false
+let isTimeoutArmed = false
+let resolveReady: () => void = () => {}
+const readyPromise = new Promise<void>((resolve) => {
+  resolveReady = resolve
+})
+
+const markScopedIntrospectionReady = () => {
+  if (isReadySettled) return
+  isReadySettled = true
+  resolveReady()
+}
+
+/**
+ * Resolves once `useSyncScopedIntrospection` has hydrated the accessor from a
+ * loaded flag store (or immediately if self-hosted) -- never before, by
+ * construction (see that hook). Arms a 5s fallback timer on its *first call*
+ * rather than at module load, so a query firing long after boot isn't bound
+ * by a timer that already expired before hydration got a chance to run.
+ */
+export const scopedIntrospectionReady = (): Promise<void> => {
+  if (!isTimeoutArmed) {
+    isTimeoutArmed = true
+    setTimeout(markScopedIntrospectionReady, READY_TIMEOUT_MS)
+  }
+  return readyPromise
+}
+
+/**
+ * Syncs the flag's current value into `isScopedIntrospection()`. Call once
+ * near the root, under `FeatureFlagProviderWithOrgContext` (currently
+ * `DefaultLayout`).
+ */
+export const useSyncScopedIntrospection = () => {
+  const { hasLoaded } = useFeatureFlags()
+  const scoped = !!useFlag(PG_META_SCOPED_INTROSPECTION_FLAG)
+
+  useEffect(() => {
+    setScopedIntrospection(scoped)
+
+    // Self-hosted disables the flag provider (enabled={IS_PLATFORM}), so
+    // hasLoaded never flips true -- there's nothing to wait for there.
+    if (hasLoaded || !IS_PLATFORM) {
+      markScopedIntrospectionReady()
+    }
+  }, [scoped, hasLoaded])
+}
diff --git a/apps/studio/data/table-rows/table-rows-count-query.ts b/apps/studio/data/table-rows/table-rows-count-query.ts
index 636d4ef1406..d002da4aef5 100644
--- a/apps/studio/data/table-rows/table-rows-count-query.ts
+++ b/apps/studio/data/table-rows/table-rows-count-query.ts
@@ -74,6 +74,7 @@ export async function getTableRowsCount(
       filters: formattedFilters,
       enforceExactCount,
       isReadOnlyContext,
+      scoped,
     }),
     roleImpersonationState
   )
@@ -111,7 +112,7 @@ export const useTableRowsCountQuery = <TData = TableRowsCountData>(
     identifier: readReplicaIdentifier,
     type,
   } = useConnectionStringForReadOps()
-  const { can: canSQLAdminWrite } = useAsyncCheckPermissions(
+  const { can: canSQLAdminWrite, isLoading: isPermissionsLoading } = useAsyncCheckPermissions(
     PermissionAction.TENANT_SQL_ADMIN_WRITE,
     'tables'
   )
@@ -141,7 +142,15 @@ export const useTableRowsCountQuery = <TData = TableRowsCountData>(
       enabled &&
       typeof projectRef !== 'undefined' &&
       typeof tableId !== 'undefined' &&
-      (!IS_PLATFORM || typeof connectionString !== 'undefined'),
+      (!IS_PLATFORM || typeof connectionString !== 'undefined') &&
+      // isReadOnlyContext resolves to `type === 'replica' || !canSQLAdminWrite`: for
+      // read replicas it's already known synchronously, but otherwise it depends on
+      // canSQLAdminWrite, which starts out `false` while permissions are loading.
+      // Firing while that's still in flight would cache a transient
+      // isReadOnlyContext:true (and, on a never-analyzed table, a scoped
+      // count:-1/is_estimate:true) for what may actually be a writable user. Wait
+      // for the permission check to settle before firing in that case.
+      (type === 'replica' || !isPermissionsLoading),
     ...options,
   })
 }
diff --git a/apps/studio/data/tables/table-retrieve-query.ts b/apps/studio/data/tables/table-retrieve-query.ts
index 7df19581b23..55e855238f3 100644
--- a/apps/studio/data/tables/table-retrieve-query.ts
+++ b/apps/studio/data/tables/table-retrieve-query.ts
@@ -3,6 +3,7 @@ import { useQuery } from '@tanstack/react-query'
 
 import { tableKeys } from './keys'
 import { getQueryClient } from '@/data/query-client'
+import { isScopedIntrospection, scopedIntrospectionReady } from '@/data/scoped-introspection'
 import { executeSql } from '@/data/sql/execute-sql-mutation'
 import type { SafePostgresTable } from '@/lib/postgres-types'
 import type { ResponseError, UseCustomQueryOptions } from '@/types'
@@ -18,7 +19,9 @@ export async function getTable(
   { projectRef, connectionString, name, schema }: TablesVariables,
   signal?: AbortSignal
 ): Promise<SafePostgresTable> {
-  const { sql, zod } = pgMeta.tables.retrieve({ name, schema })
+  // Cold-load race guard -- see the module comment on scoped-introspection.ts.
+  await scopedIntrospectionReady()
+  const { sql, zod } = pgMeta.tables.retrieve({ name, schema, scoped: isScopedIntrospection() })
 
   const { result } = await executeSql(
     {
@@ -48,6 +51,9 @@ export const useTableQuery = <TData = RetrieveTableResult>(
     queryKey: tableKeys.retrieve(projectRef, name, schema),
     queryFn: ({ signal }) => getTable({ projectRef, connectionString, name, schema }, signal),
     enabled: enabled && typeof projectRef !== 'undefined',
+    refetchOnWindowFocus: false,
+    retryOnMount: false,
+    staleTime: 5 * 60 * 1000,
     ...options,
   })
 }
diff --git a/examples/product-sample-supabase-kt/.gitignore b/examples/product-sample-supabase-kt/.gitignore
new file mode 100644
index 00000000000..aa8487e5205
--- /dev/null
+++ b/examples/product-sample-supabase-kt/.gitignore
@@ -0,0 +1,3 @@
+.gradle/
+build/
+local.properties
diff --git a/packages/pg-meta/src/helpers.ts b/packages/pg-meta/src/helpers.ts
index 36c7de29ce9..4ef7b378b40 100644
--- a/packages/pg-meta/src/helpers.ts
+++ b/packages/pg-meta/src/helpers.ts
@@ -1,11 +1,19 @@
 import { ident, joinSqlFragments, literal, safeSql, type SafeSqlFragment } from './pg-format'
 
-export const coalesceRowsToArray = (source: string, filter: SafeSqlFragment) => {
+export const coalesceRowsToArray = (
+  source: string,
+  filter: SafeSqlFragment,
+  // Optional ORDER BY (a column reference of `source`) applied INSIDE array_agg
+  // to make the emitted array deterministic. Omit it to keep the historical
+  // (plan-order) rendering byte-for-byte identical for existing callers.
+  orderBy?: SafeSqlFragment
+) => {
+  const orderClause = orderBy ? safeSql` ORDER BY ${orderBy}` : safeSql``
   return safeSql`
 COALESCE(
   (
     SELECT
-      array_agg(row_to_json(${ident(source)})) FILTER (WHERE ${filter})
+      array_agg(row_to_json(${ident(source)})${orderClause}) FILTER (WHERE ${filter})
     FROM
       ${ident(source)}
   ),
diff --git a/packages/pg-meta/src/pg-meta-table-privileges.ts b/packages/pg-meta/src/pg-meta-table-privileges.ts
index 984a1b9b22b..28a8d0d2cd5 100644
--- a/packages/pg-meta/src/pg-meta-table-privileges.ts
+++ b/packages/pg-meta/src/pg-meta-table-privileges.ts
@@ -10,7 +10,7 @@ import {
   safeSql,
   type SafeSqlFragment,
 } from './pg-format'
-import { TABLE_PRIVILEGES_SQL } from './sql/table-privileges'
+import { getScopedTablePrivilegesSql, TABLE_PRIVILEGES_SQL } from './sql/table-privileges'
 
 const pgTablePrivilegesZod = z.object({
   relation_id: z.number(),
@@ -50,26 +50,48 @@ function list({
   excludedSchemas,
   limit,
   offset,
+  scoped = false,
 }: {
   includeSystemSchemas?: boolean
   includedSchemas?: string[]
   excludedSchemas?: string[]
   limit?: number
   offset?: number
+  scoped?: boolean
 } = {}): {
   sql: SafeSqlFragment
   zod: typeof pgTablePrivilegesArrayZod
 } {
-  let sql = safeSql`
-with table_privileges as (${TABLE_PRIVILEGES_SQL})
-select *
-from table_privileges
-`
   const filter = filterByList(
     includedSchemas,
     excludedSchemas,
     !includeSystemSchemas ? DEFAULT_SYSTEM_SCHEMAS : undefined
   )
+  // Scoped path: push the schema filter into the base query's WHERE (before the
+  // aclexplode lateral / GROUP BY) instead of filtering the aggregated CTE.
+  if (scoped) {
+    const base = getScopedTablePrivilegesSql(filter ? safeSql`and nc.nspname ${filter}` : undefined)
+    let sql = safeSql`
+with table_privileges as (${base})
+select *
+from table_privileges
+`
+    if (limit) {
+      sql = safeSql`${sql} limit ${literal(limit)}`
+    }
+    if (offset) {
+      sql = safeSql`${sql} offset ${literal(offset)}`
+    }
+    return {
+      sql,
+      zod: pgTablePrivilegesArrayZod,
+    }
+  }
+  let sql = safeSql`
+with table_privileges as (${TABLE_PRIVILEGES_SQL})
+select *
+from table_privileges
+`
   if (filter) {
     sql = safeSql`${sql} where schema ${filter}`
   }
@@ -85,11 +107,11 @@ from table_privileges
   }
 }
 
-function retrieve({ id }: { id: number }): {
+function retrieve({ id, scoped }: { id: number; scoped?: boolean }): {
   sql: SafeSqlFragment
   zod: typeof pgTablePrivilegesOptionalZod
 }
-function retrieve({ name, schema }: { name: string; schema?: string }): {
+function retrieve({ name, schema, scoped }: { name: string; schema?: string; scoped?: boolean }): {
   sql: SafeSqlFragment
   zod: typeof pgTablePrivilegesOptionalZod
 }
@@ -97,14 +119,33 @@ function retrieve({
   id,
   name,
   schema = 'public',
+  scoped = false,
 }: {
   id?: number
   name?: string
   schema?: string
+  scoped?: boolean
 }): {
   sql: SafeSqlFragment
   zod: typeof pgTablePrivilegesOptionalZod
 } {
+  // Scoped path: push the oid / schema+name predicate into the base query's
+  // WHERE (before the aclexplode lateral / GROUP BY) so pg_class is pruned to
+  // the target relation instead of scanning every relation then filtering.
+  if (scoped) {
+    const scopeFilter = id
+      ? safeSql`and c.oid = ${literal(id)}`
+      : safeSql`and nc.nspname = ${literal(schema)} and c.relname = ${literal(name)}`
+    const sql = /* SQL */ safeSql`
+with table_privileges as (${getScopedTablePrivilegesSql(scopeFilter)})
+select *
+from table_privileges
+`
+    return {
+      sql,
+      zod: pgTablePrivilegesOptionalZod,
+    }
+  }
   if (id) {
     const sql = /* SQL */ safeSql`
 with table_privileges as (${TABLE_PRIVILEGES_SQL})
diff --git a/packages/pg-meta/src/pg-meta-tables.ts b/packages/pg-meta/src/pg-meta-tables.ts
index c2305894324..7209b51ae97 100644
--- a/packages/pg-meta/src/pg-meta-tables.ts
+++ b/packages/pg-meta/src/pg-meta-tables.ts
@@ -11,8 +11,8 @@ import {
   type SafeSqlFragment,
 } from './pg-format'
 import { pgColumnArrayZod } from './pg-meta-columns'
-import { COLUMNS_SQL } from './sql/columns'
-import { TABLES_SQL } from './sql/tables'
+import { COLUMNS_SQL, getColumnsSql } from './sql/columns'
+import { getTablesSql, TABLES_SQL } from './sql/tables'
 
 const pgTablePrimaryKeyZod = z.object({
   table_id: z.number(),
@@ -115,12 +115,48 @@ function list<T extends boolean | undefined = true>(
   }
 }
 
-function retrieve(identifier: TableIdentifier): {
+function retrieve(identifier: TableIdentifier & { scoped?: boolean }): {
   sql: SafeSqlFragment
   zod: z.ZodType<TableWithColumns>
 } {
   let whereClause = getIdentifierWhereClause(identifier)
 
+  // Scoped path: resolve the target relation's OID once (via pg_class's
+  // (relname, relnamespace) index for the name+schema branch, or the literal
+  // for the id branch) and push it into BOTH enrichment CTEs, so the query
+  // computes sizes / PKs / relationships / columns only for the requested
+  // relation instead of the entire catalog (which timed out at ~58s on a
+  // hundreds-of-schemas database). Output rows are identical -- the legacy outer
+  // identifier filter already restricted the result to this relation.
+  if (identifier.scoped) {
+    // Resolve the target OID as a scalar the planner evaluates once (initplan):
+    // a literal for the id branch, or an uncorrelated scalar subquery resolving
+    // via pg_class's (relname, relnamespace) index for the name+schema branch.
+    let targetOid: SafeSqlFragment
+    if ('id' in identifier && identifier.id) {
+      targetOid = safeSql`${literal(identifier.id)}`
+    } else if ('name' in identifier && identifier.name && identifier.schema) {
+      targetOid = safeSql`(select tc.oid from pg_class tc join pg_namespace tn on tn.oid = tc.relnamespace where tc.relname = ${literal(identifier.name)} and tn.nspname = ${literal(identifier.schema)})`
+    } else {
+      throw new Error('Must provide either id or name and schema')
+    }
+    const scopedTables = getTablesSql(targetOid)
+    const scopedColumns = getColumnsSql({
+      filter: { column: 'oid', predicate: safeSql`= ${targetOid}` },
+    })
+    const sql = safeSql`
+  with tables as (${scopedTables})
+  , columns as (${scopedColumns})
+  select
+    *
+    , ${coalesceRowsToArray('columns', safeSql`columns.table_id = tables.id`, safeSql`columns.ordinal_position`)}
+  from tables where ${whereClause};`
+    return {
+      sql,
+      zod: pgTableZod,
+    }
+  }
+
   const sql = safeSql`${generateEnrichedTablesSql({ includeColumns: true })} where ${whereClause};`
   return {
     sql,
diff --git a/packages/pg-meta/src/pg-meta-types.ts b/packages/pg-meta/src/pg-meta-types.ts
index 095db69a614..6185b4ac205 100644
--- a/packages/pg-meta/src/pg-meta-types.ts
+++ b/packages/pg-meta/src/pg-meta-types.ts
@@ -3,7 +3,7 @@ import { z } from 'zod'
 import { DEFAULT_SYSTEM_SCHEMAS } from './constants'
 import { filterByList } from './helpers'
 import { literal, safeSql, type SafeSqlFragment } from './pg-format'
-import { TYPES_SQL } from './sql/types'
+import { SCOPED_TYPES_SQL, TYPES_SQL } from './sql/types'
 
 const pgTypeZod = z.object({
   id: z.number(),
@@ -30,6 +30,7 @@ function list({
   excludedSchemas,
   limit,
   offset,
+  scoped = false,
 }: {
   includeArrayTypes?: boolean
   includeSystemSchemas?: boolean
@@ -37,11 +38,15 @@ function list({
   excludedSchemas?: string[]
   limit?: number
   offset?: number
+  scoped?: boolean
 } = {}): {
   sql: SafeSqlFragment
   zod: typeof pgTypeArrayZod
 } {
-  let sql = TYPES_SQL
+  // Both bases end at the same trailing WHERE `)`, so the filter/limit fragments
+  // below extend the same single-level WHERE regardless of which base is used.
+  // scoped=false keeps the rendered SQL byte-identical to the pre-change query.
+  let sql = scoped ? SCOPED_TYPES_SQL : TYPES_SQL
   if (!includeArrayTypes) {
     sql = safeSql`${sql} and not exists (
       select from pg_type el
@@ -57,6 +62,11 @@ function list({
   if (filter) {
     sql = safeSql`${sql} and n.nspname ${filter}`
   }
+  if (scoped) {
+    // Scoped only: legacy TYPES_SQL has no ORDER BY (plan-dependent order); sort
+    // the scoped result by t.oid for a stable, comparable order. Before LIMIT.
+    sql = safeSql`${sql} order by t.oid`
+  }
   if (limit) {
     sql = safeSql`${sql} limit ${literal(limit)}`
   }
diff --git a/packages/pg-meta/src/sql/studio/database/get-count-estimate.ts b/packages/pg-meta/src/sql/studio/database/get-count-estimate.ts
index 6ee03d3eb81..6a265bee4cb 100644
--- a/packages/pg-meta/src/sql/studio/database/get-count-estimate.ts
+++ b/packages/pg-meta/src/sql/studio/database/get-count-estimate.ts
@@ -2,6 +2,19 @@ import { safeSql, type SafeSqlFragment } from '../../../pg-format'
 
 export const THRESHOLD_COUNT = 50000
 
+/**
+ * Heap-size gate (bytes) for the never-analyzed (reltuples = -1) case, which
+ * covers BOTH an empty/small new table and a bulk-loaded huge one. Gate on the
+ * real heap size (pg_relation_size, + pg_partition_tree sum for a partitioned
+ * parent whose own size is 0; relpages is equally stale pre-vacuum): at/below ->
+ * exact count(*), above -> EXPLAIN estimate. ~10 MB = THRESHOLD_COUNT rows at a
+ * conservative ~200 bytes/row (an exact count over that is subsecond).
+ */
+export const THRESHOLD_ESTIMATE_BYTES = THRESHOLD_COUNT * 200
+
+// FROZEN legacy path: served while the pgMetaScopedIntrospection flag is off.
+// Do not edit -- it must keep matching production behavior until the flag
+// cleanup deletes it. The scoped path reuses this same function.
 export const COUNT_ESTIMATE_SQL: SafeSqlFragment = safeSql`
 CREATE OR REPLACE FUNCTION pg_temp.count_estimate(
     query text
diff --git a/packages/pg-meta/src/sql/studio/database/rows.ts b/packages/pg-meta/src/sql/studio/database/rows.ts
index ad6e7517453..8c826168a38 100644
--- a/packages/pg-meta/src/sql/studio/database/rows.ts
+++ b/packages/pg-meta/src/sql/studio/database/rows.ts
@@ -1,24 +1,29 @@
 import { literal, safeSql, type SafeSqlFragment } from '../../../pg-format'
 import { Filter, Query } from '../../../query'
-import { COUNT_ESTIMATE_SQL, THRESHOLD_COUNT } from './get-count-estimate'
+import { COUNT_ESTIMATE_SQL, THRESHOLD_COUNT, THRESHOLD_ESTIMATE_BYTES } from './get-count-estimate'
 
 /**
- * [Joshen] Initially check reltuples from pg_class for an estimate of row count on the table
- * - If reltuples = -1, table never been analyzed, assume small table -> return exact count
- * - If reltuples exceeds threshold, return estimate count
- * - Else return exact count
+ * Row-count for a table. reltuples = -1 (never analyzed) covers BOTH an
+ * empty/small new table and a freshly bulk-loaded huge one, so the opt-in
+ * `scoped` path gates on the real heap size (pg_relation_size, + pg_partition_tree
+ * sum for partitioned parents whose own size is 0; relpages is equally stale
+ * pre-vacuum): large -> EXPLAIN estimate, small/empty -> exact count. scoped=false
+ * is byte-identical to the legacy query. Full matrix in rows-count.test.ts.
  */
 export const getTableRowsCountSql = ({
   table,
   filters = [],
   enforceExactCount = false,
   isReadOnlyContext = false,
+  scoped = false,
 }: {
   table: any
   filters?: Filter[]
   enforceExactCount?: boolean
   /** Skips using the count estimate function if true and fallsback to checking reltuples from pg_class  */
   isReadOnlyContext?: boolean
+  /** Opt-in optimized counting; gates never-analyzed tables on real heap size. */
+  scoped?: boolean
 }): SafeSqlFragment => {
   if (!table) return safeSql``
 
@@ -61,6 +66,40 @@ export const getTableRowsCountSql = ({
       : countBaseSql
 
     if (isReadOnlyContext) {
+      if (scoped) {
+        // Readonly can't create the pg_temp function: an over-threshold or
+        // physically-large never-analyzed table reports -1 (is_estimate=true);
+        // a small/empty one still gets an exact count. CASE and flag share the
+        // condition.
+        const sql = safeSql`
+with approximation as (
+    select
+      reltuples as estimate,
+      -- Whole-tree heap size. A partitioned PARENT (relkind 'p') has no storage
+      -- of its own, so its size is the sum over pg_partition_tree; every other
+      -- relkind uses its own heap directly (pg_partition_tree returns NO rows
+      -- for a plain non-partitioned table, so it cannot be used unconditionally).
+      -- Views/foreign tables yield 0 (-> exact count, unchanged behavior).
+      case when relkind = 'p'
+        then (select coalesce(sum(pg_relation_size(relid)), 0) from pg_partition_tree(oid))
+        else pg_relation_size(oid)
+      end as bytes
+    from pg_class
+    where oid = ${literal(table.id)}
+)
+select
+  case
+    when estimate > ${literal(THRESHOLD_COUNT)} or (estimate = -1 and bytes > ${literal(THRESHOLD_ESTIMATE_BYTES)}) then -1
+    else (${countBaseSqlWithoutSemicolon})
+  end as count,
+  (estimate > ${literal(THRESHOLD_COUNT)} or (estimate = -1 and bytes > ${literal(THRESHOLD_ESTIMATE_BYTES)})) as is_estimate
+from approximation;
+`
+
+        return sql
+      }
+      // FROZEN legacy path (pgMetaScopedIntrospection off): do not edit -- it
+      // must keep matching production behavior until the flag cleanup deletes it.
       const sql = safeSql`
 with approximation as (
     select reltuples as estimate
@@ -78,6 +117,45 @@ from approximation;
 
       return sql
     } else {
+      if (scoped) {
+        // estimate = -1 (never analyzed) gated on heap size (see CTE): large ->
+        // EXPLAIN estimate, small/empty -> exact count (avoids Postgres's ~10-page
+        // phantom estimate). Over-threshold keeps legacy behavior. CASE and flag
+        // share the condition. literal() quotes the embedded select so backslash
+        // identifiers survive under any standard_conforming_strings.
+        const estimateExpr = safeSql`pg_temp.count_estimate(${literal(selectBaseSqlWithoutSemicolon)})`
+        const sql = safeSql`
+${COUNT_ESTIMATE_SQL}
+
+with approximation as (
+    select
+      reltuples as estimate,
+      -- Whole-tree heap size. A partitioned PARENT (relkind 'p') has no storage
+      -- of its own, so its size is the sum over pg_partition_tree; every other
+      -- relkind uses its own heap directly (pg_partition_tree returns NO rows
+      -- for a plain non-partitioned table, so it cannot be used unconditionally).
+      -- Views/foreign tables yield 0 (-> exact count, unchanged behavior).
+      case when relkind = 'p'
+        then (select coalesce(sum(pg_relation_size(relid)), 0) from pg_partition_tree(oid))
+        else pg_relation_size(oid)
+      end as bytes
+    from pg_class
+    where oid = ${literal(table.id)}
+)
+select
+  case
+    when estimate = -1 and bytes > ${literal(THRESHOLD_ESTIMATE_BYTES)} then ${estimateExpr}
+    when estimate > ${literal(THRESHOLD_COUNT)} then ${filters.length > 0 ? estimateExpr : safeSql`estimate`}
+    else (${countBaseSqlWithoutSemicolon})
+  end as count,
+  (estimate > ${literal(THRESHOLD_COUNT)} or (estimate = -1 and bytes > ${literal(THRESHOLD_ESTIMATE_BYTES)})) as is_estimate
+from approximation;
+`
+
+        return sql
+      }
+      // FROZEN legacy path (pgMetaScopedIntrospection off): do not edit -- it
+      // must keep matching production behavior until the flag cleanup deletes it.
       const sql = safeSql`
 ${COUNT_ESTIMATE_SQL}
 
diff --git a/packages/pg-meta/src/sql/studio/database/table-definition.ts b/packages/pg-meta/src/sql/studio/database/table-definition.ts
index 070921b0c6e..a7f234deb1d 100644
--- a/packages/pg-meta/src/sql/studio/database/table-definition.ts
+++ b/packages/pg-meta/src/sql/studio/database/table-definition.ts
@@ -24,6 +24,9 @@ import { joinSqlFragments, literal, safeSql, type SafeSqlFragment } from '../../
  * The two branches are kept as complete, standalone templates (no interpolated
  * conditional fragments) so each rendered statement is easy to read and diff.
  */
+// FROZEN legacy path: served while the pgMetaScopedIntrospection flag is off.
+// Do not edit -- it must keep matching production behavior until the flag
+// cleanup deletes it. SCOPED_PG_GET_TABLEDEF_SQL is the replacement.
 const LEGACY_PG_GET_TABLEDEF_SQL: SafeSqlFragment = safeSql`
   DROP TYPE IF EXISTS pg_temp.tabledefs CASCADE;
   CREATE TYPE pg_temp.tabledefs AS ENUM ('PKEY_INTERNAL','PKEY_EXTERNAL','FKEYS_INTERNAL', 'FKEYS_EXTERNAL', 'COMMENTS', 'FKEYS_NONE', 'INCLUDE_TRIGGERS', 'NO_TRIGGERS');
diff --git a/packages/pg-meta/src/sql/studio/table-editor/table.ts b/packages/pg-meta/src/sql/studio/table-editor/table.ts
index 98347cf9a55..32513011dd4 100644
--- a/packages/pg-meta/src/sql/studio/table-editor/table.ts
+++ b/packages/pg-meta/src/sql/studio/table-editor/table.ts
@@ -345,6 +345,8 @@ export const getTableEditorSql = ({
     left join columns c on b.id = c.table_id;
   `
     : safeSql`
+    -- FROZEN legacy path (pgMetaScopedIntrospection off): do not edit -- it must
+    -- keep matching production behavior until the flag cleanup deletes it.
     with base_table_info as (
         select
             c.oid::int8 as id,
diff --git a/packages/pg-meta/src/sql/table-privileges.ts b/packages/pg-meta/src/sql/table-privileges.ts
index bd2e031783d..fcf8d789af2 100644
--- a/packages/pg-meta/src/sql/table-privileges.ts
+++ b/packages/pg-meta/src/sql/table-privileges.ts
@@ -1,6 +1,10 @@
-import { safeSql } from '../pg-format'
+import { safeSql, type SafeSqlFragment } from '../pg-format'
 
 export const TABLE_PRIVILEGES_SQL = /* SQL */ safeSql`
+-- FROZEN legacy path: served while the pgMetaScopedIntrospection flag is off.
+-- Do not edit -- it must keep matching production behavior until the flag
+-- cleanup deletes it. getScopedTablePrivilegesSql is the replacement.
+--
 -- Despite the name \`table_privileges\`, this includes other kinds of relations:
 -- views, matviews, etc. "Relation privileges" just doesn't roll off the tongue.
 --
@@ -78,3 +82,74 @@ group by
   c.relname,
   c.relkind
 `
+
+/**
+ * Scoped variant of {@link TABLE_PRIVILEGES_SQL}: injects the wrapper's
+ * predicate into the base WHERE (before the aclexplode lateral / GROUP BY) so
+ * pg_class is pruned first, instead of scanning all relations then filtering.
+ * `scopeFilter` must be a `safeSql` fragment starting with `and ` (or empty)
+ * referencing base aliases `c` (pg_class) and `nc` (pg_namespace). Kept as a
+ * standalone template so the legacy constant stays byte-identical; rows match.
+ */
+export const getScopedTablePrivilegesSql = (
+  scopeFilter: SafeSqlFragment = safeSql``
+): SafeSqlFragment => safeSql`
+select
+  c.oid as relation_id,
+  nc.nspname as schema,
+  c.relname as name,
+  case
+    when c.relkind = 'r' then 'table'
+    when c.relkind = 'v' then 'view'
+    when c.relkind = 'm' then 'materialized_view'
+    when c.relkind = 'f' then 'foreign_table'
+    when c.relkind = 'p' then 'partitioned_table'
+  end as kind,
+  coalesce(
+    jsonb_agg(
+      jsonb_build_object(
+        'grantor', grantor.rolname,
+        'grantee', grantee.rolname,
+        'privilege_type', _priv.privilege_type,
+        'is_grantable', _priv.is_grantable
+      )
+    ) filter (where _priv is not null),
+    '[]'
+  ) as privileges
+from pg_class c
+join pg_namespace as nc
+  on nc.oid = c.relnamespace
+left join lateral (
+  select grantor, grantee, privilege_type, is_grantable
+  from aclexplode(coalesce(c.relacl, acldefault('r', c.relowner)))
+) as _priv on true
+left join pg_roles as grantor
+  on grantor.oid = _priv.grantor
+left join (
+  select
+    pg_roles.oid,
+    pg_roles.rolname
+  from pg_roles
+  union all
+  select
+    (0)::oid as oid, 'PUBLIC'
+) as grantee (oid, rolname)
+  on grantee.oid = _priv.grantee
+where c.relkind in ('r', 'v', 'm', 'f', 'p')
+  and not pg_is_other_temp_schema(c.relnamespace)
+  ${scopeFilter}
+  and (
+    pg_has_role(c.relowner, 'USAGE')
+    or has_table_privilege(
+      c.oid,
+      'SELECT, INSERT, UPDATE, DELETE, TRUNCATE, REFERENCES, TRIGGER'
+      || case when current_setting('server_version_num')::int4 >= 170000 then ', MAINTAIN' else '' end
+    )
+    or has_any_column_privilege(c.oid, 'SELECT, INSERT, UPDATE, REFERENCES')
+  )
+group by
+  c.oid,
+  nc.nspname,
+  c.relname,
+  c.relkind
+`
diff --git a/packages/pg-meta/src/sql/tables.ts b/packages/pg-meta/src/sql/tables.ts
index feb28fb487f..88db7ce6784 100644
--- a/packages/pg-meta/src/sql/tables.ts
+++ b/packages/pg-meta/src/sql/tables.ts
@@ -1,6 +1,47 @@
-import { safeSql } from '../pg-format'
+import { safeSql, type SafeSqlFragment } from '../pg-format'
 
-export const TABLES_SQL = /* SQL */ safeSql`
+/**
+ * Builder for the tables introspection query.
+ *
+ * `targetOid`, when provided, is a scalar SQL fragment yielding the single OID
+ * to restrict the query to -- either a literal (`123`) or an uncorrelated scalar
+ * subquery (`(select tc.oid from pg_class tc join ... where relname=..)`). It is
+ * compared with `=` so the planner evaluates it once as an initplan constant and
+ * drives INDEX scans on the base pg_class scan AND the primary-key /
+ * relationships subqueries, instead of computing sizes, PKs and FK relationships
+ * for the ENTIRE catalog. (A multiply-referenced CTE would be materialized and
+ * act as an optimization barrier, forcing seq scans -- hence a scalar.) The
+ * relationships filter keeps BOTH directions (conrelid OR confrelid), matching
+ * the outgoing/incoming FK rows the unscoped query would have matched by name.
+ *
+ * When `targetOid` is omitted the injected fragments are empty and the rendered
+ * SQL is the legacy full-catalog query -- `TABLES_SQL` below is exactly that
+ * rendering, so every existing consumer is unaffected. Behavioral equivalence
+ * between the scoped and unscoped forms is enforced by execution-based tests in
+ * test/tables.test.ts, not by a byte-for-byte SQL snapshot.
+ */
+export const getTablesSql = (targetOid?: SafeSqlFragment) => {
+  const mainScope = targetOid
+    ? safeSql`
+  AND c.oid = ${targetOid}`
+    : safeSql``
+  const pkScope = targetOid
+    ? safeSql`
+      and c.oid = ${targetOid}`
+    : safeSql``
+  const relScope = targetOid
+    ? safeSql`
+      and (c.conrelid = ${targetOid} or c.confrelid = ${targetOid})`
+    : safeSql``
+  // Scoped path only: deterministic relationships order (plan-order dependent
+  // otherwise). A composite FK expands to one entry per source×target column
+  // pair sharing constraint_name, so tie-break on the column names. Empty for
+  // legacy, keeping TABLES_SQL byte-for-byte unchanged.
+  const relOrder = targetOid
+    ? safeSql` order by relationships.constraint_name, relationships.source_column_name, relationships.target_column_name`
+    : safeSql``
+
+  return /* SQL */ safeSql`
 SELECT
   c.oid :: int8 AS id,
   nc.nspname AS schema,
@@ -22,7 +63,7 @@ SELECT
   obj_description(c.oid) AS comment,
   coalesce(pk.primary_keys, '[]') as primary_keys,
   coalesce(
-    jsonb_agg(relationships) filter (where relationships is not null),
+    jsonb_agg(relationships${relOrder}) filter (where relationships is not null),
     '[]'
   ) as relationships
 FROM
@@ -46,7 +87,7 @@ FROM
       join pg_namespace n on c.relnamespace = n.oid
       join pg_attribute a on a.attrelid = c.oid and a.attnum = any(i.indkey)
     where
-      i.indisprimary
+      i.indisprimary${pkScope}
     group by c.oid
   ) as pk
   on pk.table_id = c.oid
@@ -73,7 +114,7 @@ FROM
       join pg_namespace nta on cta.relnamespace = nta.oid
     ) on ta.attrelid = c.confrelid and ta.attnum = any (c.confkey)
     where
-      c.contype = 'f'
+      c.contype = 'f'${relScope}
   ) as relationships
   on (relationships.source_schema = nc.nspname and relationships.source_table_name = c.relname)
   or (relationships.target_table_schema = nc.nspname and relationships.target_table_name = c.relname)
@@ -87,7 +128,7 @@ WHERE
       'SELECT, INSERT, UPDATE, DELETE, TRUNCATE, REFERENCES, TRIGGER'
     )
     OR has_any_column_privilege(c.oid, 'SELECT, INSERT, UPDATE, REFERENCES')
-  )
+  )${mainScope}
 group by
   c.oid,
   c.relname,
@@ -97,3 +138,10 @@ group by
   nc.nspname,
   pk.primary_keys
 `
+}
+
+// FROZEN legacy path: the unscoped rendering served while the
+// pgMetaScopedIntrospection flag is off. Do not edit its shape -- it must keep
+// matching production behavior until the flag cleanup deletes it. The scoped
+// form is getTablesSql(targetOid) (used by tables.retrieve).
+export const TABLES_SQL = getTablesSql()
diff --git a/packages/pg-meta/src/sql/types.ts b/packages/pg-meta/src/sql/types.ts
index b3da5666d24..d33c31c79ba 100644
--- a/packages/pg-meta/src/sql/types.ts
+++ b/packages/pg-meta/src/sql/types.ts
@@ -1,6 +1,18 @@
 import { safeSql } from '../pg-format'
 
+/**
+ * User-defined types introspection. TYPES_SQL (legacy, FROZEN below) left-joins
+ * two catalog-wide GROUP BY aggregates (slow on a large catalog). SCOPED_TYPES_SQL
+ * (opt-in) returns identical rows but computes enums/attrs per surviving row via
+ * correlated index-scan subqueries after the schema/array filters. Both end at
+ * the same trailing WHERE so pg-meta-types.ts#list appends the same filter/limit
+ * fragments (and, scoped-only, ORDER BY t.oid) to either. Equivalence is proven
+ * by execution tests in test/types.test.ts, not a byte snapshot.
+ */
 export const TYPES_SQL = /* SQL */ safeSql`
+-- FROZEN legacy path: served while the pgMetaScopedIntrospection flag is off.
+-- Do not edit -- it must keep matching production behavior until the flag
+-- cleanup deletes it. SCOPED_TYPES_SQL is the replacement.
 select
   t.oid::int8 as id,
   t.typname as name,
@@ -49,3 +61,52 @@ where
     )
   )
 `
+
+export const SCOPED_TYPES_SQL = /* SQL */ safeSql`
+select
+  t.oid::int8 as id,
+  t.typname as name,
+  n.nspname as schema,
+  format_type (t.oid, null) as format,
+  coalesce(
+    (
+      select
+        jsonb_agg(e.enumlabel order by e.enumsortorder)
+      from
+        pg_enum e
+      where
+        e.enumtypid = t.oid
+    ),
+    '[]'
+  ) as enums,
+  coalesce(
+    (
+      select
+        jsonb_agg(
+          jsonb_build_object('name', a.attname, 'type_id', a.atttypid::int8)
+          order by a.attnum asc
+        )
+      from
+        pg_attribute a
+      where
+        a.attrelid = t.typrelid and not a.attisdropped
+    ),
+    '[]'
+  ) as attributes,
+  obj_description (t.oid, 'pg_type') as comment
+from
+  pg_type t
+  left join pg_namespace n on n.oid = t.typnamespace
+where
+  (
+    t.typrelid = 0
+    or (
+      select
+        c.relkind = 'c'
+      from
+        pg_class c
+      where
+        c.oid = t.typrelid
+    )
+  )
+`
diff --git a/packages/pg-meta/test/db/stress-catalog.ts b/packages/pg-meta/test/db/stress-catalog.ts
index cd622c13c74..7da50902084 100644
--- a/packages/pg-meta/test/db/stress-catalog.ts
+++ b/packages/pg-meta/test/db/stress-catalog.ts
@@ -76,5 +76,29 @@ export async function buildStressCatalog(
     create table stress.p_root_east partition of stress.p_root for values in ('east');
   `)
 
+  // Enums and composite types so the user-defined-types introspection query has
+  // non-trivial results at scale: multi-label enums with a deliberately
+  // NON-alphabetical label order (to exercise the enumsortorder ordering), plus
+  // composite types -- one of which has a dropped attribute (to exercise the
+  // `not attisdropped` filter). A bulk loop grows pg_type/pg_attribute/pg_enum
+  // so the scoped correlated subqueries are exercised against a large catalog.
+  await db.executeQuery(`
+    create type stress.mood as enum ('ecstatic', 'sad', 'happy', 'ok');
+    create type stress.priority as enum ('critical', 'low', 'high', 'medium');
+    create type stress.addr as (street text, city text, zip int);
+    create type stress.pair as (a int, dropme text, b text);
+    alter type stress.pair drop attribute dropme;
+
+    create procedure stress.build_types(n int) language plpgsql as $$
+    begin
+      for i in 0..n-1 loop
+        execute format('create type stress.e_%s as enum (''c'', ''a'', ''b'')', i);
+        execute format('create type stress.ct_%s as (x int, y text)', i);
+        if i % 100 = 99 then commit; end if;
+      end loop;
+    end $$;
+  `)
+  await db.executeQuery(`call stress.build_types(200);`)
+
   await db.executeQuery(`analyze;`)
 }
diff --git a/packages/pg-meta/test/sql/studio/catalog-plan-guard.test.ts b/packages/pg-meta/test/sql/studio/catalog-plan-guard.test.ts
index 86dcb468539..dab67cf19fa 100644
--- a/packages/pg-meta/test/sql/studio/catalog-plan-guard.test.ts
+++ b/packages/pg-meta/test/sql/studio/catalog-plan-guard.test.ts
@@ -13,6 +13,9 @@ import {
   getTablesPaginatedSql,
   getViewDefinitionSql,
 } from '../../../src'
+import tablePrivileges from '../../../src/pg-meta-table-privileges'
+import * as tables from '../../../src/pg-meta-tables'
+import * as types from '../../../src/pg-meta-types'
 import { assertPlanWithinBudget, explainAnalyze } from '../../db/plan-guard'
 import { buildStressCatalog, STRESS_TABLE_COUNT } from '../../db/stress-catalog'
 import { cleanupRoot, createTestDatabase } from '../../db/utils'
@@ -150,7 +153,15 @@ test('getEntityTypesSQL: plan stays scoped for a schema listing', async () => {
       page: 0,
     })
   )
-  assertPlanWithinBudget(result, {})
+  assertPlanWithinBudget(result, {
+    allowedSeqScans: {
+      pg_class: {
+        max: 1,
+        reason:
+          'per-schema entity listing must read every relation in the schema; no index leads on pg_class.relnamespace, so the filter cannot prune it. The planner choice between a seq scan and a full-index bitmap scan is marginal and version/stats dependent (observed flipping on PG17) -- one filtered pass of pg_class either way',
+      },
+    },
+  })
 }, 60_000)
 
 // ── getTablesPaginatedSql (database/tables-paginated.ts) — per-schema page ───
@@ -287,3 +298,118 @@ test('getViewDefinitionSql: plan stays scoped for a single view', async () => {
   const result = await explainAnalyze(db, getViewDefinitionSql({ id: viewId }))
   assertPlanWithinBudget(result, {})
 }, 60_000)
+
+// ── types.list (sql/types.ts) — per-schema user-defined types listing ────────
+// The scoped rewrite filters pg_type by schema FIRST and then computes enums /
+// composite attributes per surviving row via correlated subqueries (index scans
+// on pg_enum's (enumtypid) and pg_attribute's (attrelid, attnum)), instead of
+// the legacy catalog-wide GROUP BY aggregates. The one structurally unavoidable
+// pass is over pg_type itself: a per-schema type listing must read every type in
+// the schema, and pg_type has no index leading on typnamespace (only
+// pg_type_typname_nsp_index on (typname, typnamespace)), so the schema filter
+// cannot prune the scan -- one filtered pass of pg_type either way. Same class
+// as the pg_constraint.confrelid / pg_class.relnamespace justifications.
+const TYPES_LIST_BUDGET = {
+  allowedSeqScans: {
+    pg_type: {
+      max: 1,
+      reason:
+        'per-schema type listing must read every type in the schema; no index leads on pg_type.typnamespace (only (typname, typnamespace)), so the filter cannot prune it. The planner picks a seq scan (PG17) or a full-index bitmap scan (PG14/15) depending on version/stats -- one filtered pass of pg_type either way',
+    },
+  },
+}
+test('types.list: scoped plan stays scoped for a schema', async () => {
+  const result = await explainAnalyze(
+    db,
+    types.list({ includedSchemas: ['stress'], scoped: true }).sql
+  )
+  assertPlanWithinBudget(result, TYPES_LIST_BUDGET)
+}, 60_000)
+
+test('types.list: scoped real query returns the schema’s enums and composites', async () => {
+  const { sql, zod } = types.list({ includedSchemas: ['stress'], scoped: true })
+  const rows = zod.parse(await db.executeQuery(sql))
+  // stress has enums (mood/priority/e_*) and composites (addr/pair/ct_*).
+  const mood = rows.find((t) => t.name === 'mood')
+  expect(mood?.enums).toEqual(['ecstatic', 'sad', 'happy', 'ok'])
+  // `pair` dropped its middle attribute; the scoped path must skip it.
+  expect(rows.find((t) => t.name === 'pair')?.attributes.map((a) => a.name)).toEqual(['a', 'b'])
+  expect(rows.length).toBeGreaterThan(400)
+}, 60_000)
+
+// The grantee/grantor resolution joins pg_roles (a view over pg_authid) TWICE —
+// once for the grantor, once for the grantee UNION that appends PUBLIC — so the
+// plan seq-scans pg_authid twice. pg_authid holds a fixed handful of roles and
+// does NOT scale with schema size (tables/columns/constraints), so this is
+// structural and unrelated to the O(catalog) regression this harness guards.
+const TABLE_PRIVILEGES_BUDGET = {
+  allowedSeqScans: {
+    pg_authid: {
+      max: 2,
+      reason:
+        'grantee/grantor resolution joins pg_roles (view over pg_authid) twice (grantor + grantee-with-PUBLIC union); pg_authid scales with role count, not schema size',
+    },
+  },
+}
+
+// ── tablePrivileges.list (sql/table-privileges.ts) — per-schema listing ──────
+// Scoped pushes the schema filter into the base WHERE (before the aclexplode
+// lateral / GROUP BY) so pg_class is pruned before privileges are exploded.
+test('tablePrivileges.list: scoped plan stays scoped for a schema', async () => {
+  const result = await explainAnalyze(
+    db,
+    tablePrivileges.list({ includedSchemas: ['stress'], scoped: true }).sql
+  )
+  assertPlanWithinBudget(result, TABLE_PRIVILEGES_BUDGET)
+}, 60_000)
+
+// ── tablePrivileges.retrieve (sql/table-privileges.ts) — per-relation ────────
+// Scoped pushes the schema+name (or oid) predicate into the base WHERE so the
+// relation is resolved via pg_class's (relname, relnamespace) index.
+test('tablePrivileges.retrieve: scoped plan stays scoped for a single relation', async () => {
+  const result = await explainAnalyze(
+    db,
+    tablePrivileges.retrieve({ name: 't_1000', schema: 'stress', scoped: true }).sql
+  )
+  assertPlanWithinBudget(result, TABLE_PRIVILEGES_BUDGET)
+}, 60_000)
+
+// ── tables.retrieve (pg-meta-tables.ts / sql/tables.ts) — single table ───────
+// Scoped pushes the target OID (a literal, or an initplan scalar subquery
+// resolved via pg_class's (relname, relnamespace) index) into the base scan and
+// every enrichment subquery, so pg_class/pg_attribute/pg_type/pg_index are all
+// index-driven. Two scaling-catalog seq scans remain and are structural:
+//   - pg_constraint: the relationships subquery keeps BOTH FK directions and
+//     pg_constraint.confrelid has no index, so the incoming-FK half is a seq
+//     scan (identical to getTableEditorSql's relationships CTE);
+//   - pg_attrdef: the columns enrichment resolves column defaults; pg_attrdef is
+//     scanned once and hash-joined against the target's (few) attributes.
+const TABLES_RETRIEVE_BUDGET = {
+  allowedSeqScans: {
+    pg_constraint: {
+      max: 2,
+      reason:
+        'relationships subquery keeps both FK directions; no index on pg_constraint.confrelid, so the incoming-FK half is a seq scan (same as getTableEditorSql)',
+    },
+    pg_attrdef: {
+      max: 1,
+      reason: 'columns enrichment resolves defaults; pg_attrdef scanned once and hash-joined',
+    },
+  },
+}
+
+test('tables.retrieve: scoped plan stays scoped for a single table by id', async () => {
+  const result = await explainAnalyze(
+    db,
+    tables.retrieve({ id: midChainTableId, scoped: true }).sql
+  )
+  assertPlanWithinBudget(result, TABLES_RETRIEVE_BUDGET)
+}, 60_000)
+
+test('tables.retrieve: scoped plan stays scoped for a single table by name+schema', async () => {
+  const result = await explainAnalyze(
+    db,
+    tables.retrieve({ name: 't_1000', schema: 'stress', scoped: true }).sql
+  )
+  assertPlanWithinBudget(result, TABLES_RETRIEVE_BUDGET)
+}, 60_000)
diff --git a/packages/pg-meta/test/sql/studio/rows-count.test.ts b/packages/pg-meta/test/sql/studio/rows-count.test.ts
new file mode 100644
index 00000000000..41d3bfcc975
--- /dev/null
+++ b/packages/pg-meta/test/sql/studio/rows-count.test.ts
@@ -0,0 +1,369 @@
+import { afterAll, expect, test } from 'vitest'
+
+import { getTableRowsCountSql } from '../../../src'
+import type { Filter } from '../../../src/query'
+import { cleanupRoot, createTestDatabase } from '../../db/utils'
+
+type Db = Awaited<ReturnType<typeof createTestDatabase>>
+type CountRow = { count: number; is_estimate: boolean }
+type CountArgs = Parameters<typeof getTableRowsCountSql>[0]
+
+afterAll(async () => {
+  await cleanupRoot()
+})
+
+const withTestDatabase = (name: string, fn: (db: Db) => Promise<void>) => {
+  test(name, async () => {
+    const db = await createTestDatabase()
+    try {
+      await fn(db)
+    } finally {
+      await db.cleanup()
+    }
+  })
+}
+
+const tableOf = async (db: Db, qualified: string, name: string, schema: string) => {
+  const [{ id }] = await db.executeQuery<{ id: number }[]>(
+    `select '${qualified}'::regclass::oid::int8 as id;`
+  )
+  return { id: Number(id), name, schema }
+}
+
+const reltuplesOf = async (db: Db, qualified: string) => {
+  const [{ reltuples }] = await db.executeQuery<{ reltuples: number }[]>(
+    `select reltuples::int8 as reltuples from pg_class where oid = '${qualified}'::regclass;`
+  )
+  return Number(reltuples)
+}
+
+const runCount = async (db: Db, args: CountArgs) => {
+  const [row] = await db.executeQuery<CountRow[]>(getTableRowsCountSql(args))
+  return { count: Number(row.count), is_estimate: row.is_estimate }
+}
+
+// Execute BOTH the scoped and legacy renderings of the same args against the DB
+// and assert identical results -- the equivalence contract for every case where
+// the two paths must agree (the ONLY intentional divergence is a never-analyzed
+// table whose heap exceeds the byte gate; that asymmetry is the fix and is
+// asserted separately below).
+const assertScopedEqualsLegacy = async (db: Db, base: Omit<CountArgs, 'scoped'>) => {
+  const legacy = await runCount(db, { ...base, scoped: false })
+  const scoped = await runCount(db, { ...base, scoped: true })
+  expect(scoped, 'scoped must equal legacy for this case').toEqual(legacy)
+  return scoped
+}
+
+// A never-analyzed table has pg_class.reltuples = -1 -- true for a brand-new
+// EMPTY table, a small one, AND a freshly bulk-loaded huge one. autovacuum is
+// disabled on every fixture so reltuples cannot flip mid-test.
+
+withTestDatabase(
+  'scoped: empty never-analyzed table -> exact count 0, is_estimate=false (both modes)',
+  async (db) => {
+    await db.executeQuery(
+      `create table public.empty_t (id int primary key) with (autovacuum_enabled = false);`
+    )
+    expect(await reltuplesOf(db, 'public.empty_t')).toBe(-1)
+    const table = await tableOf(db, 'public.empty_t', 'empty_t', 'public')
+
+    // Postgres estimates a never-vacuumed heap at a ~10-page minimum, so a naive
+    // reltuples=-1 -> estimate would report phantom rows here. The size gate
+    // routes an empty (0-byte) heap to an exact count instead.
+    expect(await runCount(db, { table, scoped: true })).toEqual({ count: 0, is_estimate: false })
+    expect(await runCount(db, { table, scoped: true, isReadOnlyContext: true })).toEqual({
+      count: 0,
+      is_estimate: false,
+    })
+  }
+)
+
+withTestDatabase(
+  'scoped: small never-analyzed table -> exact count, is_estimate=false (both modes)',
+  async (db) => {
+    await db.executeQuery(`
+      create table public.small_unanalyzed (id int primary key, val text)
+        with (autovacuum_enabled = false);
+      insert into public.small_unanalyzed select g, 'r' || g from generate_series(1, 1000) g;
+    `)
+    expect(await reltuplesOf(db, 'public.small_unanalyzed')).toBe(-1)
+    const table = await tableOf(db, 'public.small_unanalyzed', 'small_unanalyzed', 'public')
+
+    // Heap is a few tens of KB -- well under the byte gate -> exact count.
+    expect(await runCount(db, { table, scoped: true })).toEqual({ count: 1000, is_estimate: false })
+    expect(await runCount(db, { table, scoped: true, isReadOnlyContext: true })).toEqual({
+      count: 1000,
+      is_estimate: false,
+    })
+  }
+)
+
+withTestDatabase(
+  'scoped INTENTIONALLY diverges from legacy: large never-analyzed table -> estimate',
+  async (db) => {
+    // Wide rows (~300-byte payload) so the heap clears the ~10MB byte gate with a
+    // modest, fast-to-insert row count (~19MB at 60k rows) -- the case the legacy
+    // path mishandles (treats reltuples=-1 as small, runs a timing-out count).
+    await db.executeQuery(`
+      create table public.bulk_unanalyzed (id int primary key, val text)
+        with (autovacuum_enabled = false);
+      insert into public.bulk_unanalyzed
+        select g, repeat('x', 300) from generate_series(1, 60000) g;
+    `)
+    expect(await reltuplesOf(db, 'public.bulk_unanalyzed')).toBe(-1)
+    const [{ bytes }] = await db.executeQuery<{ bytes: number }[]>(
+      `select pg_relation_size('public.bulk_unanalyzed'::regclass)::int8 as bytes;`
+    )
+    expect(Number(bytes)).toBeGreaterThan(10_000_000)
+    const table = await tableOf(db, 'public.bulk_unanalyzed', 'bulk_unanalyzed', 'public')
+
+    // Non-readonly scoped: EXPLAIN-based estimate (works without ANALYZE).
+    const scoped = await runCount(db, { table, scoped: true })
+    expect(scoped.is_estimate).toBe(true)
+    expect(scoped.count).toBeGreaterThan(1000)
+    expect(scoped.count).not.toBe(-1)
+
+    // Readonly scoped: cannot create the estimate function -> reports -1 as an
+    // estimate rather than a timing-out exact count.
+    expect(await runCount(db, { table, scoped: true, isReadOnlyContext: true })).toEqual({
+      count: -1,
+      is_estimate: true,
+    })
+
+    // The intentional divergence: legacy (scoped:false) still runs an exact count
+    // on the -1 table (the pre-fix behavior the scoped path corrects).
+    const legacy = await runCount(db, { table })
+    expect(legacy).toEqual({ count: 60000, is_estimate: false })
+    expect(legacy.is_estimate).not.toBe(scoped.is_estimate)
+  }
+)
+
+withTestDatabase(
+  'scoped == legacy for an analyzed table below THRESHOLD_COUNT (default, filtered, enforceExactCount)',
+  async (db) => {
+    await db.executeQuery(`
+      create table public.analyzed_small (id int primary key, status text);
+      insert into public.analyzed_small
+        select g, case when g % 2 = 0 then 'active' else 'inactive' end
+        from generate_series(1, 10) g;
+      analyze public.analyzed_small;
+    `)
+    const table = await tableOf(db, 'public.analyzed_small', 'analyzed_small', 'public')
+    const activeFilter: Filter[] = [{ column: 'status', operator: '=', value: 'active' }]
+
+    // Default count: both paths exact-count a small analyzed table.
+    expect(await assertScopedEqualsLegacy(db, { table })).toEqual({ count: 10, is_estimate: false })
+    // Read-only default count agrees too.
+    expect(await assertScopedEqualsLegacy(db, { table, isReadOnlyContext: true })).toEqual({
+      count: 10,
+      is_estimate: false,
+    })
+    // Filtered count agrees.
+    expect(await assertScopedEqualsLegacy(db, { table, filters: activeFilter })).toEqual({
+      count: 5,
+      is_estimate: false,
+    })
+    // enforceExactCount ignores scoped entirely and agrees, with/without filters.
+    expect(await assertScopedEqualsLegacy(db, { table, enforceExactCount: true })).toEqual({
+      count: 10,
+      is_estimate: false,
+    })
+    expect(
+      await assertScopedEqualsLegacy(db, {
+        table,
+        enforceExactCount: true,
+        filters: activeFilter,
+      })
+    ).toEqual({ count: 5, is_estimate: false })
+  }
+)
+
+withTestDatabase(
+  'scoped == legacy for an analyzed table over THRESHOLD_COUNT (estimate path unchanged)',
+  async (db) => {
+    // reltuples > 50000 after analyze routes BOTH paths to the estimate branch
+    // (raw reltuples when unfiltered) -- identical output; the byte gate only
+    // affects the reltuples = -1 case, not this one.
+    await db.executeQuery(`
+      create table public.big_analyzed (id int primary key)
+        with (autovacuum_enabled = false);
+      insert into public.big_analyzed select generate_series(1, 60000);
+      analyze public.big_analyzed;
+    `)
+    expect(await reltuplesOf(db, 'public.big_analyzed')).toBeGreaterThan(50000)
+    const table = await tableOf(db, 'public.big_analyzed', 'big_analyzed', 'public')
+
+    // Non-readonly: both return the raw reltuples estimate.
+    const scoped = await assertScopedEqualsLegacy(db, { table })
+    expect(scoped.is_estimate).toBe(true)
+    expect(scoped.count).toBeGreaterThan(1000)
+
+    // Read-only: both report -1 as an estimate.
+    expect(await assertScopedEqualsLegacy(db, { table, isReadOnlyContext: true })).toEqual({
+      count: -1,
+      is_estimate: true,
+    })
+
+    // enforceExactCount over the threshold still runs a real count in both paths.
+    expect(await assertScopedEqualsLegacy(db, { table, enforceExactCount: true })).toEqual({
+      count: 60000,
+      is_estimate: false,
+    })
+  }
+)
+
+// A partitioned PARENT (relkind 'p') has no storage of its own, so
+// pg_relation_size(parent) is 0. The size gate must use the whole partition tree
+// or a large never-analyzed partitioned table would be misclassified as small
+// and exact-counted across all partitions -- the exact timeout being fixed.
+withTestDatabase(
+  'scoped: large never-analyzed PARTITIONED table -> estimate (partition-tree size gate)',
+  async (db) => {
+    await db.executeQuery(`
+      create table public.part_big (id int, region text, val text) partition by list (region);
+      create table public.part_big_e partition of public.part_big for values in ('east')
+        with (autovacuum_enabled = false);
+      create table public.part_big_w partition of public.part_big for values in ('west')
+        with (autovacuum_enabled = false);
+      insert into public.part_big
+        select g, case when g % 2 = 0 then 'east' else 'west' end, repeat('x', 300)
+        from generate_series(1, 45000) g;
+    `)
+    // Parent is never-analyzed (reltuples = -1) and has zero own heap size...
+    expect(await reltuplesOf(db, 'public.part_big')).toBe(-1)
+    const [{ own, tree }] = await db.executeQuery<{ own: number; tree: number }[]>(`
+      select
+        pg_relation_size('public.part_big'::regclass)::int8 as own,
+        (select coalesce(sum(pg_relation_size(relid)), 0)
+         from pg_partition_tree('public.part_big'::regclass))::int8 as tree;
+    `)
+    expect(Number(own)).toBe(0) // ...so pg_relation_size alone would say "small"
+    expect(Number(tree)).toBeGreaterThan(10_000_000) // the tree sum clears the gate
+    const table = await tableOf(db, 'public.part_big', 'part_big', 'public')
+
+    const scoped = await runCount(db, { table, scoped: true })
+    expect(scoped.is_estimate).toBe(true)
+    expect(scoped.count).toBeGreaterThan(1000)
+    expect(scoped.count).not.toBe(-1)
+
+    expect(await runCount(db, { table, scoped: true, isReadOnlyContext: true })).toEqual({
+      count: -1,
+      is_estimate: true,
+    })
+
+    // Legacy still exact-counts across all partitions (the pre-fix behavior).
+    expect(await runCount(db, { table })).toEqual({ count: 45000, is_estimate: false })
+  }
+)
+
+withTestDatabase(
+  'scoped: small never-analyzed PARTITIONED table -> exact count (both modes)',
+  async (db) => {
+    await db.executeQuery(`
+      create table public.part_small (id int, region text) partition by list (region);
+      create table public.part_small_e partition of public.part_small for values in ('east')
+        with (autovacuum_enabled = false);
+      create table public.part_small_w partition of public.part_small for values in ('west')
+        with (autovacuum_enabled = false);
+      insert into public.part_small
+        select g, case when g % 2 = 0 then 'east' else 'west' end from generate_series(1, 100) g;
+    `)
+    expect(await reltuplesOf(db, 'public.part_small')).toBe(-1)
+    const table = await tableOf(db, 'public.part_small', 'part_small', 'public')
+
+    expect(await runCount(db, { table, scoped: true })).toEqual({ count: 100, is_estimate: false })
+    expect(await runCount(db, { table, scoped: true, isReadOnlyContext: true })).toEqual({
+      count: 100,
+      is_estimate: false,
+    })
+  }
+)
+
+withTestDatabase(
+  'scoped == legacy for an ANALYZED partitioned table below THRESHOLD_COUNT',
+  async (db) => {
+    await db.executeQuery(`
+      create table public.part_analyzed (id int, region text) partition by list (region);
+      create table public.part_analyzed_e partition of public.part_analyzed for values in ('east');
+      create table public.part_analyzed_w partition of public.part_analyzed for values in ('west');
+      insert into public.part_analyzed
+        select g, case when g % 2 = 0 then 'east' else 'west' end from generate_series(1, 100) g;
+      analyze public.part_analyzed;
+    `)
+    const table = await tableOf(db, 'public.part_analyzed', 'part_analyzed', 'public')
+    expect(await assertScopedEqualsLegacy(db, { table })).toEqual({
+      count: 100,
+      is_estimate: false,
+    })
+    expect(await assertScopedEqualsLegacy(db, { table, isReadOnlyContext: true })).toEqual({
+      count: 100,
+      is_estimate: false,
+    })
+  }
+)
+
+withTestDatabase(
+  'scoped == legacy for a view flowing through the row-count builder',
+  async (db) => {
+    await db.executeQuery(`
+    create table public.view_src (id int primary key);
+    insert into public.view_src select generate_series(1, 7);
+    create view public.v_rows as select * from public.view_src;
+  `)
+    const table = await tableOf(db, 'public.v_rows', 'v_rows', 'public')
+
+    // A view has no heap (pg_relation_size 0, no partition tree) -> the gate keeps
+    // an exact count; scoped and legacy agree.
+    expect(await assertScopedEqualsLegacy(db, { table })).toEqual({ count: 7, is_estimate: false })
+    expect(await assertScopedEqualsLegacy(db, { table, isReadOnlyContext: true })).toEqual({
+      count: 7,
+      is_estimate: false,
+    })
+  }
+)
+
+// ── Fix #1: the embedded estimate select is quoted with literal(), so backslash
+// identifiers survive regardless of the session's standard_conforming_strings.
+withTestDatabase(
+  'scoped estimate path quotes the embedded select safely (backslash names, scs on & off)',
+  async (db) => {
+    // Names contain a backslash; in the JS template `\\` is one literal backslash.
+    await db.executeQuery(`
+      create table public."wei\\rd" ("col\\umn" int) with (autovacuum_enabled = false);
+      insert into public."wei\\rd" select g % 3 from generate_series(1, 60000) g;
+      analyze public."wei\\rd";
+    `)
+    const [{ id }] = await db.executeQuery<{ id: number }[]>(
+      `select 'public."wei\\rd"'::regclass::oid::int8 as id;`
+    )
+    // reltuples > THRESHOLD_COUNT + a filter -> the estimate (count_estimate)
+    // branch runs, embedding the filtered select (with the backslash names) as a
+    // literal inside the function call.
+    const table = { id: Number(id), name: 'wei\\rd', schema: 'public' }
+    const filters: Filter[] = [{ column: 'col\\umn', operator: '=', value: 1 }]
+
+    // Default standard_conforming_strings (on): both paths take the estimate
+    // branch and agree on the value.
+    const scopedOn = await runCount(db, { table, scoped: true, filters })
+    expect(scopedOn.is_estimate).toBe(true)
+    expect(Number.isFinite(scopedOn.count)).toBe(true)
+    expect(await assertScopedEqualsLegacy(db, { table, filters })).toEqual(scopedOn)
+
+    // standard_conforming_strings = off in the SAME connection: the SET, the
+    // CREATE FUNCTION, and the count must share one query (the test uses a pool).
+    const withScsOff = (sql: string) => `set standard_conforming_strings = off;\n${sql}`
+    const [scopedOff] = await db.executeQuery<CountRow[]>(
+      withScsOff(getTableRowsCountSql({ table, scoped: true, filters }))
+    )
+    // literal()/E'...' keeps the backslash identifiers intact under scs=off.
+    expect(scopedOff.is_estimate).toBe(true)
+    expect(Number.isFinite(Number(scopedOff.count))).toBe(true)
+
+    // The legacy apostrophe-only escaping mangles the backslashes under scs=off
+    // (the bug the scoped path fixes), so legacy errors there -- assert only the
+    // scoped behavior, per contract.
+    await expect(
+      db.executeQuery(withScsOff(getTableRowsCountSql({ table, filters })))
+    ).rejects.toThrow()
+  }
+)
diff --git a/packages/pg-meta/test/table-privileges.test.ts b/packages/pg-meta/test/table-privileges.test.ts
index d7ac82d0cd4..246448dad52 100644
--- a/packages/pg-meta/test/table-privileges.test.ts
+++ b/packages/pg-meta/test/table-privileges.test.ts
@@ -88,6 +88,71 @@ withTestDatabase('list table privileges', async ({ executeQuery }) => {
   )
 })
 
+type Priv = { grantor: string; grantee: string; privilege_type: string; is_grantable: boolean }
+type PrivRow = { relation_id: number; schema: string; name: string; privileges: Priv[] }
+
+withTestDatabase(
+  'scoped tablePrivileges.list/retrieve matches legacy (multiple grantees incl PUBLIC)',
+  async ({ executeQuery }) => {
+    // Multiple grantees including PUBLIC to exercise the PUBLIC (oid 0) branch of
+    // the grantee union and multi-grantee aggregation. Roles are cluster-global
+    // (not dropped by the per-test database cleanup), so guard the creation to
+    // stay idempotent when the test cluster is reused across runs.
+    await executeQuery(`
+      drop role if exists grantee_a;
+      drop role if exists grantee_b;
+      create role grantee_a;
+      create role grantee_b;
+      create table public.priv_demo (id int primary key, data text);
+      grant select, insert on public.priv_demo to grantee_a;
+      grant update (data) on public.priv_demo to grantee_b with grant option;
+      grant select on public.priv_demo to public;
+    `)
+
+    // list(): default and schema-filtered combos. RAW comparison -- no
+    // normalization, no sorting of rows or the privileges array.
+    for (const options of [{}, { includedSchemas: ['public'] }, { excludedSchemas: ['public'] }]) {
+      const legacy = await pgMeta.tablePrivileges.list(options)
+      const scoped = await pgMeta.tablePrivileges.list({ ...options, scoped: true })
+      const legacyRes = legacy.zod.parse(await executeQuery(legacy.sql)) as PrivRow[]
+      const scopedRes = scoped.zod.parse(await executeQuery(scoped.sql)) as PrivRow[]
+      expect(scopedRes, `list options: ${JSON.stringify(options)}`).toEqual(legacyRes)
+    }
+
+    // The scoped path must surface the PUBLIC grantee for priv_demo.
+    const { sql, zod } = await pgMeta.tablePrivileges.list({
+      includedSchemas: ['public'],
+      scoped: true,
+    })
+    const rows = zod.parse(await executeQuery(sql)) as PrivRow[]
+    const demo = rows.find((r) => r.name === 'priv_demo')!
+    expect(demo.privileges.some((p) => p.grantee === 'PUBLIC')).toBe(true)
+
+    // retrieve() by id and by schema+name.
+    const legacyById = await pgMeta.tablePrivileges.retrieve({ id: demo.relation_id })
+    const scopedById = await pgMeta.tablePrivileges.retrieve({
+      id: demo.relation_id,
+      scoped: true,
+    })
+    expect(scopedById.zod.parse((await executeQuery(scopedById.sql))[0])).toEqual(
+      legacyById.zod.parse((await executeQuery(legacyById.sql))[0])
+    )
+
+    const legacyByName = await pgMeta.tablePrivileges.retrieve({
+      name: 'priv_demo',
+      schema: 'public',
+    })
+    const scopedByName = await pgMeta.tablePrivileges.retrieve({
+      name: 'priv_demo',
+      schema: 'public',
+      scoped: true,
+    })
+    expect(scopedByName.zod.parse((await executeQuery(scopedByName.sql))[0])).toEqual(
+      legacyByName.zod.parse((await executeQuery(legacyByName.sql))[0])
+    )
+  }
+)
+
 withTestDatabase('revoke & grant table privileges', async ({ executeQuery }) => {
   // Get initial table privileges
   const { sql: listSql, zod: listZod } = await pgMeta.tablePrivileges.list()
diff --git a/packages/pg-meta/test/tables.test.ts b/packages/pg-meta/test/tables.test.ts
index c5eb3405e79..49144fa4823 100644
--- a/packages/pg-meta/test/tables.test.ts
+++ b/packages/pg-meta/test/tables.test.ts
@@ -35,6 +35,168 @@ const withTestDatabase = (name: string, fn: (db: TestDb) => Promise<void>) => {
   })
 }
 
+// Legacy relationships come out in plan-dependent order (frozen TABLES_SQL has
+// no ORDER BY; proven by the adversarial-FK test below), so canonicalize ONLY
+// the legacy side to the scoped ORDER BY: constraint_name, then the column names
+// (a composite FK expands to one entry per source×target column pair).
+const sortRels = (rels: any[]) =>
+  [...rels].sort(
+    (a, b) =>
+      a.constraint_name.localeCompare(b.constraint_name) ||
+      a.source_column_name.localeCompare(b.source_column_name) ||
+      a.target_column_name.localeCompare(b.target_column_name)
+  )
+
+withTestDatabase(
+  'scoped tables.retrieve matches legacy (FKs both directions, PK, comment, enums)',
+  async ({ executeQuery }) => {
+    await executeQuery(`
+    create type mood as enum ('sad', 'ok', 'happy');
+    create table public.parent (id int primary key, label text);
+    create table public.child (
+      id int primary key,
+      parent_id int references public.parent(id),
+      feeling mood,
+      self_ref int references public.child(id)
+    );
+    comment on table public.child is 'a child table';
+  `)
+
+    const [{ parent_id }] = await executeQuery<{ parent_id: number }[]>(
+      `select 'public.parent'::regclass::oid::int8 as parent_id;`
+    )
+    const [{ child_id }] = await executeQuery<{ child_id: number }[]>(
+      `select 'public.child'::regclass::oid::int8 as child_id;`
+    )
+
+    // parent: incoming FK (child.parent_id -> parent.id). child: outgoing FK to
+    // parent + a self-referential FK. Cover the id and name+schema branches.
+    const cases = [
+      {
+        label: 'parent by id',
+        legacy: { id: Number(parent_id) },
+        scoped: { id: Number(parent_id), scoped: true },
+      },
+      {
+        label: 'child by name+schema',
+        legacy: { name: 'child', schema: 'public' },
+        scoped: { name: 'child', schema: 'public', scoped: true },
+      },
+      {
+        label: 'child by id',
+        legacy: { id: Number(child_id) },
+        scoped: { id: Number(child_id), scoped: true },
+      },
+    ] as const
+
+    for (const c of cases) {
+      const legacy = pgMeta.tables.retrieve(c.legacy as any)
+      const scoped = pgMeta.tables.retrieve(c.scoped as any)
+      const legacyRow: any = legacy.zod.parse((await executeQuery(legacy.sql))[0])
+      const scopedRow: any = scoped.zod.parse((await executeQuery(scoped.sql))[0])
+
+      // Everything raw except the legacy relationships array, canonicalized to
+      // the scoped ORDER BY (legacy order is plan-dependent, see above).
+      legacyRow.relationships = sortRels(legacyRow.relationships)
+      expect(scopedRow, c.label).toEqual(legacyRow)
+      // Scoped's array is already in that order raw (no scoped-side sort).
+      expect(scopedRow.relationships, c.label).toEqual(sortRels(scopedRow.relationships))
+    }
+
+    // Sanity: the scoped parent retrieve surfaces the INCOMING FK from child.
+    const scopedParent = pgMeta.tables.retrieve({ id: Number(parent_id), scoped: true })
+    const parentRow: any = scopedParent.zod.parse((await executeQuery(scopedParent.sql))[0])
+    expect(
+      parentRow.relationships.some(
+        (r: any) => r.source_table_name === 'child' && r.target_table_name === 'parent'
+      )
+    ).toBe(true)
+  }
+)
+
+// Regression proof for the relationships exception: FK creation order (zzz, aaa,
+// mmm) differs from constraint_name order, so legacy's plan order cannot match
+// scoped's without the legacy-side sort, while scoped is deterministic by name.
+withTestDatabase(
+  'scoped tables.retrieve orders relationships by constraint_name (adversarial)',
+  async ({ executeQuery }) => {
+    await executeQuery(`
+      create table public.ref (id int primary key);
+      create table public.multi (id int primary key, a int, b int, c int);
+      alter table public.multi add constraint zzz_fk foreign key (a) references public.ref (id);
+      alter table public.multi add constraint aaa_fk foreign key (b) references public.ref (id);
+      alter table public.multi add constraint mmm_fk foreign key (c) references public.ref (id);
+    `)
+    const legacy = pgMeta.tables.retrieve({ name: 'multi', schema: 'public' })
+    const scoped = pgMeta.tables.retrieve({ name: 'multi', schema: 'public', scoped: true })
+    const legacyRow: any = legacy.zod.parse((await executeQuery(legacy.sql))[0])
+    const scopedRow: any = scoped.zod.parse((await executeQuery(scoped.sql))[0])
+
+    // Scoped is deterministically constraint_name-ordered, raw.
+    expect(scopedRow.relationships.map((r: any) => r.constraint_name)).toEqual([
+      'aaa_fk',
+      'mmm_fk',
+      'zzz_fk',
+    ])
+    // Equal only after canonicalizing the (plan-dependent) legacy side.
+    legacyRow.relationships = sortRels(legacyRow.relationships)
+    expect(scopedRow).toEqual(legacyRow)
+  }
+)
+
+// Composite (multi-column) FK: the relationships subquery expands it to one
+// entry per source×target column pair, all sharing constraint_name, so the
+// scoped ORDER BY tie-breaks on the column names to stay deterministic.
+withTestDatabase(
+  'scoped tables.retrieve orders composite-FK relationship entries deterministically',
+  async ({ executeQuery }) => {
+    await executeQuery(`
+      create table public.ctgt (x int, y int, primary key (x, y));
+      create table public.csrc (a int, b int, foreign key (a, b) references public.ctgt (x, y));
+    `)
+    const scoped = pgMeta.tables.retrieve({ name: 'csrc', schema: 'public', scoped: true })
+    const legacy = pgMeta.tables.retrieve({ name: 'csrc', schema: 'public' })
+    const scopedRow: any = scoped.zod.parse((await executeQuery(scoped.sql))[0])
+    const legacyRow: any = legacy.zod.parse((await executeQuery(legacy.sql))[0])
+
+    // Four entries (2 source cols × 2 target cols), all one constraint_name.
+    expect(scopedRow.relationships).toHaveLength(4)
+    expect(new Set(scopedRow.relationships.map((r: any) => r.constraint_name)).size).toBe(1)
+    // Scoped is already ordered by (name, source col, target col), raw.
+    expect(
+      scopedRow.relationships.map((r: any) => [r.source_column_name, r.target_column_name])
+    ).toEqual([
+      ['a', 'x'],
+      ['a', 'y'],
+      ['b', 'x'],
+      ['b', 'y'],
+    ])
+    legacyRow.relationships = sortRels(legacyRow.relationships)
+    expect(scopedRow).toEqual(legacyRow)
+  }
+)
+
+withTestDatabase(
+  'scoped tables.retrieve preserves composite primary-key column order (indkey, not sorted)',
+  async ({ executeQuery }) => {
+    // PK declared (b, a) -- index column order is the reverse of alphabetical, so
+    // an accidental name-sort would be caught here.
+    await executeQuery(
+      `create table public.composite_pk (a int not null, b int not null, c int, primary key (b, a));`
+    )
+    const scoped = pgMeta.tables.retrieve({ name: 'composite_pk', schema: 'public', scoped: true })
+    const scopedRow: any = scoped.zod.parse((await executeQuery(scoped.sql))[0])
+    // Emitted in index (indkey) order (b, a), the semantically meaningful order.
+    expect(scopedRow.primary_keys.map((pk: any) => pk.name)).toEqual(['b', 'a'])
+
+    // Legacy emits the same PK order (the PK aggregate is ordered identically in
+    // both renderings), so primary_keys match without any normalization.
+    const legacy = pgMeta.tables.retrieve({ name: 'composite_pk', schema: 'public' })
+    const legacyRow: any = legacy.zod.parse((await executeQuery(legacy.sql))[0])
+    expect(scopedRow.primary_keys).toEqual(legacyRow.primary_keys)
+  }
+)
+
 /** Original tests ported from postgres-meta */
 withTestDatabase('list tables', async ({ executeQuery }) => {
   const { sql, zod } = await pgMeta.tables.list()
diff --git a/packages/pg-meta/test/types.test.ts b/packages/pg-meta/test/types.test.ts
index 20ff701170e..ad548ec2d30 100644
--- a/packages/pg-meta/test/types.test.ts
+++ b/packages/pg-meta/test/types.test.ts
@@ -87,6 +87,64 @@ withTestDatabase(
   }
 )
 
+const OPTION_MATRIX = [
+  { label: 'default', options: {} },
+  { label: 'includeArrayTypes', options: { includeArrayTypes: true } },
+  { label: 'includedSchemas=public', options: { includedSchemas: ['public'] } },
+  { label: 'excludedSchemas=public', options: { excludedSchemas: ['public'] } },
+  {
+    label: 'excludedSchemas=public + includeSystemSchemas',
+    options: { excludedSchemas: ['public'], includeSystemSchemas: true },
+  },
+  {
+    label: 'includedSchemas=public + includeArrayTypes',
+    options: { includedSchemas: ['public'], includeArrayTypes: true },
+  },
+]
+
+withTestDatabase(
+  'scoped types.list matches legacy for all option combos (enums, composites, dropped attrs, array types)',
+  async ({ executeQuery }) => {
+    // Fixture coverage: multi-label enum with a NON-alphabetical label order (so
+    // enumsortorder matters), a composite type with a dropped attribute, and a
+    // composite referencing the enum.
+    await executeQuery(`
+      create type color as enum ('red', 'green', 'blue', 'yellow');
+      create type shipment as (id int8, note text, tossme int, color color);
+      alter type shipment drop attribute tossme;
+      create type point3 as (x float8, y float8, z float8);
+    `)
+
+    // Legacy types.list has no ORDER BY (plan-dependent row order), so sort ONLY
+    // the legacy side by id (t.oid) to match the scoped `order by t.oid`.
+    for (const { label, options } of OPTION_MATRIX) {
+      const legacy = await pgMeta.types.list(options)
+      const scoped = await pgMeta.types.list({ ...options, scoped: true })
+
+      const legacyRes = [...legacy.zod.parse(await executeQuery(legacy.sql))].sort(
+        (a, b) => a.id - b.id
+      )
+      const scopedRes = scoped.zod.parse(await executeQuery(scoped.sql))
+
+      expect(scopedRes, `option combo: ${label}`).toEqual(legacyRes)
+      // Scoped is already in t.oid order raw (no scoped-side sort).
+      const scopedIds = scopedRes.map((t) => t.id)
+      expect(scopedIds, `${label} oid-ordered`).toEqual([...scopedIds].sort((a, b) => a - b))
+    }
+
+    // Sanity: the enum/composite fixtures actually surface with correct
+    // ordering + dropped-attribute handling in the scoped path.
+    const { sql, zod } = await pgMeta.types.list({ includedSchemas: ['public'], scoped: true })
+    const res = zod.parse(await executeQuery(sql))
+    expect(res.find((t) => t.name === 'color')?.enums).toEqual(['red', 'green', 'blue', 'yellow'])
+    expect(res.find((t) => t.name === 'shipment')?.attributes.map((a) => a.name)).toEqual([
+      'id',
+      'note',
+      'color',
+    ])
+  }
+)
+
 withTestDatabase('composite type attributes', async ({ executeQuery }) => {
   await executeQuery(`create type test_composite as (id int8, data text);`)