diff --git a/apps/studio/components/interfaces/Database/Hooks/EditHookPanel.tsx b/apps/studio/components/interfaces/Database/Hooks/EditHookPanel.tsx
index 2d48173a94e..40db2124f0a 100644
--- a/apps/studio/components/interfaces/Database/Hooks/EditHookPanel.tsx
+++ b/apps/studio/components/interfaces/Database/Hooks/EditHookPanel.tsx
@@ -2,7 +2,7 @@ import { zodResolver } from '@hookform/resolvers/zod'
 import { keyword } from '@supabase/pg-meta'
 import type { PGTrigger, PGTriggerCreate } from '@supabase/pg-meta'
 import { useQueryClient } from '@tanstack/react-query'
-import { useParams } from 'common'
+import { useFlag, useParams } from 'common'
 import { parseAsBoolean, parseAsString, useQueryState } from 'nuqs'
 import { useEffect, useRef, useState } from 'react'
 import { SubmitHandler, useForm } from 'react-hook-form'
@@ -15,7 +15,10 @@ import { DiscardChangesConfirmationDialog } from '@/components/ui-patterns/Dialo
 import { useDatabaseTriggerCreateMutation } from '@/data/database-triggers/database-trigger-create-mutation'
 import { useDatabaseTriggerUpdateMutation } from '@/data/database-triggers/database-trigger-update-transaction-mutation'
 import { useDatabaseHooksQuery } from '@/data/database-triggers/database-triggers-query'
-import { tableEditorQueryOptions } from '@/data/table-editor/table-editor-query'
+import {
+  PG_META_SCOPED_INTROSPECTION_FLAG,
+  tableEditorQueryOptions,
+} from '@/data/table-editor/table-editor-query'
 import { useSelectedProjectQuery } from '@/hooks/misc/useSelectedProject'
 import { useConfirmOnClose } from '@/hooks/ui/useConfirmOnClose'
 import { uuidv4 } from '@/lib/helpers'
@@ -80,6 +83,7 @@ export const EditHookPanel = () => {
   const { ref } = useParams()
   const { data: project } = useSelectedProjectQuery()
   const [isLoadingTable, setIsLoadingTable] = useState(false)
+  const scoped = !!useFlag(PG_META_SCOPED_INTROSPECTION_FLAG)
 
   const { data: hooks = [], isSuccess } = useDatabaseHooksQuery({
     projectRef: project?.ref,
@@ -212,6 +216,7 @@ export const EditHookPanel = () => {
           id: Number(values.table_id),
           projectRef: project?.ref,
           connectionString: project?.connectionString,
+          scoped,
         })
       )
       if (!selectedTable) {
diff --git a/apps/studio/components/interfaces/TableGridEditor/SidePanelEditor/SidePanelEditor.tsx b/apps/studio/components/interfaces/TableGridEditor/SidePanelEditor/SidePanelEditor.tsx
index 05a0d70b445..11118f700cb 100644
--- a/apps/studio/components/interfaces/TableGridEditor/SidePanelEditor/SidePanelEditor.tsx
+++ b/apps/studio/components/interfaces/TableGridEditor/SidePanelEditor/SidePanelEditor.tsx
@@ -1,7 +1,7 @@
 import * as Sentry from '@sentry/nextjs'
 import type { PGTable } from '@supabase/pg-meta'
 import { useQueryClient } from '@tanstack/react-query'
-import { useParams } from 'common'
+import { useFlag, useParams } from 'common'
 import { isEmpty, isUndefined, noop } from 'lodash'
 import { useState } from 'react'
 import { toast } from 'sonner'
@@ -51,6 +51,7 @@ import { lintKeys } from '@/data/lint/keys'
 import { privilegeKeys } from '@/data/privileges/keys'
 import { useTableApiAccessPrivilegesMutation } from '@/data/privileges/table-api-access-mutation'
 import { tableEditorKeys } from '@/data/table-editor/keys'
+import { PG_META_SCOPED_INTROSPECTION_FLAG } from '@/data/table-editor/table-editor-query'
 import { isTableLike, type Entity } from '@/data/table-editor/table-editor-types'
 import { tableRowKeys } from '@/data/table-rows/keys'
 import { tableKeys } from '@/data/tables/keys'
@@ -193,6 +194,7 @@ export const SidePanelEditor = ({
   const { data: project } = useSelectedProjectQuery()
   const isQueueOperationsEnabled = useIsQueueOperationsEnabled()
   const { updateRow, addRow, isEditPending } = useTableRowOperations()
+  const scoped = !!useFlag(PG_META_SCOPED_INTROSPECTION_FLAG)
 
   const [isEdited, setIsEdited] = useState<boolean>(false)
   const csvImportKey = useVisibleKey(snap.sidePanel?.type === 'csv-import')
@@ -677,6 +679,7 @@ export const SidePanelEditor = ({
                 isRLSEnabled,
                 importContent,
                 track,
+                scoped,
               })
 
               createTableSpan.setAttribute('table.created', 1)
@@ -785,6 +788,7 @@ export const SidePanelEditor = ({
           existingForeignKeyRelations,
           primaryKey,
           track,
+          scoped,
         })
 
         if (table === undefined) {
diff --git a/apps/studio/components/interfaces/TableGridEditor/SidePanelEditor/SidePanelEditor.utils.tsx b/apps/studio/components/interfaces/TableGridEditor/SidePanelEditor/SidePanelEditor.utils.tsx
index 63b72069d9d..6583e11a2c7 100644
--- a/apps/studio/components/interfaces/TableGridEditor/SidePanelEditor/SidePanelEditor.utils.tsx
+++ b/apps/studio/components/interfaces/TableGridEditor/SidePanelEditor/SidePanelEditor.utils.tsx
@@ -442,6 +442,7 @@ export const createTable = async ({
   isRLSEnabled,
   importContent,
   track,
+  scoped,
 }: {
   projectRef: string
   connectionString?: string | null
@@ -456,6 +457,7 @@ export const createTable = async ({
   isRLSEnabled: boolean
   importContent?: ImportContent
   track: Track
+  scoped?: boolean
 }) => {
   const queryClient = getQueryClient()
 
@@ -674,6 +676,7 @@ export const createTable = async ({
         projectRef,
         connectionString,
         id: table.id,
+        scoped,
       })
     }
   )
@@ -694,6 +697,7 @@ export const updateTable = async ({
   existingForeignKeyRelations,
   primaryKey,
   track,
+  scoped,
 }: {
   projectRef: string
   connectionString?: string | null
@@ -705,6 +709,7 @@ export const updateTable = async ({
   existingForeignKeyRelations: ForeignKeyConstraint[]
   primaryKey?: Constraint
   track: Track
+  scoped?: boolean
 }) => {
   const queryClient = getQueryClient()
 
@@ -878,6 +883,7 @@ export const updateTable = async ({
       projectRef,
       connectionString,
       id: table.id,
+      scoped,
     }),
     hasError,
   }
diff --git a/apps/studio/data/database/table-definition-query.ts b/apps/studio/data/database/table-definition-query.ts
index 69df0767286..219fa1e2fb5 100644
--- a/apps/studio/data/database/table-definition-query.ts
+++ b/apps/studio/data/database/table-definition-query.ts
@@ -1,12 +1,15 @@
 import { getTableDefinitionSql } from '@supabase/pg-meta'
 import { useQuery } from '@tanstack/react-query'
+import { useFlag } from 'common'
 
 import { databaseKeys } from './keys'
 import { executeSql } from '@/data/sql/execute-sql-mutation'
+import { PG_META_SCOPED_INTROSPECTION_FLAG } from '@/data/table-editor/table-editor-query'
 import { ResponseError, UseCustomQueryOptions } from '@/types'
 
 type GetTableDefinitionArgs = {
   id?: number
+  scoped?: boolean
 }
 
 export type TableDefinitionVariables = GetTableDefinitionArgs & {
@@ -15,12 +18,12 @@ export type TableDefinitionVariables = GetTableDefinitionArgs & {
 }
 
 export async function getTableDefinition(
-  { projectRef, connectionString, id }: TableDefinitionVariables,
+  { projectRef, connectionString, id, scoped }: TableDefinitionVariables,
   signal?: AbortSignal
 ) {
   if (!id) throw new Error('id is required')
 
-  const sql = getTableDefinitionSql({ id })
+  const sql = getTableDefinitionSql({ id, scoped })
   const { result } = await executeSql(
     {
       projectRef,
@@ -43,11 +46,15 @@ export const useTableDefinitionQuery = <TData = TableDefinitionData>(
     enabled = true,
     ...options
   }: UseCustomQueryOptions<TableDefinitionData, TableDefinitionError, TData> = {}
-) =>
-  useQuery<TableDefinitionData, TableDefinitionError, TData>({
-    queryKey: databaseKeys.tableDefinition(projectRef, id),
-    queryFn: ({ signal }) => getTableDefinition({ projectRef, connectionString, id }, signal),
+) => {
+  const scoped = !!useFlag(PG_META_SCOPED_INTROSPECTION_FLAG)
+
+  return useQuery<TableDefinitionData, TableDefinitionError, TData>({
+    queryKey: [...databaseKeys.tableDefinition(projectRef, id), { scoped }],
+    queryFn: ({ signal }) =>
+      getTableDefinition({ projectRef, connectionString, id, scoped }, signal),
     enabled:
       enabled && typeof projectRef !== 'undefined' && typeof id !== 'undefined' && !isNaN(id),
     ...options,
   })
+}
diff --git a/apps/studio/data/prefetchers/project.$ref.editor.$id.tsx b/apps/studio/data/prefetchers/project.$ref.editor.$id.tsx
index 0aeb780eefb..fdde7f38002 100644
--- a/apps/studio/data/prefetchers/project.$ref.editor.$id.tsx
+++ b/apps/studio/data/prefetchers/project.$ref.editor.$id.tsx
@@ -1,4 +1,5 @@
 import { QueryClient, useQueryClient } from '@tanstack/react-query'
+import { useFlag } from 'common'
 import { useRouter } from 'next/router'
 import { PropsWithChildren, useCallback } from 'react'
 
@@ -10,7 +11,10 @@ import {
 } from '@/components/grid/SupabaseGrid.utils'
 import { Filter, Sort } from '@/components/grid/types'
 import { useConnectionStringForReadOps } from '@/data/read-replicas/replicas-query'
-import { prefetchTableEditor } from '@/data/table-editor/table-editor-query'
+import {
+  PG_META_SCOPED_INTROSPECTION_FLAG,
+  prefetchTableEditor,
+} from '@/data/table-editor/table-editor-query'
 import { prefetchTableRows } from '@/data/table-rows/table-rows-query'
 import { useSelectedProjectQuery } from '@/hooks/misc/useSelectedProject'
 import { RoleImpersonationState } from '@/lib/role-impersonation'
@@ -26,6 +30,7 @@ interface PrefetchEditorTablePageArgs {
   sorts?: Sort[]
   filters?: Filter[]
   roleImpersonationState?: RoleImpersonationState
+  scoped?: boolean
 }
 
 export function prefetchEditorTablePage({
@@ -37,11 +42,13 @@ export function prefetchEditorTablePage({
   sorts,
   filters,
   roleImpersonationState,
+  scoped,
 }: PrefetchEditorTablePageArgs) {
   return prefetchTableEditor(queryClient, {
     projectRef,
     connectionString,
     id,
+    scoped,
   }).then((entity) => {
     if (entity) {
       const { sorts: localSorts = [], filters: localFilters = [] } =
@@ -57,6 +64,7 @@ export function prefetchEditorTablePage({
         page: 1,
         limit: TABLE_EDITOR_DEFAULT_ROWS_PER_PAGE,
         roleImpersonationState,
+        scoped,
       })
     }
   })
@@ -68,6 +76,7 @@ export function usePrefetchEditorTablePage() {
   const { data: project } = useSelectedProjectQuery()
   const { connectionString, identifier: readReplicaIdentifier } = useConnectionStringForReadOps()
   const roleImpersonationState = useRoleImpersonationStateSnapshot()
+  const scoped = !!useFlag(PG_META_SCOPED_INTROSPECTION_FLAG)
 
   return useCallback(
     ({ id: _id, filters, sorts }: { id?: string; filters?: Filter[]; sorts?: Sort[] }) => {
@@ -87,11 +96,20 @@ export function usePrefetchEditorTablePage() {
         sorts,
         filters,
         roleImpersonationState: roleImpersonationState as RoleImpersonationState,
+        scoped,
       }).catch(() => {
         // eat prefetching errors as they are not critical
       })
     },
-    [connectionString, readReplicaIdentifier, project, queryClient, roleImpersonationState, router]
+    [
+      connectionString,
+      readReplicaIdentifier,
+      project,
+      queryClient,
+      roleImpersonationState,
+      router,
+      scoped,
+    ]
   )
 }
 
diff --git a/apps/studio/data/table-editor/table-editor-query.ts b/apps/studio/data/table-editor/table-editor-query.ts
index 6ccddc7012e..e964103c8f5 100644
--- a/apps/studio/data/table-editor/table-editor-query.ts
+++ b/apps/studio/data/table-editor/table-editor-query.ts
@@ -1,13 +1,17 @@
 import { getTableEditorSql } from '@supabase/pg-meta'
 import { QueryClient, queryOptions, useQuery } from '@tanstack/react-query'
+import { useFlag } from 'common'
 
 import { tableEditorKeys } from './keys'
 import { Entity } from './table-editor-types'
 import { executeSql } from '@/data/sql/execute-sql-mutation'
 import { ResponseError, UseCustomQueryOptions } from '@/types'
 
+export const PG_META_SCOPED_INTROSPECTION_FLAG = 'pgMetaScopedIntrospection'
+
 type TableEditorArgs = {
   id?: number
+  scoped?: boolean
 }
 
 export type TableEditorVariables = TableEditorArgs & {
@@ -16,14 +20,14 @@ export type TableEditorVariables = TableEditorArgs & {
 }
 
 export async function getTableEditor(
-  { projectRef, connectionString, id }: TableEditorVariables,
+  { projectRef, connectionString, id, scoped = false }: TableEditorVariables,
   signal?: AbortSignal
 ) {
   if (!id) {
     throw new Error('id is required')
   }
 
-  const sql = getTableEditorSql({ id })
+  const sql = getTableEditorSql({ id, scoped })
   const { result } = await executeSql(
     {
       projectRef,
@@ -46,9 +50,11 @@ export const useTableEditorQuery = <TData = TableEditorData>(
     enabled = true,
     ...options
   }: UseCustomQueryOptions<TableEditorData, TableEditorError, TData> = {}
-) =>
-  useQuery<TableEditorData, TableEditorError, TData>({
-    ...tableEditorQueryOptions({ projectRef, connectionString, id }),
+) => {
+  const scoped = !!useFlag(PG_META_SCOPED_INTROSPECTION_FLAG)
+
+  return useQuery<TableEditorData, TableEditorError, TData>({
+    ...tableEditorQueryOptions({ projectRef, connectionString, id, scoped }),
     enabled:
       enabled && typeof projectRef !== 'undefined' && typeof id !== 'undefined' && !isNaN(id),
     refetchOnWindowFocus: false,
@@ -56,21 +62,23 @@ export const useTableEditorQuery = <TData = TableEditorData>(
     staleTime: 5 * 60 * 1000,
     ...options,
   })
+}
 
 export function prefetchTableEditor(
   client: QueryClient,
-  { projectRef, connectionString, id }: TableEditorVariables
+  { projectRef, connectionString, id, scoped }: TableEditorVariables
 ) {
-  return client.fetchQuery(tableEditorQueryOptions({ projectRef, connectionString, id }))
+  return client.fetchQuery(tableEditorQueryOptions({ projectRef, connectionString, id, scoped }))
 }
 
 export const tableEditorQueryOptions = <TData = TableEditorData>({
   projectRef,
   connectionString,
   id,
+  scoped,
 }: TableEditorVariables) => {
   return queryOptions<TableEditorData, TableEditorError, TData>({
-    queryKey: tableEditorKeys.tableEditor(projectRef, id),
-    queryFn: ({ signal }) => getTableEditor({ projectRef, connectionString, id }, signal),
+    queryKey: [...tableEditorKeys.tableEditor(projectRef, id), { scoped: !!scoped }],
+    queryFn: ({ signal }) => getTableEditor({ projectRef, connectionString, id, scoped }, signal),
   })
 }
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 111be34e74c..636d4ef1406 100644
--- a/apps/studio/data/table-rows/table-rows-count-query.ts
+++ b/apps/studio/data/table-rows/table-rows-count-query.ts
@@ -1,7 +1,7 @@
 import { getTableRowsCountSql } from '@supabase/pg-meta'
 import { PermissionAction } from '@supabase/shared-types/out/constants'
 import { QueryClient, useQuery, useQueryClient } from '@tanstack/react-query'
-import { IS_PLATFORM } from 'common'
+import { IS_PLATFORM, useFlag } from 'common'
 
 import { tableRowKeys } from './keys'
 import { formatFilterValue } from './utils'
@@ -9,7 +9,10 @@ import { parseSupaTable } from '@/components/grid/SupabaseGrid.utils'
 import type { Filter, SupaTable } from '@/components/grid/types'
 import { useConnectionStringForReadOps } from '@/data/read-replicas/replicas-query'
 import { executeSql } from '@/data/sql/execute-sql-mutation'
-import { prefetchTableEditor } from '@/data/table-editor/table-editor-query'
+import {
+  PG_META_SCOPED_INTROSPECTION_FLAG,
+  prefetchTableEditor,
+} from '@/data/table-editor/table-editor-query'
 import { useAsyncCheckPermissions } from '@/hooks/misc/useCheckPermissions'
 import { RoleImpersonationState, wrapWithRoleImpersonation } from '@/lib/role-impersonation'
 import { isRoleImpersonationEnabled } from '@/state/role-impersonation-state'
@@ -32,6 +35,7 @@ export type TableRowsCountVariables = Omit<GetTableRowsCountArgs, 'table'> & {
   roleImpersonationState?: RoleImpersonationState
   projectRef?: string
   connectionString?: string | null
+  scoped?: boolean
 }
 
 export type TableRowsCountData = TableRowsCount
@@ -47,6 +51,7 @@ export async function getTableRowsCount(
     roleImpersonationState,
     enforceExactCount,
     isReadOnlyContext = false,
+    scoped,
   }: TableRowsCountVariables & { isReadOnlyContext?: boolean },
   signal?: AbortSignal
 ) {
@@ -54,6 +59,7 @@ export async function getTableRowsCount(
     projectRef,
     connectionString,
     id: tableId,
+    scoped,
   })
   if (!entity) {
     throw new Error('Table not found')
@@ -109,12 +115,14 @@ export const useTableRowsCountQuery = <TData = TableRowsCountData>(
     PermissionAction.TENANT_SQL_ADMIN_WRITE,
     'tables'
   )
+  const scoped = !!useFlag(PG_META_SCOPED_INTROSPECTION_FLAG)
 
   return useQuery<TableRowsCountData, TableRowsCountError, TData>({
     queryKey: tableRowKeys.tableRowsCount(projectRef, {
       table: { id: tableId },
       readReplicaIdentifier,
       ...args,
+      scoped,
     }),
     queryFn: ({ signal }) =>
       getTableRowsCount(
@@ -125,6 +133,7 @@ export const useTableRowsCountQuery = <TData = TableRowsCountData>(
           tableId,
           isReadOnlyContext: type === 'replica' || !canSQLAdminWrite,
           ...args,
+          scoped,
         },
         signal
       ),
diff --git a/apps/studio/data/table-rows/table-rows-query.ts b/apps/studio/data/table-rows/table-rows-query.ts
index 98b1e2bb560..f2a874d6aaa 100644
--- a/apps/studio/data/table-rows/table-rows-query.ts
+++ b/apps/studio/data/table-rows/table-rows-query.ts
@@ -2,7 +2,7 @@ import { ident, joinSqlFragments, ROLE_IMPERSONATION_NO_RESULTS, safeSql } from
 import { Query, type QueryFilter } from '@supabase/pg-meta/src/query'
 import { getTableRowsSql } from '@supabase/pg-meta/src/query/table-row-query'
 import { useQuery, useQueryClient, type QueryClient } from '@tanstack/react-query'
-import { IS_PLATFORM } from 'common'
+import { IS_PLATFORM, useFlag } from 'common'
 
 import { tableRowKeys } from './keys'
 import { formatFilterValue } from './utils'
@@ -12,7 +12,10 @@ import { ENTITY_TYPE } from '@/data/entity-types/entity-type-constants'
 import { handleError } from '@/data/fetchers'
 import { useConnectionStringForReadOps } from '@/data/read-replicas/replicas-query'
 import { executeSql } from '@/data/sql/execute-sql-mutation'
-import { prefetchTableEditor } from '@/data/table-editor/table-editor-query'
+import {
+  PG_META_SCOPED_INTROSPECTION_FLAG,
+  prefetchTableEditor,
+} from '@/data/table-editor/table-editor-query'
 import { isMsSqlForeignTable } from '@/data/table-editor/table-editor-types'
 import { timeout } from '@/lib/helpers'
 import { RoleImpersonationState, wrapWithRoleImpersonation } from '@/lib/role-impersonation'
@@ -26,6 +29,7 @@ interface GetTableRowsArgs {
   limit?: number
   page?: number
   roleImpersonationState?: RoleImpersonationState
+  scoped?: boolean
 }
 
 /**
@@ -330,6 +334,7 @@ async function getTableRows(
     limit,
     page,
     preflightCheck = false,
+    scoped,
   }: TableRowsVariables,
   signal?: AbortSignal
 ) {
@@ -337,6 +342,7 @@ async function getTableRows(
     projectRef,
     connectionString,
     id: tableId,
+    scoped,
   })
   if (!entity) {
     throw new Error('Table not found')
@@ -397,6 +403,7 @@ export const useTableRowsQuery = <TData = TableRowsData>(
 ) => {
   const queryClient = useQueryClient()
   const { connectionString, identifier: readReplicaIdentifier } = useConnectionStringForReadOps()
+  const scoped = !!useFlag(PG_META_SCOPED_INTROSPECTION_FLAG)
 
   // [Ali] Exclude preflightCheck from query key — it controls how the query
   // executes (whether an EXPLAIN guard runs first), not what data is returned.
@@ -407,9 +414,10 @@ export const useTableRowsQuery = <TData = TableRowsData>(
       table: { id: tableId },
       readReplicaIdentifier,
       ...queryKeyArgs,
+      scoped,
     }),
     queryFn: ({ signal }) =>
-      getTableRows({ queryClient, projectRef, connectionString, tableId, ...args }, signal),
+      getTableRows({ queryClient, projectRef, connectionString, tableId, ...args, scoped }, signal),
     enabled:
       enabled &&
       typeof projectRef !== 'undefined' &&
diff --git a/packages/pg-meta/README.md b/packages/pg-meta/README.md
new file mode 100644
index 00000000000..52002d1ebde
--- /dev/null
+++ b/packages/pg-meta/README.md
@@ -0,0 +1,96 @@
+# @supabase/pg-meta
+
+SQL builders for Postgres catalog introspection, shared by Supabase Studio and
+[`postgres-meta`](https://github.com/supabase/postgres-meta). Each builder in
+`src/sql/` returns a safe, parameterized SQL fragment (`SafeSqlFragment`) that a
+caller executes against a user's live database to read schema metadata — tables,
+columns, constraints, indexes, relationships, entity definitions, and so on.
+
+The `studio/` subtree holds the queries the Studio dashboard runs on every page
+open (Table Editor, Database pages, entity lists, definitions).
+
+## Catalog query plan guard
+
+### Why this exists
+
+These queries run against **the user's live catalog**, whose size we don't
+control. A real production catalog had ~267K `pg_class` rows. Before
+[#47894](https://github.com/supabase/supabase/pull/47894), several CTEs in the
+Table Editor query were **unscoped**: they scanned `pg_index`/`pg_constraint`
+across the whole catalog regardless of which table was being opened. That turned
+a single Table Editor open into `O(catalog)` sequential scans — 30–58s of work,
+tripping statement timeouts, on large catalogs.
+
+The fix scoped those CTEs to the requested table OID. To keep that class of
+regression out for good, the package has a **plan guard**: a test suite that
+builds a large synthetic catalog and asserts, via `EXPLAIN (ANALYZE, FORMAT
+JSON)`, that each hot-path query's plan stays scoped.
+
+- `test/db/stress-catalog.ts` — builds a synthetic `stress` schema (default 2000
+  tables, plus a view, a materialized view, and a partitioned table).
+- `test/db/plan-guard.ts` — `explainAnalyze()` + `assertPlanWithinBudget()` and
+  the tolerated tiny-catalog set.
+- `test/sql/studio/catalog-plan-guard.test.ts` — one budget per covered query.
+
+### THE RULE for new queries
+
+> Every new introspection query added under `src/sql/` that runs on a user's
+> live catalog **must** get a budget entry in
+> `test/sql/studio/catalog-plan-guard.test.ts`.
+
+Sequential scans over catalogs that **scale with schema size** — `pg_class`,
+`pg_attribute`, `pg_index`, `pg_constraint`, `pg_attrdef`, `pg_description`,
+`pg_depend`, `pg_policy`, `pg_trigger`, `pg_rewrite`, … — are only acceptable
+with a **written structural justification**. An unscoped scan with no such
+justification is a bug: scope the query to the requested OID/schema so it uses
+an index instead.
+
+### How to add a budget entry
+
+Run the query through the harness against the stress catalog and set a budget:
+
+```ts
+test('getMyNewSql: plan stays scoped', async () => {
+  const result = await explainAnalyze(db, getMyNewSql({ id: someTableId }))
+  assertPlanWithinBudget(result, {
+    // Omit allowedSeqScans entirely for a per-object query that must be fully
+    // index-scoped. Add an entry only for a structurally unavoidable scan:
+    allowedSeqScans: {
+      pg_constraint: {
+        max: 2,
+        reason: 'no index on pg_constraint.confrelid — incoming-FK lookup',
+      },
+    },
+    maxExecutionTimeMs: 1000, // default; loosen only with a comment
+  })
+})
+```
+
+### What's tolerated
+
+- **Tiny, fixed-size catalogs** (`pg_namespace`, `pg_foreign_table`,
+  `pg_foreign_server`, `pg_foreign_data_wrapper`, `pg_enum`, `pg_proc`) are
+  always tolerated. The planner full-scans them because they hold a handful of
+  rows and don't grow with table count. See `TINY_NON_SCALING_CATALOGS`.
+- **Justified structural scans** on scaling catalogs — e.g. there is no index on
+  `pg_constraint.confrelid`, so the incoming-FK half of a relationships lookup
+  must seq scan; a per-schema listing can't prune `pg_class` because there is no
+  index on `relnamespace` alone. Each such scan carries a `reason` string and a
+  `max` node count.
+
+Budget judgment:
+
+- **Per-object** queries (single id/table) allow **no** seq scans on scaling
+  catalogs except structurally unavoidable ones (each with a reason).
+- **Per-schema / list** queries may need one structural scan (e.g. `pg_class`
+  has no `relnamespace`-only index) — allow it with a reason and keep a time
+  bound.
+
+### Reproduce at scale locally
+
+The default of 2000 tables keeps CI fast. To investigate closer to real incident
+scale, crank the table count:
+
+```bash
+PG_META_STRESS_TABLES=12000 pnpm --filter @supabase/pg-meta test catalog-plan-guard
+```
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 76530c20f79..070921b0c6e 100644
--- a/packages/pg-meta/src/sql/studio/database/table-definition.ts
+++ b/packages/pg-meta/src/sql/studio/database/table-definition.ts
@@ -11,7 +11,20 @@
 
 import { joinSqlFragments, literal, safeSql, type SafeSqlFragment } from '../../../pg-format'
 
-export const CREATE_PG_GET_TABLEDEF_SQL = safeSql`
+/**
+ * Builds the CREATE (pg_temp) pg_get_tabledef function block.
+ *
+ * PR #47894 replaced three O(catalog) information_schema scans (and a
+ * relnamespace::regnamespace cast per pg_class row) with direct string tests
+ * and an OID-scoped partial-index lookup. That NEW behavior is gated behind
+ * `scoped` (default false = legacy behavior) so Studio can roll it out behind a
+ * feature flag. When `scoped` is false the emitted function is byte-equivalent
+ * to the pre-PR version.
+ *
+ * The two branches are kept as complete, standalone templates (no interpolated
+ * conditional fragments) so each rendered statement is easy to read and diff.
+ */
+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');
 
@@ -706,9 +719,732 @@ export const CREATE_PG_GET_TABLEDEF_SQL = safeSql`
     END;
   $$;`
 
-export const getTableDefinitionSql = ({ id }: { id: number }): SafeSqlFragment => {
+const SCOPED_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');
+
+  -- SELECT * FROM pg_temp.pg_get_coldef('sample','orders','id');
+  -- DROP FUNCTION pg_temp.pg_get_coldef(text,text,text,boolean);
+  CREATE OR REPLACE FUNCTION pg_temp.pg_get_coldef(
+    in_schema text,
+    in_table  text,
+    in_column text,
+    oldway    boolean default False
+  )
+  RETURNS text
+  LANGUAGE plpgsql VOLATILE
+  AS
+  $$
+  DECLARE
+  v_coldef     text;
+  v_dt1        text;
+  v_dt2        text;
+  v_dt3        text;
+  v_nullable   boolean;
+  v_position   int;
+  v_identity   text;
+  v_generated  text;
+  v_hasdflt    boolean;
+  v_dfltexpr   text;
+
+  BEGIN
+    IF oldway THEN
+      SELECT pg_catalog.format_type(a.atttypid, a.atttypmod) INTO v_coldef FROM pg_namespace n, pg_class c, pg_attribute a, pg_type t
+      WHERE n.nspname = in_schema AND n.oid = c.relnamespace AND c.relname = in_table AND a.attname = in_column and a.attnum > 0 AND a.attrelid = c.oid AND a.atttypid = t.oid ORDER BY a.attnum;
+      -- RAISE NOTICE 'DEBUG: oldway=%',v_coldef;
+    ELSE
+      -- a.attrelid::regclass::text, a.attname
+      SELECT CASE WHEN a.atttypid = ANY ('{int,int8,int2}'::regtype[]) AND EXISTS (SELECT FROM pg_attrdef ad WHERE ad.adrelid = a.attrelid AND ad.adnum   = a.attnum AND
+      pg_get_expr(ad.adbin, ad.adrelid) = 'nextval(''' || (pg_get_serial_sequence (a.attrelid::regclass::text, a.attname))::regclass || '''::regclass)') THEN CASE a.atttypid
+      WHEN 'int'::regtype  THEN 'serial' WHEN 'int8'::regtype THEN 'bigserial' WHEN 'int2'::regtype THEN 'smallserial' END ELSE format_type(a.atttypid, a.atttypmod) END AS data_type
+      INTO v_coldef FROM pg_namespace n, pg_class c, pg_attribute a, pg_type t
+      WHERE n.nspname = in_schema AND n.oid = c.relnamespace AND c.relname = in_table AND a.attname = in_column and a.attnum > 0 AND a.attrelid = c.oid AND a.atttypid = t.oid ORDER BY a.attnum;
+      -- RAISE NOTICE 'DEBUG: newway=%',v_coldef;
+
+      -- Issue#24: not implemented yet
+      -- might replace with this below to do more detailed parsing...
+      -- SELECT a.atttypid::regtype AS dt1, format_type(a.atttypid, a.atttypmod) as dt2, t.typname as dt3, CASE WHEN not(a.attnotnull) THEN True ELSE False END AS nullable,
+      -- a.attnum, a.attidentity, a.attgenerated, a.atthasdef, pg_get_expr(ad.adbin, ad.adrelid) dfltexpr
+      -- INTO v_dt1, v_dt2, v_dt3, v_nullable, v_position, v_identity, v_generated, v_hasdflt, v_dfltexpr
+      -- FROM pg_attribute a JOIN pg_class c ON (a.attrelid = c.oid) JOIN pg_type t ON (a.atttypid = t.oid) LEFT JOIN pg_attrdef ad ON (a.attrelid = ad.adrelid AND a.attnum = ad.adnum)
+      -- WHERE c.relkind in ('r','p') AND a.attnum > 0 AND NOT a.attisdropped AND c.relnamespace::regnamespace::text = in_schema AND c.relname = in_table AND a.attname = in_column;
+      -- RAISE NOTICE 'schema=%  table=%  column=%  dt1=%  dt2=%  dt3=%  nullable=%  pos=%  identity=%   generated=%  HasDefault=%  DeftExpr=%', in_schema, in_table, in_column, v_dt1,v_dt2,v_dt3,v_nullable,v_position,v_identity,v_generated,v_hasdflt,v_dfltexpr;
+    END IF;
+    RETURN v_coldef;
+  END;
+  $$;
+
+  -- SELECT * FROM pg_temp.pg_get_tabledef('sample', 'address', false);
+  DROP FUNCTION IF EXISTS pg_temp.pg_get_tabledef(character varying,character varying,boolean,tabledefs[]);
+  CREATE OR REPLACE FUNCTION pg_temp.pg_get_tabledef(
+    in_schema varchar,
+    in_table varchar,
+    _verbose boolean,
+    VARIADIC arr pg_temp.tabledefs[] DEFAULT '{}':: pg_temp.tabledefs[]
+  )
+  RETURNS text
+  LANGUAGE plpgsql VOLATILE
+  AS
+  $$
+    DECLARE
+      v_qualified text := '';
+      v_table_ddl text;
+      v_table_oid int;
+      v_colrec record;
+      v_constraintrec record;
+      v_trigrec       record;
+      v_indexrec record;
+      v_rec           record;
+      v_constraint_name text;
+      v_constraint_def  text;
+      v_pkey_def        text := '';
+      v_fkey_def        text := '';
+      v_fkey_defs       text := '';
+      v_trigger text := '';
+      v_partition_key text := '';
+      v_partbound text;
+      v_parent text;
+      v_parent_schema text;
+      v_persist text;
+      v_temp  text := '';
+      v_temp2 text;
+      v_relopts text;
+      v_tablespace text;
+      v_pgversion int;
+      bSerial boolean;
+      bPartition boolean;
+      bInheritance boolean;
+      bRelispartition boolean;
+      constraintarr text[] := '{}';
+      constraintelement text;
+      bSkip boolean;
+      bVerbose boolean := False;
+      v_cnt1   integer;
+      v_cnt2   integer;
+      search_path_old text := '';
+      search_path_new text := '';
+      v_partial    boolean;
+      v_pos        integer;
+
+      -- assume defaults for ENUMs at the getgo
+      pkcnt            int := 0;
+      fkcnt            int := 0;
+      trigcnt          int := 0;
+      cmtcnt           int := 0;
+      pktype           pg_temp.tabledefs := 'PKEY_INTERNAL';
+      fktype           pg_temp.tabledefs := 'FKEYS_INTERNAL';
+      trigtype         pg_temp.tabledefs := 'NO_TRIGGERS';
+      arglen           integer;
+      vargs            text;
+      avarg            pg_temp.tabledefs;
+
+      -- exception variables
+      v_ret            text;
+      v_diag1          text;
+      v_diag2          text;
+      v_diag3          text;
+      v_diag4          text;
+      v_diag5          text;
+      v_diag6          text;
+
+    BEGIN
+      SET client_min_messages = 'notice';
+      IF _verbose THEN bVerbose = True; END IF;
+
+      -- v17 fix: handle case-sensitive
+      -- v_qualified = in_schema || '.' || in_table;
+
+      arglen := array_length($4, 1);
+      IF arglen IS NULL THEN
+          -- nothing to do, so assume defaults
+          NULL;
+      ELSE
+          -- loop thru args
+          -- IF 'NO_TRIGGERS' = ANY ($4)
+          -- select array_to_string($4, ',', '***') INTO vargs;
+          IF bVerbose THEN RAISE NOTICE 'arguments=%', $4; END IF;
+          FOREACH avarg IN ARRAY $4 LOOP
+              IF bVerbose THEN RAISE NOTICE 'arg=%', avarg; END IF;
+              IF avarg = 'FKEYS_INTERNAL' OR avarg = 'FKEYS_EXTERNAL' OR avarg = 'FKEYS_NONE' THEN
+                  fkcnt = fkcnt + 1;
+                  fktype = avarg;
+              ELSEIF avarg = 'INCLUDE_TRIGGERS' OR avarg = 'NO_TRIGGERS' THEN
+                  trigcnt = trigcnt + 1;
+                  trigtype = avarg;
+              ELSEIF avarg = 'PKEY_EXTERNAL' THEN
+                  pkcnt = pkcnt + 1;
+                  pktype = avarg;
+              ELSEIF avarg = 'COMMENTS' THEN
+                  cmtcnt = cmtcnt + 1;
+
+              END IF;
+          END LOOP;
+          IF fkcnt > 1 THEN
+              RAISE WARNING 'Only one foreign key option can be provided. You provided %', fkcnt;
+              RETURN '';
+          ELSEIF trigcnt > 1 THEN
+              RAISE WARNING 'Only one trigger option can be provided. You provided %', trigcnt;
+              RETURN '';
+          ELSEIF pkcnt > 1 THEN
+              RAISE WARNING 'Only one pkey option can be provided. You provided %', pkcnt;
+              RETURN '';
+          ELSEIF cmtcnt > 1 THEN
+              RAISE WARNING 'Only one comments option can be provided. You provided %', cmtcnt;
+              RETURN '';
+
+          END IF;
+      END IF;
+
+      SELECT c.oid, (select setting from pg_settings where name = 'server_version_num') INTO v_table_oid, v_pgversion FROM pg_catalog.pg_class c LEFT JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace
+      WHERE c.relkind in ('r','p') AND c.relname = in_table AND n.nspname = in_schema;
+
+    -- set search_path = public before we do anything to force explicit schema qualification but dont forget to set it back before exiting...
+      SELECT setting INTO search_path_old FROM pg_settings WHERE name = 'search_path';
+
+      -- RAISE NOTICE 'DEBUG tableddl: saving old search_path: ***%***', search_path_old;
+      EXECUTE 'SET search_path = "public"';
+      SELECT setting INTO search_path_new FROM pg_settings WHERE name = 'search_path';
+      -- RAISE NOTICE 'DEBUG tableddl: using new search path=***%***', search_path_new;
+
+      -- throw an error if table was not found
+      IF (v_table_oid IS NULL) THEN
+        RAISE EXCEPTION 'table does not exist';
+      END IF;
+
+      -- get user-defined tablespaces if applicable
+      SELECT tablespace INTO v_temp FROM pg_tables WHERE schemaname = in_schema and tablename = in_table and tablespace IS NOT NULL;
+      IF v_temp IS NULL THEN
+        v_tablespace := 'TABLESPACE pg_default';
+      ELSE
+        v_tablespace := 'TABLESPACE ' || v_temp;
+      END IF;
+
+      -- also see if there are any SET commands for this table, ie, autovacuum_enabled=off, fillfactor=70
+      WITH relopts AS (SELECT unnest(c.reloptions) relopts FROM pg_class c, pg_namespace n WHERE n.nspname = in_schema and n.oid = c.relnamespace and c.relname = in_table)
+      SELECT string_agg(r.relopts, ', ') as relopts INTO v_temp from relopts r;
+      IF v_temp IS NULL THEN
+        v_relopts := '';
+      ELSE
+        v_relopts := ' WITH (' || v_temp || ')';
+      END IF;
+
+      -- -----------------------------------------------------------------------------------
+      -- Create table defs for partitions/children using inheritance or declarative methods.
+      -- inheritance: pg_class.relkind = 'r'   pg_class.relispartition=false   pg_class.relpartbound is NULL
+      -- declarative: pg_class.relkind = 'r'   pg_class.relispartition=true    pg_class.relpartbound is NOT NULL
+      -- -----------------------------------------------------------------------------------
+      v_partbound := '';
+      bPartition := False;
+      bInheritance := False;
+      IF v_pgversion < 100000 THEN
+        -- Issue#11: handle parent schema
+        SELECT c2.relname parent, c2.relnamespace::regnamespace INTO v_parent, v_parent_schema from pg_class c1, pg_namespace n, pg_inherits i, pg_class c2
+        WHERE n.nspname = in_schema and n.oid = c1.relnamespace and c1.relname = in_table and c1.oid = i.inhrelid and i.inhparent = c2.oid and c1.relkind = 'r';
+        IF (v_parent IS NOT NULL) THEN
+          bPartition   := True;
+          bInheritance := True;
+        END IF;
+      ELSE
+        -- Issue#11: handle parent schema
+        SELECT c2.relname parent, c1.relispartition, pg_get_expr(c1.relpartbound, c1.oid, true), c2.relnamespace::regnamespace INTO v_parent, bRelispartition, v_partbound, v_parent_schema from pg_class c1, pg_namespace n, pg_inherits i, pg_class c2
+        WHERE n.nspname = in_schema and n.oid = c1.relnamespace and c1.relname = in_table and c1.oid = i.inhrelid and i.inhparent = c2.oid and c1.relkind = 'r';
+        IF (v_parent IS NOT NULL) THEN
+          bPartition   := True;
+          IF bRelispartition THEN
+            bInheritance := False;
+          ELSE
+            bInheritance := True;
+          END IF;
+        END IF;
+      END IF;
+      IF bPartition THEN
+        --Issue#17 fix for case-sensitive tables
+        -- Supabase perf fix: the original scanned all of information_schema.tables (O(catalog) with
+        -- per-row privilege checks) just to detect uppercase in the table name; the name is already
+        -- in hand, so test it directly. The table's existence was validated above via v_table_oid.
+        -- SELECT count(*) INTO v_cnt1 FROM information_schema.tables t WHERE EXISTS (SELECT REGEXP_MATCHES(s.table_name, '([A-Z]+)','g') FROM information_schema.tables s
+        -- WHERE t.table_schema=s.table_schema AND t.table_name=s.table_name AND t.table_schema = in_schema AND t.table_name = in_table AND t.table_type = 'BASE TABLE');
+        v_cnt1 := CASE WHEN in_table ~ '[A-Z]' THEN 1 ELSE 0 END;
+
+        --Issue#19 put double-quotes around SQL keyword column names
+        -- Issue#121: fix keyword lookup for table name not column name that does not apply here
+        -- SELECT COUNT(*) INTO v_cnt2 FROM pg_get_keywords() WHERE word = v_colrec.column_name AND catcode = 'R';
+        SELECT COUNT(*) INTO v_cnt2 FROM pg_get_keywords() WHERE word = in_table AND catcode = 'R';
+
+        IF bInheritance THEN
+          -- inheritance-based
+          IF v_cnt1 > 0 OR v_cnt2 > 0 THEN
+            v_table_ddl := 'CREATE TABLE ' || in_schema || '."' || in_table || '"( '|| E'\\n';
+          ELSE
+            v_table_ddl := 'CREATE TABLE ' || in_schema || '.' || in_table || '( '|| E'\\n';
+          END IF;
+
+          -- Jump to constraints section to add the check constraints
+        ELSE
+          -- declarative-based
+          IF v_relopts <> '' THEN
+            IF v_cnt1 > 0 OR v_cnt2 > 0 THEN
+              v_table_ddl := 'CREATE TABLE ' || in_schema || '."' || in_table || '" PARTITION OF ' || in_schema || '.' || v_parent || ' ' || v_partbound || v_relopts || ' ' || v_tablespace || '; ' || E'\\n';
+            ELSE
+              v_table_ddl := 'CREATE TABLE ' || in_schema || '.' || in_table || ' PARTITION OF ' || in_schema || '.' || v_parent || ' ' || v_partbound || v_relopts || ' ' || v_tablespace || '; ' || E'\\n';
+            END IF;
+          ELSE
+            IF v_cnt1 > 0 OR v_cnt2 > 0 THEN
+              v_table_ddl := 'CREATE TABLE ' || in_schema || '."' || in_table || '" PARTITION OF ' || in_schema || '.' || v_parent || ' ' || v_partbound || ' ' || v_tablespace || '; ' || E'\\n';
+            ELSE
+              v_table_ddl := 'CREATE TABLE ' || in_schema || '.' || in_table || ' PARTITION OF ' || in_schema || '.' || v_parent || ' ' || v_partbound || ' ' || v_tablespace || '; ' || E'\\n';
+            END IF;
+          END IF;
+          -- Jump to constraints and index section to add the check constraints and indexes and perhaps FKeys
+        END IF;
+      END IF;
+      IF bVerbose THEN RAISE NOTICE '(1)tabledef so far: %', v_table_ddl; END IF;
+
+      IF NOT bPartition THEN
+        -- see if this is unlogged or temporary table
+        select c.relpersistence into v_persist from pg_class c, pg_namespace n where n.nspname = in_schema and n.oid = c.relnamespace and c.relname = in_table and c.relkind = 'r';
+        IF v_persist = 'u' THEN
+          v_temp := 'UNLOGGED';
+        ELSIF v_persist = 't' THEN
+          v_temp := 'TEMPORARY';
+        ELSE
+          v_temp := '';
+        END IF;
+      END IF;
+
+      -- start the create definition for regular tables unless we are in progress creating an inheritance-based child table
+      IF NOT bPartition THEN
+        --Issue#17 fix for case-sensitive tables
+        -- Supabase perf fix: same as the partition branch above — replace the O(catalog)
+        -- information_schema.tables scan with a direct uppercase test on the known table name.
+        -- SELECT count(*) INTO v_cnt1 FROM information_schema.tables t WHERE EXISTS (SELECT REGEXP_MATCHES(s.table_name, '([A-Z]+)','g') FROM information_schema.tables s
+        -- WHERE t.table_schema=s.table_schema AND t.table_name=s.table_name AND t.table_schema = in_schema AND t.table_name = in_table AND t.table_type = 'BASE TABLE');
+        v_cnt1 := CASE WHEN in_table ~ '[A-Z]' THEN 1 ELSE 0 END;
+        IF v_cnt1 > 0 THEN
+          v_table_ddl := 'CREATE ' || v_temp || ' TABLE ' || in_schema || '."' || in_table || '" (' || E'\\n';
+        ELSE
+          v_table_ddl := 'CREATE ' || v_temp || ' TABLE ' || in_schema || '.' || in_table || ' (' || E'\\n';
+        END IF;
+      END IF;
+      -- RAISE NOTICE 'DEBUG2: tabledef so far: %', v_table_ddl;
+      -- define all of the columns in the table unless we are in progress creating an inheritance-based child table
+      IF NOT bPartition THEN
+        FOR v_colrec IN
+          SELECT c.column_name, c.data_type, c.udt_name, c.udt_schema, c.character_maximum_length, c.is_nullable, c.column_default, c.numeric_precision, c.numeric_scale, c.is_identity, c.identity_generation, c.is_generated, c.generation_expression
+          FROM information_schema.columns c WHERE (table_schema, table_name) = (in_schema, in_table) ORDER BY ordinal_position
+        LOOP
+          IF bVerbose THEN RAISE NOTICE '(col loop) name=%  type=%  udt_name=%  default=%  is_generated=%  gen_expr=%', v_colrec.column_name, v_colrec.data_type, v_colrec.udt_name, v_colrec.column_default, v_colrec.is_generated, v_colrec.generation_expression; END IF;
+
+          -- v17 fix: handle case-sensitive for pg_get_serial_sequence that requires SQL Identifier handling
+          -- SELECT CASE WHEN pg_get_serial_sequence(v_qualified, v_colrec.column_name) IS NOT NULL THEN True ELSE False END into bSerial;
+          SELECT CASE WHEN pg_get_serial_sequence(quote_ident(in_schema) || '.' || quote_ident(in_table), v_colrec.column_name) IS NOT NULL THEN True ELSE False END into bSerial;
+          IF bVerbose THEN
+            -- v17 fix: handle case-sensitive for pg_get_serial_sequence that requires SQL Identifier handling
+            -- SELECT pg_get_serial_sequence(v_qualified, v_colrec.column_name) into v_temp;
+            SELECT pg_get_serial_sequence(quote_ident(in_schema) || '.' || quote_ident(in_table), v_colrec.column_name) into v_temp;
+            IF v_temp IS NULL THEN v_temp = 'NA'; END IF;
+            SELECT pg_temp.pg_get_coldef(in_schema, in_table,v_colrec.column_name) INTO v_diag1;
+            RAISE NOTICE 'DEBUG table: %  Column: %  datatype: %  Serial=%  serialval=%  coldef=%', v_qualified, v_colrec.column_name, v_colrec.data_type, bSerial, v_temp, v_diag1;
+            RAISE NOTICE 'DEBUG tabledef: %', v_table_ddl;
+          END IF;
+
+          --Issue#17 put double-quotes around case-sensitive column names
+          -- Supabase perf fix: the original scanned all of information_schema.columns (O(total
+          -- columns in the database), with per-row privilege checks) PER COLUMN just to detect
+          -- uppercase in the column name — the dominant cost of this function on large catalogs.
+          -- The name is already in hand, so test it directly. The quote_ident(in_schema) = in_schema
+          -- comparison preserves the original's behavior of never matching (count 0) when the
+          -- schema name itself needs quoting, since it compared t.table_schema = quote_ident(in_schema).
+          -- SELECT COUNT(*) INTO v_cnt1 FROM information_schema.columns t WHERE EXISTS (SELECT REGEXP_MATCHES(s.column_name, '([A-Z]+)','g') FROM information_schema.columns s
+          -- WHERE t.table_schema=s.table_schema and t.table_name=s.table_name and t.column_name=s.column_name AND t.table_schema = quote_ident(in_schema) AND column_name = v_colrec.column_name);
+          v_cnt1 := CASE WHEN quote_ident(in_schema) = in_schema AND v_colrec.column_name ~ '[A-Z]' THEN 1 ELSE 0 END;
+
+          --Issue#19 put double-quotes around SQL keyword column names
+          SELECT COUNT(*) INTO v_cnt2 FROM pg_get_keywords() WHERE word = v_colrec.column_name AND catcode = 'R';
+
+          IF v_cnt1 > 0 OR v_cnt2 > 0 THEN
+            v_table_ddl := v_table_ddl || '  "' || v_colrec.column_name || '" ';
+          ELSE
+            v_table_ddl := v_table_ddl || '  ' || v_colrec.column_name || ' ';
+          END IF;
+
+          -- Issue#23: Handle autogenerated columns and rewrite as a simpler IF THEN ELSE branch instead of a much more complex embedded CASE STATEMENT
+          IF v_colrec.is_generated = 'ALWAYS' and v_colrec.generation_expression IS NOT NULL THEN
+              -- searchable tsvector GENERATED ALWAYS AS (to_tsvector('simple'::regconfig, COALESCE(translate(email, '@.-'::citext, ' '::text), ''::text)) ) STORED
+              v_temp = v_colrec.data_type || ' GENERATED ALWAYS AS (' || v_colrec.generation_expression || ') STORED ';
+          ELSEIF v_colrec.udt_name in ('geometry', 'box2d', 'box2df', 'box3d', 'geography', 'geometry_dump', 'gidx', 'spheroid', 'valid_detail') THEN
+              v_temp = v_colrec.udt_name;
+          ELSEIF v_colrec.data_type = 'USER-DEFINED' THEN
+              v_temp = v_colrec.udt_schema || '.' || v_colrec.udt_name;
+          ELSEIF v_colrec.data_type = 'ARRAY' THEN
+                -- Issue#6 fix: handle arrays
+              v_temp = pg_temp.pg_get_coldef(in_schema, in_table,v_colrec.column_name);
+              -- v17 fix: handle case-sensitive for pg_get_serial_sequence that requires SQL Identifier handling
+              -- WHEN pg_get_serial_sequence(v_qualified, v_colrec.column_name) IS NOT NULL
+          ELSEIF pg_get_serial_sequence(quote_ident(in_schema) || '.' || quote_ident(in_table), v_colrec.column_name) IS NOT NULL THEN
+              -- Issue#8 fix: handle serial. Note: NOT NULL is implied so no need to declare it explicitly
+              v_temp = pg_temp.pg_get_coldef(in_schema, in_table,v_colrec.column_name);
+          ELSE
+              v_temp = v_colrec.data_type;
+          END IF;
+          -- RAISE NOTICE 'column def1=%', v_temp;
+
+          -- handle IDENTITY columns
+          IF v_colrec.is_identity = 'YES' THEN
+              IF v_colrec.identity_generation = 'ALWAYS' THEN
+                  v_temp = v_temp || ' GENERATED ALWAYS AS IDENTITY';
+              ELSE
+                  v_temp = v_temp || ' GENERATED BY DEFAULT AS IDENTITY';
+              END IF;
+          ELSEIF v_colrec.character_maximum_length IS NOT NULL THEN
+              v_temp = v_temp || ('(' || v_colrec.character_maximum_length || ')');
+          ELSEIF v_colrec.numeric_precision > 0 AND v_colrec.numeric_scale > 0 THEN
+              v_temp = v_temp || '(' || v_colrec.numeric_precision || ',' || v_colrec.numeric_scale || ')';
+          END IF;
+
+          -- Handle NULL/NOT NULL
+          IF bSerial THEN
+              v_temp = v_temp || ' NOT NULL';
+          ELSEIF v_colrec.is_nullable = 'NO' THEN
+              v_temp = v_temp || ' NOT NULL';
+          ELSEIF v_colrec.is_nullable = 'YES' THEN
+              v_temp = v_temp || ' NULL';
+          END IF;
+
+          -- Handle defaults
+          IF v_colrec.column_default IS NOT null AND NOT bSerial THEN
+              -- RAISE NOTICE 'Setting default for column, %', v_colrec.column_name;
+              v_temp = v_temp || (' DEFAULT ' || v_colrec.column_default);
+          END IF;
+          v_temp = v_temp || ',' || E'\\n';
+          -- RAISE NOTICE 'column def2=%', v_temp;
+          v_table_ddl := v_table_ddl || v_temp;
+          -- RAISE NOTICE 'tabledef=%', v_table_ddl;
+
+        END LOOP;
+      END IF;
+      IF bVerbose THEN RAISE NOTICE '(2)tabledef so far: %', v_table_ddl; END IF;
+
+      -- define all the constraints: conparentid does not exist pre PGv11
+      IF v_pgversion < 110000 THEN
+        FOR v_constraintrec IN
+          SELECT con.conname as constraint_name, con.contype as constraint_type,
+            CASE
+              WHEN con.contype = 'p' THEN 1 -- primary key constraint
+              WHEN con.contype = 'u' THEN 2 -- unique constraint
+              WHEN con.contype = 'f' THEN 3 -- foreign key constraint
+              WHEN con.contype = 'c' THEN 4
+              ELSE 5
+            END as type_rank,
+            pg_get_constraintdef(con.oid) as constraint_definition
+          FROM pg_catalog.pg_constraint con JOIN pg_catalog.pg_class rel ON rel.oid = con.conrelid JOIN pg_catalog.pg_namespace nsp ON nsp.oid = connamespace
+          WHERE nsp.nspname = in_schema AND rel.relname = in_table ORDER BY type_rank
+        LOOP
+          v_constraint_name := v_constraintrec.constraint_name;
+          v_constraint_def  := v_constraintrec.constraint_definition;
+          IF v_constraintrec.type_rank = 1 THEN
+              IF pkcnt = 0 OR pktype = 'PKEY_INTERNAL' THEN
+                  -- internal def
+                  v_constraint_name := v_constraintrec.constraint_name;
+                  v_constraint_def  := v_constraintrec.constraint_definition;
+                  v_table_ddl := v_table_ddl || '  ' -- note: two char spacer to start, to indent the column
+                    || 'CONSTRAINT' || ' '
+                    || v_constraint_name || ' '
+                    || v_constraint_def
+                    || ',' || E'\\n';
+              ELSE
+                -- Issue#16 handle external PG def
+                SELECT 'ALTER TABLE ONLY ' || in_schema || '.' || c.relname || ' ADD CONSTRAINT ' || r.conname || ' ' || pg_catalog.pg_get_constraintdef(r.oid, true) || ';' INTO v_pkey_def
+                FROM pg_catalog.pg_constraint r, pg_class c, pg_namespace n where r.conrelid = c.oid and  r.contype = 'p' and n.oid = r.connamespace and n.nspname = in_schema AND c.relname = in_table and r.conname = v_constraint_name;
+              END IF;
+              IF bPartition THEN
+                continue;
+              END IF;
+          ELSIF v_constraintrec.type_rank = 3 THEN
+              -- handle foreign key constraints
+              --Issue#22 fix: added FKEY_NONE check
+              IF fktype = 'FKEYS_NONE' THEN
+                  -- skip
+                  continue;
+              ELSIF fkcnt = 0 OR fktype = 'FKEYS_INTERNAL' THEN
+                  -- internal def
+                  v_table_ddl := v_table_ddl || '  ' -- note: two char spacer to start, to indent the column
+                    || 'CONSTRAINT' || ' '
+                    || v_constraint_name || ' '
+                    || v_constraint_def
+                    || ',' || E'\\n';
+              ELSE
+                  -- external def
+                  SELECT 'ALTER TABLE ONLY ' || n.nspname || '.' || c2.relname || ' ADD CONSTRAINT ' || r.conname || ' ' || pg_catalog.pg_get_constraintdef(r.oid, true) || ';' INTO v_fkey_def
+                  FROM pg_constraint r, pg_class c1, pg_namespace n, pg_class c2 where r.conrelid = c1.oid and  r.contype = 'f' and n.nspname = in_schema and n.oid = r.connamespace and r.conrelid = c2.oid and c2.relname = in_table;
+                  v_fkey_defs = v_fkey_defs || v_fkey_def || E'\\n';
+              END IF;
+          ELSE
+              -- handle all other constraints besides PKEY and FKEYS as internal defs by default
+              v_table_ddl := v_table_ddl || '  ' -- note: two char spacer to start, to indent the column
+                || 'CONSTRAINT' || ' '
+                || v_constraint_name || ' '
+                || v_constraint_def
+                || ',' || E'\\n';
+          END IF;
+          if bVerbose THEN RAISE NOTICE 'DEBUG4: constraint name=% constraint_def=%', v_constraint_name,v_constraint_def; END IF;
+          constraintarr := constraintarr || v_constraintrec.constraint_name:: text;
+
+        END LOOP;
+      ELSE
+        -- handle PG versions 11 and up
+        -- Issue#20: Fix logic for external PKEY and FKEYS
+        FOR v_constraintrec IN
+          SELECT con.conname as constraint_name, con.contype as constraint_type,
+            CASE
+              WHEN con.contype = 'p' THEN 1 -- primary key constraint
+              WHEN con.contype = 'u' THEN 2 -- unique constraint
+              WHEN con.contype = 'f' THEN 3 -- foreign key constraint
+              WHEN con.contype = 'c' THEN 4
+              ELSE 5
+            END as type_rank,
+            pg_get_constraintdef(con.oid) as constraint_definition
+          FROM pg_catalog.pg_constraint con JOIN pg_catalog.pg_class rel ON rel.oid = con.conrelid JOIN pg_catalog.pg_namespace nsp ON nsp.oid = connamespace
+          WHERE nsp.nspname = in_schema AND rel.relname = in_table
+                --Issue#13 added this condition:
+                AND con.conparentid = 0
+                ORDER BY type_rank
+        LOOP
+          v_constraint_name := v_constraintrec.constraint_name;
+          v_constraint_def  := v_constraintrec.constraint_definition;
+          IF v_constraintrec.type_rank = 1 THEN
+              IF pkcnt = 0 OR pktype = 'PKEY_INTERNAL' THEN
+                  -- internal def
+                  v_constraint_name := v_constraintrec.constraint_name;
+                  v_constraint_def  := v_constraintrec.constraint_definition;
+                  v_table_ddl := v_table_ddl || '  ' -- note: two char spacer to start, to indent the column
+                    || 'CONSTRAINT' || ' '
+                    || v_constraint_name || ' '
+                    || v_constraint_def
+                    || ',' || E'\\n';
+              ELSE
+                -- Issue#16 handle external PG def
+                SELECT 'ALTER TABLE ONLY ' || in_schema || '.' || c.relname || ' ADD CONSTRAINT ' || r.conname || ' ' || pg_catalog.pg_get_constraintdef(r.oid, true) || ';' INTO v_pkey_def
+                FROM pg_catalog.pg_constraint r, pg_class c, pg_namespace n where r.conrelid = c.oid and  r.contype = 'p' and n.oid = r.connamespace and n.nspname = in_schema AND c.relname = in_table;
+              END IF;
+              IF bPartition THEN
+                continue;
+              END IF;
+          ELSIF v_constraintrec.type_rank = 3 THEN
+              -- handle foreign key constraints
+              --Issue#22 fix: added FKEY_NONE check
+              IF fktype = 'FKEYS_NONE' THEN
+                  -- skip
+                  continue;
+              ELSIF fkcnt = 0 OR fktype = 'FKEYS_INTERNAL' THEN
+                  -- internal def
+                  v_table_ddl := v_table_ddl || '  ' -- note: two char spacer to start, to indent the column
+                    || 'CONSTRAINT' || ' '
+                    || v_constraint_name || ' '
+                    || v_constraint_def
+                    || ',' || E'\\n';
+              ELSE
+                  -- external def
+                  SELECT 'ALTER TABLE ONLY ' || n.nspname || '.' || c2.relname || ' ADD CONSTRAINT ' || r.conname || ' ' || pg_catalog.pg_get_constraintdef(r.oid, true) || ';' INTO v_fkey_def
+                  FROM pg_constraint r, pg_class c1, pg_namespace n, pg_class c2 where r.conrelid = c1.oid and  r.contype = 'f' and n.nspname = in_schema and n.oid = r.connamespace and r.conrelid = c2.oid and c2.relname = in_table and
+                  r.conname = v_constraint_name and r.conparentid = 0;
+                  v_fkey_defs = v_fkey_defs || v_fkey_def || E'\\n';
+              END IF;
+          ELSE
+              -- handle all other constraints besides PKEY and FKEYS as internal defs by default
+              v_table_ddl := v_table_ddl || '  ' -- note: two char spacer to start, to indent the column
+                || 'CONSTRAINT' || ' '
+                || v_constraint_name || ' '
+                || v_constraint_def
+                || ',' || E'\\n';
+          END IF;
+          if bVerbose THEN RAISE NOTICE 'DEBUG4: constraint name=% constraint_def=%', v_constraint_name,v_constraint_def; END IF;
+          constraintarr := constraintarr || v_constraintrec.constraint_name:: text;
+
+        END LOOP;
+      END IF;
+
+      -- drop the last comma before ending the create statement, which should be right before the carriage return character
+      -- Issue#24: make sure the comma is there before removing it
+      select substring(v_table_ddl, length(v_table_ddl) - 1, 1) INTO v_temp;
+      IF v_temp = ',' THEN
+          v_table_ddl = substr(v_table_ddl, 0, length(v_table_ddl) - 1) || E'\\n';
+      END IF;
+      IF bVerbose THEN RAISE NOTICE '(3)tabledef so far: %', trim(v_table_ddl); END IF;
+
+      -- ---------------------------------------------------------------------------
+      -- at this point we have everything up to the last table-enclosing parenthesis
+      -- ---------------------------------------------------------------------------
+      IF bVerbose THEN RAISE NOTICE '(4)tabledef so far: %', v_table_ddl; END IF;
+
+      -- See if this is an inheritance-based child table and finish up the table create.
+      IF bPartition and bInheritance THEN
+        -- Issue#11: handle parent schema
+        -- v_table_ddl := v_table_ddl || ') INHERITS (' || in_schema || '.' || v_parent || ') ' || E'\\n' || v_relopts || ' ' || v_tablespace || ';' || E'\\n';
+        IF v_parent_schema = '' OR v_parent_schema IS NULL THEN v_parent_schema = in_schema; END IF;
+        v_table_ddl := v_table_ddl || ') INHERITS (' || v_parent_schema || '.' || v_parent || ') ' || E'\\n' || v_relopts || ' ' || v_tablespace || ';' || E'\\n';
+      END IF;
+
+      IF v_pgversion >= 100000 AND NOT bPartition and NOT bInheritance THEN
+        -- See if this is a partitioned table (pg_class.relkind = 'p') and add the partitioned key
+        SELECT pg_get_partkeydef(c1.oid) as partition_key INTO v_partition_key FROM pg_class c1 JOIN pg_namespace n ON (n.oid = c1.relnamespace) LEFT JOIN pg_partitioned_table p ON (c1.oid = p.partrelid)
+        WHERE n.nspname = in_schema and n.oid = c1.relnamespace and c1.relname = in_table and c1.relkind = 'p';
+
+        IF v_partition_key IS NOT NULL AND v_partition_key <> '' THEN
+          -- add partition clause
+          -- NOTE:  cannot specify default tablespace for partitioned relations
+          -- v_table_ddl := v_table_ddl || ') PARTITION BY ' || v_partition_key || ' ' || v_tablespace || ';' || E'\\n';
+          v_table_ddl := v_table_ddl || ') PARTITION BY ' || v_partition_key || ';' || E'\\n';
+        ELSEIF v_relopts <> '' THEN
+          v_table_ddl := v_table_ddl || ') ' || v_relopts || ' ' || v_tablespace || ';' || E'\\n';
+        ELSE
+          -- end the create definition
+          v_table_ddl := v_table_ddl || ') ' || v_tablespace || ';' || E'\\n';
+        END IF;
+      END IF;
+
+      IF bVerbose THEN RAISE NOTICE '(5)tabledef so far: %', v_table_ddl; END IF;
+
+      -- Add closing paren for regular tables
+      -- IF NOT bPartition THEN
+      -- v_table_ddl := v_table_ddl || ') ' || v_relopts || ' ' || v_tablespace || E';\\n';
+      -- END IF;
+      -- RAISE NOTICE 'ddlsofar3: %', v_table_ddl;
+
+      -- Issue#16 create the external PKEY def if indicated
+      IF v_pkey_def <> '' THEN
+          v_table_ddl := v_table_ddl || v_pkey_def || E'\\n';
+      END IF;
+
+      -- Issue#20
+      IF v_fkey_defs <> '' THEN
+            v_table_ddl := v_table_ddl || v_fkey_defs || E'\\n';
+      END IF;
+
+      IF bVerbose THEN RAISE NOTICE '(6)tabledef so far: %', v_table_ddl; END IF;
+
+      -- create indexes
+      FOR v_indexrec IN
+        SELECT indexdef, COALESCE(tablespace, 'pg_default') as tablespace, indexname FROM pg_indexes WHERE (schemaname, tablename) = (in_schema, in_table)
+      LOOP
+        -- RAISE NOTICE 'DEBUG6: indexname=%  indexdef=%', v_indexrec.indexname, v_indexrec.indexdef;
+        -- loop through constraints and skip ones already defined
+        bSkip = False;
+        FOREACH constraintelement IN ARRAY constraintarr
+        LOOP
+          IF constraintelement = v_indexrec.indexname THEN
+              -- RAISE NOTICE 'DEBUG7: skipping index, %', v_indexrec.indexname;
+              bSkip = True;
+              EXIT;
+          END IF;
+        END LOOP;
+        if bSkip THEN CONTINUE; END IF;
+
+        -- Add IF NOT EXISTS clause so partition index additions will not be created if declarative partition in effect and index already created on parent
+        v_indexrec.indexdef := REPLACE(v_indexrec.indexdef, 'CREATE INDEX', 'CREATE INDEX IF NOT EXISTS');
+        -- Fix Issue#26: do it for unique/primary key indexes as well
+        v_indexrec.indexdef := REPLACE(v_indexrec.indexdef, 'CREATE UNIQUE INDEX', 'CREATE UNIQUE INDEX IF NOT EXISTS');
+        -- RAISE NOTICE 'DEBUG8: adding index, %', v_indexrec.indexname;
+
+        -- NOTE:  cannot specify default tablespace for partitioned relations
+        IF v_partition_key IS NOT NULL AND v_partition_key <> '' THEN
+            v_table_ddl := v_table_ddl || v_indexrec.indexdef || ';' || E'\\n';
+        ELSE
+            -- Issue#25: see if partial index or not
+            -- Supabase perf fix: scope by the table OID resolved earlier instead of casting
+            -- relnamespace::regnamespace::text for every pg_class row (O(catalog) per index).
+            -- Indexes always live in the same schema as their table, so the schema quals were
+            -- redundant with v_table_oid.
+            -- select CASE WHEN i.indpred IS NOT NULL THEN True ELSE False END INTO v_partial
+            -- FROM pg_index i JOIN pg_class c1 ON (i.indexrelid = c1.oid) JOIN pg_class c2 ON (i.indrelid = c2.oid)
+            -- WHERE c1.relnamespace::regnamespace::text = in_schema AND c2.relnamespace::regnamespace::text = in_schema AND c2.relname = in_table AND c1.relname = v_indexrec.indexname;
+            select CASE WHEN i.indpred IS NOT NULL THEN True ELSE False END INTO v_partial
+            FROM pg_index i JOIN pg_class c1 ON (i.indexrelid = c1.oid)
+            WHERE i.indrelid = v_table_oid AND c1.relname = v_indexrec.indexname;
+            IF v_partial THEN
+                -- Put tablespace def before WHERE CLAUSE
+                v_temp = v_indexrec.indexdef;
+                v_pos = POSITION(' WHERE ' IN v_temp);
+                v_temp2 = SUBSTRING(v_temp, v_pos);
+                v_temp  = SUBSTRING(v_temp, 1, v_pos);
+                v_table_ddl := v_table_ddl || v_temp || ' TABLESPACE ' || v_indexrec.tablespace || v_temp2 || ';' || E'\\n';
+            ELSE
+                v_table_ddl := v_table_ddl || v_indexrec.indexdef || ' TABLESPACE ' || v_indexrec.tablespace || ';' || E'\\n';
+            END IF;
+        END IF;
+
+      END LOOP;
+      IF bVerbose THEN RAISE NOTICE '(7)tabledef so far: %', v_table_ddl; END IF;
+
+      -- Issue#20: added logic for table and column comments
+      IF  cmtcnt > 0 THEN
+          FOR v_rec IN
+            SELECT c.relname, 'COMMENT ON ' || CASE WHEN c.relkind in ('r','p') AND a.attname IS NULL THEN 'TABLE ' WHEN c.relkind in ('r','p') AND a.attname IS NOT NULL THEN 'COLUMN ' WHEN c.relkind = 'f' THEN 'FOREIGN TABLE '
+                  WHEN c.relkind = 'm' THEN 'MATERIALIZED VIEW ' WHEN c.relkind = 'v' THEN 'VIEW ' WHEN c.relkind = 'i' THEN 'INDEX ' WHEN c.relkind = 'S' THEN 'SEQUENCE ' ELSE 'XX' END || n.nspname || '.' ||
+                  CASE WHEN c.relkind in ('r','p') AND a.attname IS NOT NULL THEN quote_ident(c.relname) || '.' || a.attname ELSE quote_ident(c.relname) END || ' IS '   || quote_literal(d.description) || ';' as ddl
+            FROM pg_class c JOIN pg_namespace n ON (n.oid = c.relnamespace) LEFT JOIN pg_description d ON (c.oid = d.objoid) LEFT JOIN pg_attribute a ON (c.oid = a.attrelid AND a.attnum > 0 and a.attnum = d.objsubid)
+            WHERE d.description IS NOT NULL AND n.nspname = in_schema AND c.relname = in_table ORDER BY 2 desc, ddl
+          LOOP
+              --RAISE NOTICE 'comments:%', v_rec.ddl;
+              v_table_ddl = v_table_ddl || v_rec.ddl || E'\\n';
+          END LOOP;
+      END IF;
+      IF bVerbose THEN RAISE NOTICE '(8)tabledef so far: %', v_table_ddl; END IF;
+
+      IF trigtype = 'INCLUDE_TRIGGERS' THEN
+        -- Issue#14: handle multiple triggers for a table
+        FOR v_trigrec IN
+            select pg_get_triggerdef(t.oid, True) || ';' as triggerdef FROM pg_trigger t, pg_class c, pg_namespace n
+            WHERE n.nspname = in_schema and n.oid = c.relnamespace and c.relname = in_table and c.relkind = 'r' and t.tgrelid = c.oid and NOT t.tgisinternal
+        LOOP
+            v_table_ddl := v_table_ddl || v_trigrec.triggerdef;
+            v_table_ddl := v_table_ddl || E'\\n';
+            IF bVerbose THEN RAISE NOTICE 'triggerdef = %', v_trigrec.triggerdef; END IF;
+        END LOOP;
+      END IF;
+
+      IF bVerbose THEN RAISE NOTICE '(9)tabledef so far: %', v_table_ddl; END IF;
+      -- add empty line
+      v_table_ddl := v_table_ddl || E'\\n';
+      IF bVerbose THEN RAISE NOTICE '(10)tabledef so far: %', v_table_ddl; END IF;
+
+      -- reset search_path back to what it was
+      IF search_path_old = '' THEN
+        SELECT set_config('search_path', '', false) into v_temp;
+      ELSE
+        EXECUTE 'SET search_path = ' || search_path_old;
+      END IF;
+
+      RETURN v_table_ddl;
+
+      EXCEPTION
+      WHEN others THEN
+      BEGIN
+        GET STACKED DIAGNOSTICS v_diag1 = MESSAGE_TEXT, v_diag2 = PG_EXCEPTION_DETAIL, v_diag3 = PG_EXCEPTION_HINT, v_diag4 = RETURNED_SQLSTATE, v_diag5 = PG_CONTEXT, v_diag6 = PG_EXCEPTION_CONTEXT;
+        -- v_ret := 'line=' || v_diag6 || '. '|| v_diag4 || '. ' || v_diag1 || ' .' || v_diag2 || ' .' || v_diag3;
+        v_ret := 'line=' || v_diag6 || '. '|| v_diag4 || '. ' || v_diag1;
+        RAISE EXCEPTION '%', v_ret;
+        -- put additional coding here if necessarY
+        RETURN '';
+      END;
+
+    END;
+  $$;`
+
+export const createPgGetTabledefSql = ({
+  scoped = false,
+}: { scoped?: boolean } = {}): SafeSqlFragment =>
+  scoped ? SCOPED_PG_GET_TABLEDEF_SQL : LEGACY_PG_GET_TABLEDEF_SQL
+
+export const getTableDefinitionSql = ({
+  id,
+  scoped = false,
+}: {
+  id: number
+  scoped?: boolean
+}): SafeSqlFragment => {
   const sql = safeSql`
-    ${CREATE_PG_GET_TABLEDEF_SQL}
+    ${createPgGetTabledefSql({ scoped })}
 
     with table_info as (
       select 
@@ -734,12 +1470,14 @@ export const getTableDefinitionSql = ({ id }: { id: number }): SafeSqlFragment =
 export const getEntityDefinitionsSql = ({
   schemas,
   limit = 100,
+  scoped = false,
 }: {
   schemas: string[]
   limit?: number
+  scoped?: boolean
 }): SafeSqlFragment => {
   const sql = safeSql`
-${CREATE_PG_GET_TABLEDEF_SQL}
+${createPgGetTabledefSql({ scoped })}
 
 with records as (
   select
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 8b0af9731d9..98347cf9a55 100644
--- a/packages/pg-meta/src/sql/studio/table-editor/table.ts
+++ b/packages/pg-meta/src/sql/studio/table-editor/table.ts
@@ -31,10 +31,320 @@ export const getDuplicateRowsSQL = ({
   return safeSql`INSERT INTO ${ident(sourceTableSchema)}.${ident(duplicatedTableName)} SELECT * FROM ${ident(sourceTableSchema)}.${ident(sourceTableName)};`
 }
 
-export const getTableEditorSql = ({ id }: { id?: number }): SafeSqlFragment => {
+export const getTableEditorSql = ({
+  id,
+  scoped = false,
+}: {
+  id?: number
+  scoped?: boolean
+}): SafeSqlFragment => {
   if (!id) return safeSql``
 
-  return safeSql`
+  return scoped
+    ? safeSql`
+    with base_table_info as (
+        select
+            c.oid::int8 as id,
+            nc.nspname as schema,
+            c.relname as name,
+            c.relkind,
+            c.relrowsecurity as rls_enabled,
+            c.relforcerowsecurity as rls_forced,
+            c.relreplident,
+            c.relowner,
+            obj_description(c.oid) as comment,
+            fs.srvname as foreign_server_name,
+            fdw.fdwname as foreign_data_wrapper_name,
+            fdw_handler.proname as foreign_data_wrapper_handler
+        from pg_class c
+        join pg_namespace nc on nc.oid = c.relnamespace
+        left join pg_foreign_table ft on ft.ftrelid = c.oid
+        left join pg_foreign_server fs on fs.oid = ft.ftserver
+        left join pg_foreign_data_wrapper fdw on fdw.oid = fs.srvfdw
+        left join pg_proc fdw_handler on fdw.fdwhandler = fdw_handler.oid
+        where c.oid = ${literal(id)}
+            and not pg_is_other_temp_schema(nc.oid)
+            and (
+                pg_has_role(c.relowner, 'USAGE')
+                or has_table_privilege(
+                    c.oid,
+                    'SELECT, INSERT, UPDATE, DELETE, TRUNCATE, REFERENCES, TRIGGER'
+                )
+                or has_any_column_privilege(c.oid, 'SELECT, INSERT, UPDATE, REFERENCES')
+            )
+    ),
+    table_stats as (
+        select
+            b.id,
+            case
+                when b.relreplident = 'd' then 'DEFAULT'
+                when b.relreplident = 'i' then 'INDEX'
+                when b.relreplident = 'f' then 'FULL'
+                else 'NOTHING'
+            end as replica_identity,
+            pg_total_relation_size(format('%I.%I', b.schema, b.name))::int8 as bytes,
+            pg_size_pretty(pg_total_relation_size(format('%I.%I', b.schema, b.name))) as size,
+            pg_stat_get_live_tuples(b.id) as live_rows_estimate,
+            pg_stat_get_dead_tuples(b.id) as dead_rows_estimate
+        from base_table_info b
+        where b.relkind in ('r', 'p')
+    ),
+    primary_keys as (
+        select
+            i.indrelid as table_id,
+            jsonb_agg(
+                jsonb_build_object(
+                    'schema', n.nspname,
+                    'table_name', c.relname,
+                    'table_id', i.indrelid::int8,
+                    'name', a.attname
+                )
+                order by array_position(i.indkey, a.attnum)
+            ) as primary_keys
+        from pg_index i
+        join pg_class c on i.indrelid = c.oid
+        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
+            and i.indrelid = ${literal(id)}
+        group by i.indrelid
+    ),
+    index_cols as (
+        select
+            i.indrelid as table_id,
+            i.indkey,
+            array_agg(
+                a.attname
+                order by array_position(i.indkey, a.attnum)
+            ) as columns
+        from pg_index i
+        join pg_class c on i.indrelid = c.oid
+        join pg_attribute a on a.attrelid = c.oid
+            and a.attnum = any(i.indkey)
+        where i.indisunique
+            and i.indisprimary = false
+            and i.indrelid = ${literal(id)}
+        group by i.indrelid, i.indkey
+    ),
+    unique_indexes as (
+        select
+            ic.table_id,
+            jsonb_agg(
+                jsonb_build_object(
+                    'schema', n.nspname,
+                    'table_name', c.relname,
+                    'table_id', ic.table_id::int8,
+                    'columns', ic.columns
+                )
+            ) as unique_indexes
+        from index_cols ic
+        join pg_class c on c.oid = ic.table_id
+        join pg_namespace n on n.oid = c.relnamespace
+        group by ic.table_id
+    ),
+    relationships as (
+        select
+            c.conrelid as source_id,
+            c.confrelid as target_id,
+            jsonb_build_object(
+                'id', c.oid::int8,
+                'constraint_name', c.conname,
+                'deletion_action', c.confdeltype,
+                'update_action', c.confupdtype,
+                'source_schema', nsa.nspname,
+                'source_table_name', csa.relname,
+                'source_column_name', sa.attname,
+                'target_table_schema', nta.nspname,
+                'target_table_name', cta.relname,
+                'target_column_name', ta.attname
+            ) as rel_info
+        from pg_constraint c
+        join pg_class csa on c.conrelid = csa.oid
+        join pg_namespace nsa on csa.relnamespace = nsa.oid
+        join pg_attribute sa on (sa.attrelid = c.conrelid and sa.attnum = any(c.conkey))
+        join pg_class cta on c.confrelid = cta.oid
+        join pg_namespace nta on cta.relnamespace = nta.oid
+        join pg_attribute ta on (ta.attrelid = c.confrelid and ta.attnum = any(c.confkey))
+        where c.contype = 'f'
+            and (c.conrelid = ${literal(id)} or c.confrelid = ${literal(id)})
+    ),
+    columns as (
+        select
+            a.attrelid as table_id,
+            jsonb_agg(jsonb_build_object(
+                'id', (a.attrelid || '.' || a.attnum),
+                'table_id', c.oid::int8,
+                'schema', nc.nspname,
+                'table', c.relname,
+                'ordinal_position', a.attnum,
+                'name', a.attname,
+                'default_value', case
+                    when a.atthasdef then pg_get_expr(ad.adbin, ad.adrelid)
+                    else null
+                end,
+                'data_type', case
+                    when t.typtype = 'd' then
+                        case
+                            when bt.typelem <> 0::oid and bt.typlen = -1 then 'ARRAY'
+                            when nbt.nspname = 'pg_catalog' then format_type(t.typbasetype, null)
+                            else 'USER-DEFINED'
+                        end
+                    else
+                        case
+                            when t.typelem <> 0::oid and t.typlen = -1 then 'ARRAY'
+                            when nt.nspname = 'pg_catalog' then format_type(a.atttypid, null)
+                            else 'USER-DEFINED'
+                        end
+                end,
+                'format', coalesce(bt.typname, t.typname),
+                'format_schema', coalesce(nbt.nspname, nt.nspname),
+                'is_identity', a.attidentity in ('a', 'd'),
+                'identity_generation', case a.attidentity
+                    when 'a' then 'ALWAYS'
+                    when 'd' then 'BY DEFAULT'
+                    else null
+                end,
+                'is_generated', a.attgenerated in ('s'),
+                'is_nullable', not (a.attnotnull or t.typtype = 'd' and t.typnotnull),
+                'is_updatable', (
+                    b.relkind in ('r', 'p') or
+                    (b.relkind in ('v', 'f') and pg_column_is_updatable(b.id, a.attnum, false))
+                ),
+                'is_unique', uniques.table_id is not null,
+                'check', check_constraints.definition,
+                'comment', col_description(c.oid, a.attnum),
+                'enums', coalesce(
+                    (
+                        select jsonb_agg(e.enumlabel order by e.enumsortorder)
+                        from pg_catalog.pg_enum e
+                        where e.enumtypid = coalesce(bt.oid, t.oid)
+                            or e.enumtypid = coalesce(bt.typelem, t.typelem)
+                    ),
+                    '[]'::jsonb
+                )
+            ) order by a.attnum) as columns
+        from pg_attribute a
+        join base_table_info b on a.attrelid = b.id
+        join pg_class c on a.attrelid = c.oid
+        join pg_namespace nc on c.relnamespace = nc.oid
+        left join pg_attrdef ad on (a.attrelid = ad.adrelid and a.attnum = ad.adnum)
+        join pg_type t on a.atttypid = t.oid
+        join pg_namespace nt on t.typnamespace = nt.oid
+        left join pg_type bt on (t.typtype = 'd' and t.typbasetype = bt.oid)
+        left join pg_namespace nbt on bt.typnamespace = nbt.oid
+        left join (
+            select
+                conrelid as table_id,
+                conkey[1] as ordinal_position
+            from pg_catalog.pg_constraint
+            where contype = 'u' and cardinality(conkey) = 1
+                and conrelid = ${literal(id)}
+            group by conrelid, conkey[1]
+        ) as uniques on uniques.table_id = a.attrelid and uniques.ordinal_position = a.attnum
+        left join (
+            select distinct on (conrelid, conkey[1])
+                conrelid as table_id,
+                conkey[1] as ordinal_position,
+                substring(
+                    pg_get_constraintdef(oid, true),
+                    8,
+                    length(pg_get_constraintdef(oid, true)) - 8
+                ) as definition
+            from pg_constraint
+            where contype = 'c' and cardinality(conkey) = 1
+                and conrelid = ${literal(id)}
+            order by conrelid, conkey[1], oid asc
+        ) as check_constraints on check_constraints.table_id = a.attrelid
+                            and check_constraints.ordinal_position = a.attnum
+        where a.attnum > 0
+        and not a.attisdropped
+        group by a.attrelid
+    )
+    select
+        case b.relkind
+            when 'r' then jsonb_build_object(
+                'entity_type', b.relkind,
+                'id', b.id,
+                'schema', b.schema,
+                'name', b.name,
+                'rls_enabled', b.rls_enabled,
+                'rls_forced', b.rls_forced,
+                'replica_identity', ts.replica_identity,
+                'bytes', ts.bytes,
+                'size', ts.size,
+                'live_rows_estimate', ts.live_rows_estimate,
+                'dead_rows_estimate', ts.dead_rows_estimate,
+                'comment', b.comment,
+                'primary_keys', coalesce(pk.primary_keys, '[]'::jsonb),
+                'unique_indexes', coalesce(ui.unique_indexes, '[]'::jsonb),
+                'relationships', coalesce(
+                    (select jsonb_agg(r.rel_info)
+                    from relationships r
+                    where r.source_id = b.id or r.target_id = b.id),
+                    '[]'::jsonb
+                ),
+                'columns', coalesce(c.columns, '[]'::jsonb)
+            )
+            when 'p' then jsonb_build_object(
+                'entity_type', b.relkind,
+                'id', b.id,
+                'schema', b.schema,
+                'name', b.name,
+                'rls_enabled', b.rls_enabled,
+                'rls_forced', b.rls_forced,
+                'replica_identity', ts.replica_identity,
+                'bytes', ts.bytes,
+                'size', ts.size,
+                'live_rows_estimate', ts.live_rows_estimate,
+                'dead_rows_estimate', ts.dead_rows_estimate,
+                'comment', b.comment,
+                'primary_keys', coalesce(pk.primary_keys, '[]'::jsonb),
+                'unique_indexes', coalesce(ui.unique_indexes, '[]'::jsonb),
+                'relationships', coalesce(
+                    (select jsonb_agg(r.rel_info)
+                    from relationships r
+                    where r.source_id = b.id or r.target_id = b.id),
+                    '[]'::jsonb
+                ),
+                'columns', coalesce(c.columns, '[]'::jsonb)
+            )
+            when 'v' then jsonb_build_object(
+                'entity_type', b.relkind,
+                'id', b.id,
+                'schema', b.schema,
+                'name', b.name,
+                'is_updatable', (pg_relation_is_updatable(b.id, false) & 20) = 20,
+                'comment', b.comment,
+                'columns', coalesce(c.columns, '[]'::jsonb)
+            )
+            when 'm' then jsonb_build_object(
+                'entity_type', b.relkind,
+                'id', b.id,
+                'schema', b.schema,
+                'name', b.name,
+                'is_populated', true,
+                'comment', b.comment,
+                'columns', coalesce(c.columns, '[]'::jsonb)
+            )
+            when 'f' then jsonb_build_object(
+                'entity_type', b.relkind,
+                'id', b.id,
+                'schema', b.schema,
+                'name', b.name,
+                'comment', b.comment,
+                'foreign_server_name', b.foreign_server_name,
+                'foreign_data_wrapper_name', b.foreign_data_wrapper_name,
+                'foreign_data_wrapper_handler', b.foreign_data_wrapper_handler,
+                'columns', coalesce(c.columns, '[]'::jsonb)
+            )
+        end as entity
+    from base_table_info b
+    left join table_stats ts on b.id = ts.id
+    left join primary_keys pk on b.id = pk.table_id
+    left join unique_indexes ui on b.id = ui.table_id
+    left join columns c on b.id = c.table_id;
+  `
+    : safeSql`
     with base_table_info as (
         select
             c.oid::int8 as id,
diff --git a/packages/pg-meta/test/db/plan-guard.ts b/packages/pg-meta/test/db/plan-guard.ts
new file mode 100644
index 00000000000..8c7ed44ac7b
--- /dev/null
+++ b/packages/pg-meta/test/db/plan-guard.ts
@@ -0,0 +1,180 @@
+/**
+ * Shared EXPLAIN plan-budget harness for hot-path Studio introspection queries.
+ *
+ * ── THE RULE ──────────────────────────────────────────────────────────────
+ * Every new introspection query added under `src/sql/` that is executed on a
+ * user's live catalog (Table Editor, Database pages, entity lists, definitions,
+ * ...) MUST get a budget entry in `test/sql/studio/catalog-plan-guard.test.ts`.
+ *
+ * Seq scans over catalogs that SCALE with schema size -- pg_class, pg_attribute,
+ * pg_index, pg_constraint, pg_attrdef, pg_description, pg_depend, pg_policy,
+ * pg_trigger, pg_rewrite, ... -- are only acceptable with a written STRUCTURAL
+ * justification (e.g. "no index exists on pg_constraint.confrelid"). An unscoped
+ * O(catalog) scan with no such justification is a bug: a production catalog had
+ * ~267K pg_class rows and unscoped CTEs turned a single Table Editor open into
+ * 30-58s of seq scans. Scope your query to the requested OID/schema instead.
+ *
+ * Tiny, fixed-size system catalogs (see TINY_NON_SCALING_CATALOGS) are always
+ * tolerated: the planner full-scans them because they hold a handful of rows and
+ * do not grow with the number of tables.
+ * ────────────────────────────────────────────────────────────────────────────
+ */
+
+import type { createTestDatabase } from './utils'
+
+type TestDatabase = Awaited<ReturnType<typeof createTestDatabase>>
+
+/**
+ * Tiny, fixed-size system catalogs (a handful of schemas/FDWs/enums/procs) that
+ * PostgreSQL's planner will always choose to seq scan regardless of query
+ * scoping, because a full scan of a handful of rows is cheaper than an index
+ * scan. These don't grow with table count and are unrelated to the O(catalog)
+ * regression this harness guards against, which is specifically about catalogs
+ * that scale with the number of tables/columns/indexes/constraints (pg_class,
+ * pg_attribute, pg_index, pg_constraint, ...).
+ */
+export const TINY_NON_SCALING_CATALOGS = new Set([
+  'pg_namespace',
+  'pg_foreign_table',
+  'pg_foreign_server',
+  'pg_foreign_data_wrapper',
+  'pg_enum',
+  'pg_proc',
+])
+
+/**
+ * Recursively walk an EXPLAIN (FORMAT JSON) plan tree and collect the relation
+ * name of every `Seq Scan` node. Returns one entry per seq-scan node (so a
+ * relation scanned twice appears twice).
+ */
+export function collectSeqScans(
+  node: unknown,
+  out: Array<string | undefined> = []
+): Array<string | undefined> {
+  if (Array.isArray(node)) {
+    for (const item of node) collectSeqScans(item, out)
+  } else if (node !== null && typeof node === 'object') {
+    const obj = node as Record<string, unknown>
+    if (obj['Node Type'] === 'Seq Scan') {
+      out.push(obj['Relation Name'] as string | undefined)
+    }
+    for (const key of Object.keys(obj)) {
+      collectSeqScans(obj[key], out)
+    }
+  }
+  return out
+}
+
+export type ExplainResult = {
+  /** The top-level plan node (`{ Plan, "Execution Time", ... }`). */
+  plan: { Plan: unknown; 'Execution Time': number }
+  /** Relation name of every seq-scan node found in the plan (one per node). */
+  seqScans: Array<string | undefined>
+  /** Measured execution time in milliseconds. */
+  executionTimeMs: number
+}
+
+/**
+ * Run `EXPLAIN (ANALYZE, FORMAT JSON)` on a SQL fragment and return the plan
+ * together with the flattened list of seq-scanned relations and the measured
+ * execution time.
+ */
+export async function explainAnalyze(
+  db: TestDatabase,
+  sqlFragment: string
+): Promise<ExplainResult> {
+  const [row] = await db.executeQuery<Array<Record<string, any>>>(
+    `explain (analyze, format json) ${sqlFragment}`
+  )
+  const [plan] = row['QUERY PLAN'] as Array<{ Plan: unknown; 'Execution Time': number }>
+  return {
+    plan,
+    seqScans: collectSeqScans(plan.Plan),
+    executionTimeMs: plan['Execution Time'],
+  }
+}
+
+export type PlanBudget = {
+  /**
+   * Seq scans on scaling catalogs that are structurally unavoidable, keyed by
+   * relation name. `max` caps how many seq-scan nodes on that relation are
+   * allowed; `reason` is the written structural justification (shown on
+   * failure). Prefix a reason with `KNOWN ISSUE:` to flag a suspected unscoped
+   * scan that still needs a follow-up fix.
+   */
+  allowedSeqScans?: Record<string, { max: number; reason: string }>
+  /** Upper bound on measured execution time, in ms. Defaults to 1000. */
+  maxExecutionTimeMs?: number
+}
+
+/**
+ * Assert an EXPLAIN result stays within its plan budget:
+ *   - tiny non-scaling catalogs are always tolerated,
+ *   - every other seq scan must match an `allowedSeqScans` entry and stay at or
+ *     under its `max`,
+ *   - execution time must stay under `maxExecutionTimeMs` (default 1000ms).
+ *
+ * Failures are actionable: they name the offending relations, dump the full
+ * seq-scan list, and restate the rule so the developer knows to scope the query
+ * or add a justified budget entry.
+ *
+ * Throws an `Error` describing the first violation; the caller (a vitest `test`)
+ * surfaces it as a failed assertion.
+ */
+export function assertPlanWithinBudget(result: ExplainResult, budget: PlanBudget = {}): void {
+  const allowed = budget.allowedSeqScans ?? {}
+  const maxExecutionTimeMs = budget.maxExecutionTimeMs ?? 1000
+
+  const seqScans = result.seqScans
+  const seqScanList = JSON.stringify(seqScans)
+
+  // Count seq-scan nodes per relation, ignoring tolerated tiny catalogs.
+  const counts = new Map<string, number>()
+  const offending: string[] = []
+  for (const relation of seqScans) {
+    if (!relation) {
+      offending.push('(unknown relation)')
+      continue
+    }
+    if (TINY_NON_SCALING_CATALOGS.has(relation)) continue
+    if (!(relation in allowed)) {
+      offending.push(relation)
+      continue
+    }
+    counts.set(relation, (counts.get(relation) ?? 0) + 1)
+  }
+
+  if (offending.length > 0) {
+    throw new Error(
+      `Unexpected seq scan(s) on scaling catalog(s): ${offending.join(', ')}.\n` +
+        `RULE: scope your query to the requested OID/schema so it uses an index, ` +
+        `or -- if the scan is structurally unavoidable -- add a justified budget ` +
+        `entry to allowedSeqScans with a written reason.\n` +
+        `Tolerated tiny non-scaling catalogs: ${[...TINY_NON_SCALING_CATALOGS].join(', ')}.\n` +
+        `Declared allowedSeqScans: ${Object.keys(allowed).join(', ') || '(none)'}.\n` +
+        `All seq scans in plan: ${seqScanList}`
+    )
+  }
+
+  for (const [relation, { max, reason }] of Object.entries(allowed)) {
+    const found = counts.get(relation) ?? 0
+    if (found > max) {
+      throw new Error(
+        `Too many seq-scan nodes on ${relation}: found ${found}, budget allows ${max}.\n` +
+          `Reason on file: "${reason}".\n` +
+          `A higher count usually means a new unscoped scan crept in -- scope it or ` +
+          `raise the budget with justification.\n` +
+          `All seq scans in plan: ${seqScanList}`
+      )
+    }
+  }
+
+  if (result.executionTimeMs >= maxExecutionTimeMs) {
+    throw new Error(
+      `Query took ${result.executionTimeMs.toFixed(1)}ms, budget is ${maxExecutionTimeMs}ms. ` +
+        `At stress scale this signals O(catalog) work -- scope the query to the ` +
+        `requested OID/schema.\n` +
+        `All seq scans in plan: ${seqScanList}`
+    )
+  }
+}
diff --git a/packages/pg-meta/test/db/stress-catalog.ts b/packages/pg-meta/test/db/stress-catalog.ts
new file mode 100644
index 00000000000..cd622c13c74
--- /dev/null
+++ b/packages/pg-meta/test/db/stress-catalog.ts
@@ -0,0 +1,80 @@
+/**
+ * Synthetic large-catalog builder shared by the catalog plan-guard tests.
+ *
+ * Context: a real production catalog had ~267K pg_class rows. Unscoped catalog
+ * scans in Studio introspection queries turned into O(catalog) seq scans on
+ * every dashboard open. To guard against that we build a large synthetic
+ * catalog in a `stress` schema and assert query plans stay scoped at scale.
+ *
+ * `TABLE_COUNT` defaults to 2000 to keep CI fast. Crank `PG_META_STRESS_TABLES`
+ * up (e.g. 12000+) for local investigation closer to real incident scale.
+ */
+
+import type { createTestDatabase } from './utils'
+
+type TestDatabase = Awaited<ReturnType<typeof createTestDatabase>>
+
+/** Default number of tables to build; override with PG_META_STRESS_TABLES. */
+export const DEFAULT_STRESS_TABLE_COUNT = 2000
+
+/** Resolved table count, honoring the PG_META_STRESS_TABLES env override. */
+export const STRESS_TABLE_COUNT = Number(
+  process.env.PG_META_STRESS_TABLES ?? DEFAULT_STRESS_TABLE_COUNT
+)
+
+/**
+ * Build a synthetic `stress` schema with `tableCount` tables plus a handful of
+ * non-table relations (a view, a materialized view, and a partitioned table
+ * with one partition) so relkind-filtering queries have something to hit.
+ *
+ * Per table: a PK index, a unique constraint+index, a check constraint, and an
+ * FK to the previous table. Every 10th table also FKs to t_0, making t_0 a hub
+ * with many incoming FKs (the structurally unavoidable pg_constraint seq scan,
+ * since there is no index on pg_constraint.confrelid).
+ *
+ * The catalog is built via a server-side procedure with batched commits every
+ * 100 tables -- a single transaction creating thousands of tables/indexes/
+ * constraints would exhaust the lock table. `analyze` runs at the end so the
+ * planner has real statistics.
+ */
+export async function buildStressCatalog(
+  db: TestDatabase,
+  tableCount: number = STRESS_TABLE_COUNT
+): Promise<void> {
+  await db.executeQuery(`
+    create schema stress;
+
+    create procedure stress.build(n int) language plpgsql as $$
+    begin
+      for i in 0..n-1 loop
+        execute format(
+          'create table stress.t_%s (id int primary key, u int unique, c int check (c > 0)%s%s)',
+          i,
+          case when i > 0 then format(', fk int references stress.t_%s(id)', i - 1) else '' end,
+          case when i > 0 and i % 10 = 0 then ', hub int references stress.t_0(id)' else '' end
+        );
+        if i % 100 = 99 then commit; end if;
+      end loop;
+    end $$;
+  `)
+
+  // `call` performs commits internally, so it must run as its own statement: a
+  // pg client query is autocommit by default, but bundling it with other
+  // statements in one multi-statement message implicitly wraps the whole
+  // message in one transaction, which conflicts with the commits inside the
+  // procedure's loop.
+  await db.executeQuery(`call stress.build(${tableCount});`)
+
+  // A view, a materialized view, and a partitioned table (with one partition)
+  // so queries that filter by relkind (views, matviews, partitioned tables)
+  // exercise a non-empty result at scale.
+  await db.executeQuery(`
+    create view stress.v_hub as select id, u, c from stress.t_0;
+    create materialized view stress.mv_hub as select id, u, c from stress.t_0;
+    create table stress.p_root (id int not null, region text not null, primary key (id, region))
+      partition by list (region);
+    create table stress.p_root_east partition of stress.p_root for values in ('east');
+  `)
+
+  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
new file mode 100644
index 00000000000..86dcb468539
--- /dev/null
+++ b/packages/pg-meta/test/sql/studio/catalog-plan-guard.test.ts
@@ -0,0 +1,289 @@
+import { afterAll, beforeAll, expect, test } from 'vitest'
+
+import {
+  createPgGetTabledefSql,
+  getEntityDefinitionsSql,
+  getEntityTypesSQL,
+  getForeignKeyConstraintsSql,
+  getIndexesSQL,
+  getTableColumnsSql,
+  getTableConstraintsSql,
+  getTableDefinitionSql,
+  getTableEditorSql,
+  getTablesPaginatedSql,
+  getViewDefinitionSql,
+} from '../../../src'
+import { assertPlanWithinBudget, explainAnalyze } from '../../db/plan-guard'
+import { buildStressCatalog, STRESS_TABLE_COUNT } from '../../db/stress-catalog'
+import { cleanupRoot, createTestDatabase } from '../../db/utils'
+
+/**
+ * Catalog query plan guard for hot-path Studio introspection queries.
+ *
+ * Builds a large synthetic catalog ONCE (see test/db/stress-catalog.ts) and, for
+ * each covered query, asserts via EXPLAIN (ANALYZE, FORMAT JSON) that the plan
+ * stays scoped: no seq scans over catalogs that scale with schema size, aside
+ * from tiny non-scaling catalogs (always tolerated) and structurally
+ * unavoidable scans carrying a written justification (see test/db/plan-guard.ts
+ * for THE RULE).
+ *
+ * Context: a real production catalog had ~267K pg_class rows. Before
+ * https://github.com/supabase/supabase/pull/47894, unscoped CTEs in the Table
+ * Editor query full-scanned pg_index/pg_constraint on every open -- O(catalog)
+ * work regardless of which table was opened, taking 30-58s at incident scale.
+ *
+ * NOTE: the scoped introspection behavior from PR #47894 defaults to OFF and is
+ * gated behind a `scoped` flag so Studio can enable it progressively via a
+ * feature flag. The legacy (scoped:false) path is intentionally NOT exercised
+ * here -- this guard tests the SCOPED path, so every builder below is called
+ * with `scoped: true`. Once the rollout completes, drop the legacy path and the
+ * flag, and this guard becomes the only shape.
+ */
+
+let db: Awaited<ReturnType<typeof createTestDatabase>>
+let midChainTableId: number
+let hubTableId: number
+let viewId: number
+
+beforeAll(async () => {
+  db = await createTestDatabase()
+  await buildStressCatalog(db, STRESS_TABLE_COUNT)
+
+  const [{ id: midChainId }] = await db.executeQuery<{ id: number }[]>(
+    `select 'stress.t_1000'::regclass::oid::int8 as id;`
+  )
+  const [{ id: hubId }] = await db.executeQuery<{ id: number }[]>(
+    `select 'stress.t_0'::regclass::oid::int8 as id;`
+  )
+  const [{ id: vId }] = await db.executeQuery<{ id: number }[]>(
+    `select 'stress.v_hub'::regclass::oid::int8 as id;`
+  )
+  midChainTableId = midChainId
+  hubTableId = hubId
+  viewId = vId
+}, 120_000)
+
+afterAll(async () => {
+  await db.cleanup()
+  await cleanupRoot()
+})
+
+// ── getTableEditorSql (table-editor/table.ts) — per-table ────────────────────
+// The only scaling-catalog seq scan that's structurally unavoidable: there is
+// no index on pg_constraint.confrelid, so the incoming-FK half of the
+// `relationships` CTE always does one filtered seq scan of pg_constraint.
+const TABLE_EDITOR_BUDGET = {
+  allowedSeqScans: {
+    pg_constraint: {
+      max: 2,
+      reason: 'no index on pg_constraint.confrelid -- incoming-FK half of relationships CTE',
+    },
+  },
+}
+
+test('getTableEditorSql: plan stays scoped for a mid-chain table (stress.t_1000)', async () => {
+  const result = await explainAnalyze(db, getTableEditorSql({ id: midChainTableId, scoped: true }))
+  assertPlanWithinBudget(result, TABLE_EDITOR_BUDGET)
+}, 60_000)
+
+test('getTableEditorSql: plan stays scoped for the hub table with many incoming FKs (stress.t_0)', async () => {
+  const result = await explainAnalyze(db, getTableEditorSql({ id: hubTableId, scoped: true }))
+  assertPlanWithinBudget(result, TABLE_EDITOR_BUDGET)
+}, 60_000)
+
+test('getTableEditorSql: real (non-EXPLAIN) query returns a well-formed entity for t_1000', async () => {
+  const [{ entity }] = await db.executeQuery<Array<{ entity: any }>>(
+    getTableEditorSql({ id: midChainTableId, scoped: true })
+  )
+
+  expect(entity.schema).toBe('stress')
+  expect(entity.name).toBe('t_1000')
+  expect(entity.primary_keys.length).toBeGreaterThan(0)
+  expect(entity.relationships.length).toBeGreaterThan(0)
+  expect(
+    entity.relationships.some(
+      (r: any) => r.source_table_name === 't_1000' && r.target_table_name === 't_999'
+    )
+  ).toBe(true)
+  expect(
+    entity.relationships.some(
+      (r: any) => r.source_table_name === 't_1001' && r.target_table_name === 't_1000'
+    )
+  ).toBe(true)
+}, 60_000)
+
+// ── getTableConstraintsSql (table-editor/constraints.ts) — per-table ─────────
+test('getTableConstraintsSql: plan stays scoped for a single table', async () => {
+  const result = await explainAnalyze(db, getTableConstraintsSql({ id: midChainTableId }))
+  assertPlanWithinBudget(result, {})
+}, 60_000)
+
+// ── getForeignKeyConstraintsSql (table-editor/foreign-keys.ts) — per-schema ──
+// A schema-wide FK listing must scan every FK constraint (no index on
+// pg_constraint.contype) and join pg_class (no index on relnamespace alone).
+test('getForeignKeyConstraintsSql: plan stays scoped for a schema', async () => {
+  const result = await explainAnalyze(db, getForeignKeyConstraintsSql({ schema: 'stress' }))
+  assertPlanWithinBudget(result, {
+    allowedSeqScans: {
+      pg_constraint: {
+        max: 1,
+        reason: "per-schema FK listing filters pg_constraint.contype='f'; no index on contype",
+      },
+      pg_class: {
+        max: 1,
+        reason:
+          'per-schema listing; no index on pg_class.relnamespace alone, so the schema filter cannot prune the scan',
+      },
+    },
+  })
+}, 60_000)
+
+// ── getEntityTypesSQL (table-editor/entities.ts) — per-schema list ───────────
+test('getEntityTypesSQL: plan stays scoped for a schema listing', async () => {
+  const result = await explainAnalyze(
+    db,
+    getEntityTypesSQL({
+      schemas: ['stress'],
+      sort: 'alphabetical',
+      filterTypes: ['r', 'v', 'm', 'f', 'p'],
+      limit: 100,
+      page: 0,
+    })
+  )
+  assertPlanWithinBudget(result, {})
+}, 60_000)
+
+// ── getTablesPaginatedSql (database/tables-paginated.ts) — per-schema page ───
+// The query IS scoped: the `page` CTE picks <=limit OIDs via pg_class_oid_index,
+// and every enrichment CTE constrains to `in (select oid from page)`. The
+// planner still full-scans the small enrichment catalogs once and hash-joins
+// them against the page set (cheaper than N index probes for a batch of OIDs).
+// These are structural for a list-with-enrichment query, not the unscoped
+// O(catalog) pattern (which had no page constraint at all).
+test('getTablesPaginatedSql: plan stays scoped for a schema page', async () => {
+  const result = await explainAnalyze(
+    db,
+    getTablesPaginatedSql({ schema: 'stress', limit: 100, afterOid: 0, includeColumns: true })
+  )
+  assertPlanWithinBudget(result, {
+    allowedSeqScans: {
+      pg_class: {
+        max: 1,
+        reason:
+          'page CTE + columns enrichment; no index on pg_class.relnamespace, and the planner batches the page-OID lookup as one scan',
+      },
+      pg_index: {
+        max: 1,
+        reason:
+          'primary-key CTE filters pg_index.indisprimary (unindexed), joined against the page set',
+      },
+      pg_constraint: {
+        max: 4,
+        reason:
+          'relationships CTE (two UNION arms), incl. confrelid which has no index; scanned once per arm and hash-joined against the page set',
+      },
+      pg_attribute: {
+        max: 3,
+        reason:
+          'primary-key/relationships/columns enrichment; scanned once and hash-joined against the page set rather than probed per OID',
+      },
+      pg_attrdef: {
+        max: 1,
+        reason: 'columns enrichment resolves defaults; scanned once against the page set',
+      },
+      pg_type: {
+        max: 1,
+        reason: 'columns enrichment resolves column types; scanned once against the page set',
+      },
+    },
+    // EXPLAIN ANALYZE inflates timing with per-node instrumentation on this
+    // multi-CTE query (~2.9s measured at 2000 tables vs a far lower real
+    // execution time). The seq-scan budget above is the primary guard here;
+    // the time bound is loosened to catch only gross regressions.
+    maxExecutionTimeMs: 6000,
+  })
+}, 60_000)
+
+// ── getTableColumnsSql (database/columns.ts) — per-table ─────────────────────
+test('getTableColumnsSql: plan stays scoped for a single table', async () => {
+  const result = await explainAnalyze(db, getTableColumnsSql({ table: 't_1000', schema: 'stress' }))
+  assertPlanWithinBudget(result, {})
+}, 60_000)
+
+// ── getIndexesSQL (database/indexes.ts) — per-schema list ────────────────────
+// A schema-wide index listing starts from pg_index (no index on the
+// schema-side columns), joins pg_class twice (table + index relations) and
+// pg_attribute for column names -- all scanned once and filtered by namespace.
+test('getIndexesSQL: plan stays scoped for a schema', async () => {
+  const result = await explainAnalyze(db, getIndexesSQL({ schema: 'stress' }))
+  assertPlanWithinBudget(result, {
+    allowedSeqScans: {
+      pg_index: {
+        max: 1,
+        reason:
+          'per-schema index listing enumerates all indexes; namespace filter is on the joined pg_class',
+      },
+      pg_class: {
+        max: 2,
+        reason:
+          'joins the table and index relations; no index on pg_class.relnamespace to prune by schema',
+      },
+      pg_attribute: {
+        max: 1,
+        reason: 'resolves index column names; scanned once and hash-joined',
+      },
+    },
+  })
+}, 60_000)
+
+// The two *DefinitionSql builders prepend a CREATE (pg_temp) function block
+// (CREATE_PG_GET_TABLEDEF_SQL) that cannot be wrapped in EXPLAIN. Run that
+// setup on the (single, pooled) connection first, then EXPLAIN only the tail
+// SELECT -- which is the part whose plan we actually care about.
+async function explainDefinitionQuery(fullSql: string) {
+  const setup = createPgGetTabledefSql({ scoped: true }) as unknown as string
+  const tail = fullSql.slice(fullSql.indexOf(setup) + setup.length)
+  await db.executeQuery(setup)
+  return explainAnalyze(db, tail)
+}
+
+// ── getTableDefinitionSql (database/table-definition.ts) — per-table ─────────
+// Only the outer table_info lookup is planned (pg_get_tabledef is an opaque
+// plpgsql function); it resolves the table by oid, so no scaling seq scans.
+test('getTableDefinitionSql: plan stays scoped for a single table', async () => {
+  const result = await explainDefinitionQuery(
+    getTableDefinitionSql({ id: midChainTableId, scoped: true })
+  )
+  assertPlanWithinBudget(result, {})
+}, 60_000)
+
+// ── getEntityDefinitionsSql (database/table-definition.ts) — per-schema list ─
+// The SQL-level plan is clean (only pg_namespace); the real cost is the opaque
+// pg_temp.pg_get_tabledef plpgsql function, which reconstructs full DDL per
+// entity via its own catalog introspection that the plan guard cannot see
+// inside. That function used to scan all of information_schema.columns once
+// PER COLUMN (and information_schema.tables once per call) just to decide
+// whether a name needs double-quoting — ~5.6s for a 25-entity page at 2000
+// tables, growing superlinearly with catalog size (~3.7s for a SINGLE entity
+// at 12K tables). Those scans were replaced with direct string tests, so the
+// whole page now runs in well under a second; the time bound below is the
+// guard against that per-entity cost regressing, since seq-scan detection is
+// blind inside the function. The production default limit is 100; 100 entities
+// on a 12K-table catalog measured ~0.9s post-fix.
+test('getEntityDefinitionsSql: plan stays scoped for a schema', async () => {
+  const result = await explainDefinitionQuery(
+    getEntityDefinitionsSql({ schemas: ['stress'], limit: 100, scoped: true })
+  )
+  assertPlanWithinBudget(result, {
+    // ~0.3s measured at 2000 tables post-fix for a 100-entity page; loose
+    // enough for slow CI, tight enough to catch reintroduced O(catalog)
+    // work inside pg_get_tabledef (which measured 5.6s for just 25 entities).
+    maxExecutionTimeMs: 3_000,
+  })
+}, 60_000)
+
+// ── getViewDefinitionSql (database/views.ts) — per-view ──────────────────────
+test('getViewDefinitionSql: plan stays scoped for a single view', async () => {
+  const result = await explainAnalyze(db, getViewDefinitionSql({ id: viewId }))
+  assertPlanWithinBudget(result, {})
+}, 60_000)
diff --git a/packages/pg-meta/test/sql/studio/table-editor.test.ts b/packages/pg-meta/test/sql/studio/table-editor.test.ts
new file mode 100644
index 00000000000..3c11fecf3b4
--- /dev/null
+++ b/packages/pg-meta/test/sql/studio/table-editor.test.ts
@@ -0,0 +1,132 @@
+import { afterAll, expect, test } from 'vitest'
+
+import { getTableEditorSql } from '../../../src'
+import { cleanupRoot, createTestDatabase } from '../../db/utils'
+
+afterAll(async () => {
+  await cleanupRoot()
+})
+
+type Entity = {
+  entity_type: string
+  id: number
+  schema: string
+  name: string
+  primary_keys: Array<{ table_id: number; schema: string; table_name: string; name: string }>
+  unique_indexes: Array<{ table_id: number; schema: string; table_name: string; columns: string[] }>
+  relationships: Array<{
+    id: number
+    constraint_name: string
+    source_schema: string
+    source_table_name: string
+    source_column_name: string
+    target_table_schema: string
+    target_table_name: string
+    target_column_name: string
+  }>
+  columns: Array<{
+    name: string
+    is_unique: boolean
+    check: string | null
+    comment: string | null
+  }>
+}
+
+const withTestDatabase = (
+  name: string,
+  fn: (db: Awaited<ReturnType<typeof createTestDatabase>>) => Promise<void>
+) => {
+  test(name, async () => {
+    const db = await createTestDatabase()
+    try {
+      await fn(db)
+    } finally {
+      await db.cleanup()
+    }
+  })
+}
+
+// The `scoped` flag gates the PR #47894 scoping predicates (default OFF for
+// progressive rollout). Both code paths must return semantically identical
+// entities; running the same assertions for scoped:true and scoped:false is the
+// CI equivalence guard. Once the rollout completes, drop scoped:false here.
+for (const scoped of [true, false]) {
+  withTestDatabase(
+    `scopes primary keys, unique indexes, relationships and columns to the target table (scoped=${scoped})`,
+    async ({ executeQuery }) => {
+      await executeQuery(`
+      create schema if not exists editor_scope;
+
+      create table editor_scope.authors (
+        id serial primary key
+      );
+
+      create table editor_scope.books (
+        id serial primary key,
+        isbn text not null unique,
+        price numeric not null check (price > 0),
+        author_id int not null references editor_scope.authors (id)
+      );
+      comment on column editor_scope.books.isbn is 'International Standard Book Number';
+
+      create table editor_scope.reviews (
+        id serial primary key,
+        book_id int not null references editor_scope.books (id)
+      );
+    `)
+
+      const [{ id: booksId }] = await executeQuery<{ id: number }[]>(
+        `select 'editor_scope.books'::regclass::oid::int8 as id;`
+      )
+
+      const sql = getTableEditorSql({ id: booksId, scoped })
+      const [{ entity }] = await executeQuery<{ entity: Entity }[]>(sql)
+
+      // Basic identity.
+      expect(entity.id).toBe(booksId)
+      expect(entity.schema).toBe('editor_scope')
+      expect(entity.name).toBe('books')
+
+      // Primary key scoped to `books`.
+      expect(entity.primary_keys).toEqual([
+        { schema: 'editor_scope', table_name: 'books', table_id: booksId, name: 'id' },
+      ])
+
+      // Unique index scoped to `books`.
+      expect(entity.unique_indexes).toHaveLength(1)
+      expect(entity.unique_indexes[0]).toMatchObject({
+        schema: 'editor_scope',
+        table_name: 'books',
+        table_id: booksId,
+        columns: ['isbn'],
+      })
+
+      // Relationships include both the outgoing FK (books -> authors) and the
+      // incoming FK (reviews -> books).
+      expect(entity.relationships).toContainEqual(
+        expect.objectContaining({
+          source_table_name: 'books',
+          source_column_name: 'author_id',
+          target_table_name: 'authors',
+          target_column_name: 'id',
+        })
+      )
+      expect(entity.relationships).toContainEqual(
+        expect.objectContaining({
+          source_table_name: 'reviews',
+          source_column_name: 'book_id',
+          target_table_name: 'books',
+          target_column_name: 'id',
+        })
+      )
+
+      // Columns: unique flag, check constraint definition, and comment.
+      const isbnCol = entity.columns.find((c) => c.name === 'isbn')!
+      expect(isbnCol.is_unique).toBe(true)
+      expect(isbnCol.comment).toBe('International Standard Book Number')
+
+      const priceCol = entity.columns.find((c) => c.name === 'price')!
+      expect(priceCol.check).toContain('price > 0')
+    }
+  )
+}