diff --git a/apps/docs/openapi.json b/apps/docs/openapi.json
index 49d6e7ed91b..7a258ebde0e 100644
--- a/apps/docs/openapi.json
+++ b/apps/docs/openapi.json
@@ -2602,7 +2602,7 @@
                     "properties": {
                       "name": {
                         "type": "string",
-                        "description": "Column name (alphanumeric and underscores, must start with letter or underscore)"
+                        "description": "Column display name. Any printable characters; no control characters or leading/trailing whitespace."
                       },
                       "type": {
                         "type": "string",
@@ -5577,9 +5577,8 @@
         "properties": {
           "name": {
             "type": "string",
-            "description": "Column name. Must start with a letter or underscore.",
-            "example": "email",
-            "pattern": "^[a-zA-Z_][a-zA-Z0-9_]*$"
+            "description": "Column display name. Any printable characters; no control characters or leading/trailing whitespace.",
+            "example": "email"
           },
           "type": {
             "type": "string",
diff --git a/apps/sim/app/api/table/[tableId]/columns/route.ts b/apps/sim/app/api/table/[tableId]/columns/route.ts
index 7eecb5ee466..68e50b66f2b 100644
--- a/apps/sim/app/api/table/[tableId]/columns/route.ts
+++ b/apps/sim/app/api/table/[tableId]/columns/route.ts
@@ -68,7 +68,8 @@ export const POST = withRouteHandler(async (request: NextRequest, context: Colum
       msg.includes('already exists') ||
       msg.includes('maximum column') ||
       msg.includes('Invalid column') ||
-      msg.includes('exceeds maximum')
+      msg.includes('exceeds maximum') ||
+      msg.includes("conflicts with another column's id")
     ) {
       return NextResponse.json({ error: msg }, { status: 400 })
     }
@@ -161,6 +162,7 @@ export const PATCH = withRouteHandler(async (request: NextRequest, context: Colu
       msg.includes('Cannot set unique column') ||
       msg.includes('Invalid column') ||
       msg.includes('exceeds maximum') ||
+      msg.includes("conflicts with another column's id") ||
       msg.includes('incompatible') ||
       msg.includes('duplicate')
     ) {
diff --git a/apps/sim/app/api/table/[tableId]/import/route.ts b/apps/sim/app/api/table/[tableId]/import/route.ts
index 63ed87220fb..e040694be51 100644
--- a/apps/sim/app/api/table/[tableId]/import/route.ts
+++ b/apps/sim/app/api/table/[tableId]/import/route.ts
@@ -30,9 +30,10 @@ import {
   inferColumnType,
   markTableJobRunning,
   releaseJobClaim,
-  sanitizeName,
+  sanitizeColumnName,
   type TableDefinition,
   type TableSchema,
+  uniqueColumnName,
   validateMapping,
   wouldExceedRowLimit,
 } from '@/lib/table'
@@ -55,6 +56,30 @@ interface RouteParams {
   params: Promise<{ tableId: string }>
 }
 
+/**
+ * Classifies a thrown import error as caller-caused (→ 400 with the message)
+ * vs unexpected (→ 500, message hidden). The shared column/row services throw
+ * bare `Error`s, so this matches on their message markers — one list for every
+ * catch site in this route so a new marker can't be added to only some of
+ * them. Interim until those services throw typed errors.
+ */
+function isImportClientError(message: string): boolean {
+  return (
+    message.includes('row limit') ||
+    message.includes('Insufficient capacity') ||
+    message.includes('Schema validation') ||
+    message.includes('must be unique') ||
+    message.includes('Row size exceeds') ||
+    message.includes('already exists') ||
+    message.includes('Invalid column name') ||
+    message.includes('exceeds maximum length') ||
+    message.includes('maximum column limit') ||
+    message.includes("conflicts with another column's id") ||
+    message.includes('CSV file has no') ||
+    /^Row \d+:/.test(message)
+  )
+}
+
 export const POST = withRouteHandler(async (request: NextRequest, { params }: RouteParams) => {
   const requestId = generateRequestId()
   const { tableId } = tableIdParamsSchema.parse(await params)
@@ -211,14 +236,7 @@ export const POST = withRouteHandler(async (request: NextRequest, { params }: Ro
       const newColumns: TableSchema['columns'] = []
 
       for (const header of createColumns) {
-        const base = sanitizeName(header)
-        let columnName = base
-        let suffix = 2
-        while (usedNames.has(columnName.toLowerCase())) {
-          columnName = `${base}_${suffix}`
-          suffix++
-        }
-        usedNames.add(columnName.toLowerCase())
+        const columnName = uniqueColumnName(sanitizeColumnName(header), usedNames)
         const inferredType = inferColumnType(rows.map((r) => r[header]))
         // Pre-assign the id so the prospective schema (used to coerce rows) and
         // the persisted column (created in importAppendRows) share the same key.
@@ -334,15 +352,7 @@ export const POST = withRouteHandler(async (request: NextRequest, { params }: Ro
           createdColumns: additions.length,
           error: message,
         })
-        const isClientError =
-          message.includes('row limit') ||
-          message.includes('Insufficient capacity') ||
-          message.includes('Schema validation') ||
-          message.includes('must be unique') ||
-          message.includes('Row size exceeds') ||
-          message.includes('already exists') ||
-          message.includes('Invalid column name') ||
-          /^Row \d+:/.test(message)
+        const isClientError = isImportClientError(message)
         return NextResponse.json(
           {
             error: isClientError ? message : 'Failed to import CSV',
@@ -386,14 +396,7 @@ export const POST = withRouteHandler(async (request: NextRequest, { params }: Ro
       })
     } catch (err) {
       const message = toError(err).message
-      const isClientError =
-        message.includes('row limit') ||
-        message.includes('Schema validation') ||
-        message.includes('must be unique') ||
-        message.includes('Row size exceeds') ||
-        message.includes('already exists') ||
-        message.includes('Invalid column name') ||
-        /^Row \d+:/.test(message)
+      const isClientError = isImportClientError(message)
       if (isClientError) {
         return NextResponse.json({ error: message }, { status: 400 })
       }
@@ -405,10 +408,7 @@ export const POST = withRouteHandler(async (request: NextRequest, { params }: Ro
     const message = toError(error).message
     logger.error(`[${requestId}] CSV import into existing table failed:`, error)
 
-    const isClientError =
-      message.includes('CSV file has no') ||
-      message.includes('already exists') ||
-      message.includes('Invalid column name')
+    const isClientError = isImportClientError(message)
 
     return NextResponse.json(
       { error: isClientError ? message : 'Failed to import CSV' },
diff --git a/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/components/enrichments-sidebar/enrichment-config.tsx b/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/components/enrichments-sidebar/enrichment-config.tsx
index 23820321337..22a51e2eaa8 100644
--- a/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/components/enrichments-sidebar/enrichment-config.tsx
+++ b/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/components/enrichments-sidebar/enrichment-config.tsx
@@ -18,7 +18,7 @@ import { generateId } from '@sim/utils/id'
 import type { AddWorkflowGroupBodyInput } from '@/lib/api/contracts/tables'
 import type { ColumnDefinition, WorkflowGroup, WorkflowGroupOutput } from '@/lib/table'
 import { columnMatchesRef, getColumnId } from '@/lib/table/column-keys'
-import { deriveOutputColumnName } from '@/lib/table/column-naming'
+import { uniqueColumnName } from '@/lib/table/column-naming'
 import { FieldError } from '@/app/workspace/[workspaceId]/tables/[tableId]/components/sidebar-fields'
 import type { EnrichmentConfig as EnrichmentDef } from '@/enrichments/types'
 import {
@@ -110,11 +110,9 @@ export function EnrichmentConfig({
       }
       return seed
     }
-    const taken = new Set(allColumns.map((c) => c.name))
+    const taken = new Set(allColumns.map((c) => c.name.toLowerCase()))
     for (const o of enrichment.outputs) {
-      const colName = deriveOutputColumnName(o.name, taken)
-      taken.add(colName)
-      seed[o.id] = colName
+      seed[o.id] = uniqueColumnName(o.name, taken)
     }
     return seed
   })
@@ -192,13 +190,12 @@ export function EnrichmentConfig({
     }
 
     const groupId = generateId()
-    const taken = new Set(allColumns.map((c) => c.name))
+    const taken = new Set(allColumns.map((c) => c.name.toLowerCase()))
     const outputColumns: AddWorkflowGroupBodyInput['outputColumns'] = []
     const outputs: WorkflowGroupOutput[] = []
     for (const o of enrichment.outputs) {
       const desired = (outputNames[o.id] ?? '').trim() || o.name
-      const colName = deriveOutputColumnName(desired, taken)
-      taken.add(colName)
+      const colName = uniqueColumnName(desired, taken)
       outputColumns.push({
         name: colName,
         type: o.type,
diff --git a/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/components/table-grid/headers/column-header-menu.tsx b/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/components/table-grid/headers/column-header-menu.tsx
index 8d775e8b8ac..bbd7c6219a6 100644
--- a/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/components/table-grid/headers/column-header-menu.tsx
+++ b/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/components/table-grid/headers/column-header-menu.tsx
@@ -4,6 +4,7 @@ import React, { useCallback, useEffect, useRef, useState } from 'react'
 import { cn } from '@sim/emcn'
 import { ChevronDown } from '@sim/emcn/icons'
 import type { WorkflowGroup } from '@/lib/table'
+import { TABLE_LIMITS } from '@/lib/table/constants'
 import type { WorkflowMetadata } from '@/stores/workflows/registry/types'
 import { COL_WIDTH, SELECTION_TINT_BG } from '../constants'
 import type { ColumnSourceInfo, DisplayColumn } from '../types'
@@ -278,6 +279,7 @@ export const ColumnHeaderMenu = React.memo(function ColumnHeaderMenu({
             ref={renameInputRef}
             type='text'
             value={renameValue}
+            maxLength={TABLE_LIMITS.MAX_COLUMN_NAME_LENGTH}
             onChange={(e) => onRenameValueChange(e.target.value)}
             onKeyDown={(e) => {
               if (e.key === 'Enter') onRenameSubmit()
diff --git a/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/components/workflow-sidebar/workflow-sidebar.tsx b/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/components/workflow-sidebar/workflow-sidebar.tsx
index 8392000e51a..f7cc7bc0d40 100644
--- a/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/components/workflow-sidebar/workflow-sidebar.tsx
+++ b/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/components/workflow-sidebar/workflow-sidebar.tsx
@@ -685,7 +685,7 @@ export function WorkflowSidebarBody({
           toast.success(`Saved "${targetColumnName}"`)
         } else {
           // edit-group: full output diff with new-column derivation.
-          const taken = new Set(allColumns.map((c) => c.name))
+          const taken = new Set(allColumns.map((c) => c.name.toLowerCase()))
           const fullOutputs: WorkflowGroupOutput[] = []
           const newOutputColumns: NonNullable<UpdateWorkflowGroupBodyInput['newOutputColumns']> = []
           for (const o of orderedOutputs) {
@@ -696,7 +696,6 @@ export function WorkflowSidebarBody({
               fullOutputs.push(existingOut)
             } else {
               const colName = deriveOutputColumnName(o.path, taken)
-              taken.add(colName)
               fullOutputs.push({ blockId: o.blockId, path: o.path, columnName: colName })
               newOutputColumns.push({
                 name: colName,
@@ -723,12 +722,11 @@ export function WorkflowSidebarBody({
       } else {
         // Create path: brand-new group with auto-derived output column names.
         const groupId = generateId()
-        const taken = new Set(allColumns.map((c) => c.name))
+        const taken = new Set(allColumns.map((c) => c.name.toLowerCase()))
         const newOutputColumns: AddWorkflowGroupBodyInput['outputColumns'] = []
         const groupOutputs: WorkflowGroupOutput[] = []
         for (const o of orderedOutputs) {
           const colName = deriveOutputColumnName(o.path, taken)
-          taken.add(colName)
           newOutputColumns.push({
             name: colName,
             type: columnTypeForLeaf(o.leafType),
diff --git a/apps/sim/app/workspace/[workspaceId]/tables/components/import-csv-dialog/import-csv-dialog.tsx b/apps/sim/app/workspace/[workspaceId]/tables/components/import-csv-dialog/import-csv-dialog.tsx
index 2eee7a1c4ee..8962a954d08 100644
--- a/apps/sim/app/workspace/[workspaceId]/tables/components/import-csv-dialog/import-csv-dialog.tsx
+++ b/apps/sim/app/workspace/[workspaceId]/tables/components/import-csv-dialog/import-csv-dialog.tsx
@@ -46,15 +46,15 @@ const MAX_EXAMPLES_IN_ERROR = 3
 const CSV_PREVIEW_BYTES = 512 * 1024
 /**
  * Sentinel value for the "Do not import" option in the mapping combobox. The
- * whitespace is intentional: valid column names must match `NAME_PATTERN`
- * (`/^[a-z_][a-z0-9_]*$/i`), so no real column can share this value.
+ * embedded NUL is intentional: control characters are invalid in column names
+ * (`COLUMN_NAME_PATTERN`), so no real column can share this value.
  */
-const SKIP_VALUE = '__ skip __'
+const SKIP_VALUE = '\u0000skip'
 /**
- * Sentinel for the "Create new column" option. Same whitespace trick as
+ * Sentinel for the "Create new column" option. Same control-character trick as
  * `SKIP_VALUE` to avoid colliding with any valid column name.
  */
-const CREATE_VALUE = '__ create __'
+const CREATE_VALUE = '\u0000create'
 
 /**
  * Converts the verbose backend error messages into a short, human-friendly
diff --git a/apps/sim/lib/api/contracts/tables.test.ts b/apps/sim/lib/api/contracts/tables.test.ts
index 0f368363d3b..9d320154e20 100644
--- a/apps/sim/lib/api/contracts/tables.test.ts
+++ b/apps/sim/lib/api/contracts/tables.test.ts
@@ -2,7 +2,31 @@
  * @vitest-environment node
  */
 import { describe, expect, it } from 'vitest'
-import { tableEventStreamQuerySchema } from '@/lib/api/contracts/tables'
+import { tableColumnSchema, tableEventStreamQuerySchema } from '@/lib/api/contracts/tables'
+
+describe('tableColumnSchema', () => {
+  it('accepts display names with spaces, digits-first, punctuation, and unicode', () => {
+    for (const name of ['First Name', '2024 Revenue', 'price ($)', 'caf\u00e9']) {
+      expect(tableColumnSchema.safeParse({ name, type: 'string' }).success).toBe(true)
+    }
+  })
+
+  it('rejects invisible characters, edge whitespace, leading $, and the CSV-dialog sentinels', () => {
+    const bad = [' leading', 'trailing ', 'a\u0000b', 'zero\u200bwidth', '$or', '\u0000skip']
+    for (const name of bad) {
+      expect(tableColumnSchema.safeParse({ name, type: 'string' }).success).toBe(false)
+    }
+  })
+
+  it('accepts identifier-shaped column ids and rejects non-identifier ids', () => {
+    const good = { name: 'x', type: 'string' }
+    expect(tableColumnSchema.safeParse({ ...good, id: 'col_ab12' }).success).toBe(true)
+    expect(tableColumnSchema.safeParse({ ...good, id: 'legacy_name' }).success).toBe(true)
+    for (const id of ["a-b'", 'has space', '1leading', '']) {
+      expect(tableColumnSchema.safeParse({ ...good, id }).success).toBe(false)
+    }
+  })
+})
 
 describe('tableEventStreamQuerySchema', () => {
   it('parses an explicit cursor', () => {
diff --git a/apps/sim/lib/api/contracts/tables.ts b/apps/sim/lib/api/contracts/tables.ts
index c357cc585e8..fa335c14e8c 100644
--- a/apps/sim/lib/api/contracts/tables.ts
+++ b/apps/sim/lib/api/contracts/tables.ts
@@ -13,7 +13,13 @@ import type {
   TableRow,
   TableRowsCursor,
 } from '@/lib/table'
-import { COLUMN_TYPES, NAME_PATTERN, TABLE_LIMITS } from '@/lib/table/constants'
+import {
+  COLUMN_NAME_PATTERN,
+  COLUMN_NAME_RULE,
+  COLUMN_TYPES,
+  NAME_PATTERN,
+  TABLE_LIMITS,
+} from '@/lib/table/constants'
 import { CSV_MAX_FILE_SIZE_BYTES } from '@/lib/table/import'
 
 export const domainObjectSchema = <T>() => z.custom<T>(isRecordLike)
@@ -25,7 +31,7 @@ export const domainObjectSchema = <T>() => z.custom<T>(isRecordLike)
 export const columnTypeSchema = z.enum(COLUMN_TYPES)
 
 /**
- * Identifier for tables/columns: starts with letter or underscore, contains
+ * Identifier for tables: starts with letter or underscore, contains
  * only alphanumerics + underscores, capped at `MAX_TABLE_NAME_LENGTH`.
  */
 const tableNameSchema = z
@@ -47,9 +53,21 @@ const columnNameSchema = z
     TABLE_LIMITS.MAX_COLUMN_NAME_LENGTH,
     `Column name must be ${TABLE_LIMITS.MAX_COLUMN_NAME_LENGTH} characters or less`
   )
+  .regex(COLUMN_NAME_PATTERN, COLUMN_NAME_RULE)
+
+/**
+ * Stable column id — the JSONB row-data storage key, interpolated into SQL
+ * paths after a `NAME_PATTERN` guard. Enforce the identifier shape at the
+ * boundary so a caller-supplied id (undo re-create, schema copy) fails fast at
+ * write time instead of wedging every later filter/sort/unique-check.
+ */
+const columnIdSchema = z
+  .string()
+  .min(1, 'Column id cannot be empty')
+  .max(64, 'Column id must be 64 characters or less')
   .regex(
     NAME_PATTERN,
-    'Column name must start with a letter or underscore and contain only alphanumeric characters and underscores'
+    'Column id must start with a letter or underscore and contain only alphanumeric characters and underscores'
   )
 
 const descriptionSchema = z
@@ -80,7 +98,7 @@ export const getTableQuerySchema = z.object({
 
 export const tableColumnSchema = z.object({
   /** Stable column id (server-assigned). Absent on legacy/ pre-backfill columns. */
-  id: z.string().optional(),
+  id: columnIdSchema.optional(),
   name: columnNameSchema,
   type: columnTypeSchema,
   required: z.boolean().optional().default(false),
@@ -115,7 +133,7 @@ export const createTableColumnBodySchema = z.object({
   column: z.object({
     // Optional stable id — first-party undo of a delete re-creates the column
     // with its original id so saved (id-keyed) cell data restores correctly.
-    id: z.string().optional(),
+    id: columnIdSchema.optional(),
     name: columnNameSchema,
     type: columnTypeSchema,
     required: z.boolean().optional(),
@@ -1020,7 +1038,7 @@ const workflowGroupInputMappingSchema = z.object({
 })
 
 const workflowGroupOutputColumnSchema = z.object({
-  name: z.string().min(1),
+  name: columnNameSchema,
   type: columnTypeSchema,
   required: z.boolean().optional(),
   unique: z.boolean().optional(),
diff --git a/apps/sim/lib/copilot/tools/server/table/user-table.ts b/apps/sim/lib/copilot/tools/server/table/user-table.ts
index 540ea5f7d25..0e2d63f14cd 100644
--- a/apps/sim/lib/copilot/tools/server/table/user-table.ts
+++ b/apps/sim/lib/copilot/tools/server/table/user-table.ts
@@ -34,7 +34,11 @@ import {
   rowDataNameToId,
   sortNamesToIds,
 } from '@/lib/table/column-keys'
-import { columnTypeForLeaf, deriveOutputColumnName } from '@/lib/table/column-naming'
+import {
+  columnTypeForLeaf,
+  deriveOutputColumnName,
+  uniqueColumnName,
+} from '@/lib/table/column-naming'
 import {
   addTableColumn,
   deleteColumn,
@@ -1645,13 +1649,13 @@ export const userTableServerTool: BaseServerTool<UserTableArgs, UserTableResult>
             flattened.map((f) => [`${f.blockId}::${f.path}`, f.leafType])
           )
 
-          const taken = new Set(tableForGroup.schema.columns.map((c) => c.name))
+          const taken = new Set(tableForGroup.schema.columns.map((c) => c.name.toLowerCase()))
           const groupId = generateId()
           const outputs: WorkflowGroupOutput[] = []
           const outputColumns: ColumnDefinition[] = []
           for (const o of rawOutputs) {
             const colName = o.columnName ?? deriveOutputColumnName(o.path, taken)
-            taken.add(colName)
+            taken.add(colName.toLowerCase())
             outputs.push({ blockId: o.blockId, path: o.path, columnName: colName })
             const leafType = o.columnType ?? leafTypeByKey.get(`${o.blockId}::${o.path}`)
             outputColumns.push({
@@ -2009,14 +2013,13 @@ export const userTableServerTool: BaseServerTool<UserTableArgs, UserTableResult>
           // Each enrichment output becomes a new column. Names can be overridden
           // per output id; otherwise the enrichment's default name is used.
           const outputNameOverrides = (args.outputColumnNames ?? {}) as Record<string, string>
-          const taken = new Set(tableForEnrichment.schema.columns.map((c) => c.name))
+          const taken = new Set(tableForEnrichment.schema.columns.map((c) => c.name.toLowerCase()))
           const groupId = generateId()
           const outputs: WorkflowGroupOutput[] = []
           const outputColumns: ColumnDefinition[] = []
           for (const out of enrichment.outputs) {
             const desired = (outputNameOverrides[out.id] ?? '').trim() || out.name
-            const colName = deriveOutputColumnName(desired, taken)
-            taken.add(colName)
+            const colName = uniqueColumnName(desired, taken)
             outputs.push({ blockId: '', path: '', outputId: out.id, columnName: colName })
             outputColumns.push({
               name: colName,
diff --git a/apps/sim/lib/table/__tests__/column-keys.test.ts b/apps/sim/lib/table/__tests__/column-keys.test.ts
index 08832e09906..341079897cf 100644
--- a/apps/sim/lib/table/__tests__/column-keys.test.ts
+++ b/apps/sim/lib/table/__tests__/column-keys.test.ts
@@ -17,6 +17,7 @@ import {
   filterNamesToIds,
   generateColumnId,
   getColumnId,
+  nameShadowsColumnId,
   remapGroupColumnRefs,
   rowDataIdToName,
   rowDataNameToId,
@@ -55,8 +56,10 @@ describe('name ↔ id maps', () => {
     ],
   }
 
-  it('buildIdByName maps display name → storage id', () => {
+  it('buildIdByName maps lowercased display name → storage id', () => {
     expect(Object.fromEntries(buildIdByName(schema))).toEqual({ email: 'col_1', age: 'age' })
+    const mixed: TableSchema = { columns: [{ id: 'col_9', name: 'First Name', type: 'string' }] }
+    expect(Object.fromEntries(buildIdByName(mixed))).toEqual({ 'first name': 'col_9' })
   })
   it('buildNameById maps storage id → display name', () => {
     expect(Object.fromEntries(buildNameById(schema))).toEqual({ col_1: 'email', age: 'age' })
@@ -84,6 +87,10 @@ describe('row data translation', () => {
     expect(rowDataNameToId({ email: 'x', ghost: 1 }, idByName)).toEqual({ col_1: 'x' })
     expect(rowDataIdToName({ col_1: 'x', col_gone: 9 }, nameById)).toEqual({ email: 'x' })
   })
+
+  it('matches wire keys case-insensitively instead of silently dropping them', () => {
+    expect(rowDataNameToId({ EMAIL: 'x', Age: 1 }, idByName)).toEqual({ col_1: 'x', age: 1 })
+  })
 })
 
 describe('filter / sort translation', () => {
@@ -109,6 +116,29 @@ describe('filter / sort translation', () => {
       createdAt: 'desc',
     })
   })
+
+  it('matches filter/sort fields case-insensitively, preserving unknown keys verbatim', () => {
+    expect(filterNamesToIds({ EMAIL: 'x', Ghost: 1 }, idByName)).toEqual({ col_1: 'x', Ghost: 1 })
+    expect(sortNamesToIds({ Email: 'asc' }, idByName)).toEqual({ col_1: 'asc' })
+  })
+})
+
+describe('nameShadowsColumnId', () => {
+  const columns: TableSchema['columns'] = [
+    { id: 'col_abc', name: 'email', type: 'string' },
+    { name: 'age', type: 'number' }, // legacy: storage key is the name
+  ]
+
+  it('flags a name equal to another column storage key', () => {
+    expect(nameShadowsColumnId(columns, 'col_abc')).toBe(true)
+    expect(nameShadowsColumnId(columns, 'age')).toBe(true)
+    expect(nameShadowsColumnId(columns, 'First Name')).toBe(false)
+  })
+
+  it('excludes the column being renamed', () => {
+    expect(nameShadowsColumnId(columns, 'age', 1)).toBe(false)
+    expect(nameShadowsColumnId(columns, 'col_abc', 1)).toBe(true)
+  })
 })
 
 describe('withGeneratedColumnIds', () => {
diff --git a/apps/sim/lib/table/__tests__/sql.test.ts b/apps/sim/lib/table/__tests__/sql.test.ts
index ca293848779..f536214d19c 100644
--- a/apps/sim/lib/table/__tests__/sql.test.ts
+++ b/apps/sim/lib/table/__tests__/sql.test.ts
@@ -253,7 +253,7 @@ describe('SQL Builder', () => {
 
     it('throws on invalid field name', () => {
       expect(() => buildFilterClause({ 'invalid-field': 'v' }, TABLE, NO_COLUMNS)).toThrow(
-        'Invalid field name'
+        'Unknown field'
       )
     })
 
@@ -418,7 +418,7 @@ describe('SQL Builder', () => {
 
     it('throws on invalid field name', () => {
       const sort: Sort = { 'invalid-field': 'asc' }
-      expect(() => buildSortClause(sort, TABLE, NO_COLUMNS)).toThrow('Invalid field name')
+      expect(() => buildSortClause(sort, TABLE, NO_COLUMNS)).toThrow('Unknown field')
     })
 
     it('throws on invalid direction', () => {
@@ -437,25 +437,21 @@ describe('SQL Builder', () => {
 
     it('rejects identifiers starting with a digit', () => {
       expect(() => buildFilterClause({ '123name': 'v' }, TABLE, NO_COLUMNS)).toThrow(
-        'Invalid field name'
+        'Unknown field'
       )
     })
 
     it('rejects identifiers with special characters', () => {
       const invalid = ['field-name', 'field.name', 'field name', 'field@name']
       for (const name of invalid) {
-        expect(() => buildFilterClause({ [name]: 'v' }, TABLE, NO_COLUMNS)).toThrow(
-          'Invalid field name'
-        )
+        expect(() => buildFilterClause({ [name]: 'v' }, TABLE, NO_COLUMNS)).toThrow('Unknown field')
       }
     })
 
     it('rejects SQL injection attempts in field names', () => {
       const attempts = ["'; DROP TABLE users; --", 'name OR 1=1', 'name; DELETE FROM']
       for (const a of attempts) {
-        expect(() => buildFilterClause({ [a]: 'v' }, TABLE, NO_COLUMNS)).toThrow(
-          'Invalid field name'
-        )
+        expect(() => buildFilterClause({ [a]: 'v' }, TABLE, NO_COLUMNS)).toThrow('Unknown field')
       }
     })
   })
diff --git a/apps/sim/lib/table/__tests__/validation.test.ts b/apps/sim/lib/table/__tests__/validation.test.ts
index 40e3187270c..d0bcf393013 100644
--- a/apps/sim/lib/table/__tests__/validation.test.ts
+++ b/apps/sim/lib/table/__tests__/validation.test.ts
@@ -93,6 +93,42 @@ describe('Validation', () => {
       expect(result.errors).toContain('Column name is required')
     })
 
+    it('should accept names with spaces, digits, and punctuation', () => {
+      const names = ['First Name', '2024 Revenue', 'e-mail', 'price ($)', 'café', '№ of items']
+
+      for (const name of names) {
+        const result = validateColumnDefinition({ name, type: 'string' })
+        expect(result.valid).toBe(true)
+      }
+    })
+
+    it('should reject names with control or invisible characters or leading/trailing whitespace', () => {
+      const names = [
+        ' leading',
+        'trailing ',
+        'tab\tinside',
+        'new\nline',
+        'nul\u0000char',
+        'zero\u200bwidth',
+        '\u200b',
+        '\u202eevil',
+        'lone\ud800surrogate',
+      ]
+
+      for (const name of names) {
+        const result = validateColumnDefinition({ name, type: 'string' })
+        expect(result.valid).toBe(false)
+        expect(result.errors[0]).toContain('is invalid')
+      }
+    })
+
+    it('should reject names starting with "$" (filter-grammar collision)', () => {
+      for (const name of ['$or', '$and', '$ Amount']) {
+        expect(validateColumnDefinition({ name, type: 'string' }).valid).toBe(false)
+      }
+      expect(validateColumnDefinition({ name: 'Amount ($)', type: 'string' }).valid).toBe(true)
+    })
+
     it('should reject invalid column type', () => {
       const result = validateColumnDefinition({
         name: 'test',
diff --git a/apps/sim/lib/table/column-keys.ts b/apps/sim/lib/table/column-keys.ts
index ec770b047ea..c53c06d55d2 100644
--- a/apps/sim/lib/table/column-keys.ts
+++ b/apps/sim/lib/table/column-keys.ts
@@ -5,10 +5,12 @@
  * refs, and filter/sort all key on a column's stable **id**. `name` is a
  * display label that changes on rename. The two name-translating boundaries
  * (public v1 API, mothership tool) and CSV convert between the two with the
- * map builders here.
+ * map builders here. Also hosts the write-time display-name validator
+ * (`assertValidNewColumnName`) since it depends on the id/name semantics.
  */
 
 import { generateId } from '@sim/utils/id'
+import { COLUMN_NAME_PATTERN, COLUMN_NAME_RULE, TABLE_LIMITS } from '@/lib/table/constants'
 import type {
   ColumnDefinition,
   Filter,
@@ -51,6 +53,60 @@ export function columnMatchesRef(col: ColumnDefinition, ref: string): boolean {
   return getColumnId(col) === ref || col.name.toLowerCase() === ref.toLowerCase()
 }
 
+/**
+ * True when `name` equals some other column's storage key. Such a name would
+ * make `columnMatchesRef` and the name→id remaps ambiguous (a ref intended for
+ * the other column's id would resolve by name instead) — reject it at write
+ * time. `excludeIndex` skips the column being renamed.
+ */
+export function nameShadowsColumnId(
+  columns: readonly ColumnDefinition[],
+  name: string,
+  excludeIndex = -1
+): boolean {
+  return columns.some((c, i) => i !== excludeIndex && getColumnId(c) === name)
+}
+
+/**
+ * The single write-time validator for a column display name: pattern, length,
+ * case-insensitive uniqueness against `columns`, and id shadowing. Every
+ * column-creating or renaming service path calls this so the rule can never
+ * drift between them. The pattern/length rules mirror `validateColumnDefinition`
+ * (validation.ts), which accumulates errors for bulk schema checks — duplicated
+ * because validation.ts is server-only while this module is client-safe.
+ *
+ * `excludeIndex` skips the column being renamed. `extraTakenLower` holds
+ * LOWERCASED names claimed earlier in the same batch; an accepted name is
+ * claimed into it before returning, so batch callers never hand-maintain the
+ * set. Throws with a caller-facing message on the first violation.
+ */
+export function assertValidNewColumnName(
+  name: string,
+  columns: readonly ColumnDefinition[],
+  options: { excludeIndex?: number; extraTakenLower?: Set<string> } = {}
+): void {
+  const { excludeIndex = -1, extraTakenLower } = options
+  if (!COLUMN_NAME_PATTERN.test(name)) {
+    throw new Error(`Invalid column name "${name}". ${COLUMN_NAME_RULE}.`)
+  }
+  if (name.length > TABLE_LIMITS.MAX_COLUMN_NAME_LENGTH) {
+    throw new Error(
+      `Column name exceeds maximum length (${TABLE_LIMITS.MAX_COLUMN_NAME_LENGTH} characters)`
+    )
+  }
+  const lower = name.toLowerCase()
+  if (
+    columns.some((c, i) => i !== excludeIndex && c.name.toLowerCase() === lower) ||
+    extraTakenLower?.has(lower)
+  ) {
+    throw new Error(`Column "${name}" already exists`)
+  }
+  if (nameShadowsColumnId(columns, name, excludeIndex)) {
+    throw new Error(`Column name "${name}" conflicts with another column's id`)
+  }
+  extraTakenLower?.add(lower)
+}
+
 /**
  * Returns a schema copy with a generated id stamped onto every column that
  * lacks one, remapping any workflow-group refs that still hold a column **name**
@@ -119,13 +175,23 @@ export function remapGroupColumnRefs(
   }
 }
 
-/** `name → id` for translating inbound wire data (v1 / mothership / CSV import). */
+/**
+ * `lowercased name → id` for translating inbound wire data (v1 / mothership /
+ * CSV import). Keys are lowercased so lookups match the schema's
+ * case-insensitive name uniqueness — the same semantics as `columnMatchesRef` —
+ * instead of silently dropping a miscased key. Look up via `idForName`.
+ */
 export function buildIdByName(schema: TableSchema): Map<string, string> {
   const map = new Map<string, string>()
-  for (const col of schema.columns) map.set(col.name, getColumnId(col))
+  for (const col of schema.columns) map.set(col.name.toLowerCase(), getColumnId(col))
   return map
 }
 
+/** Case-insensitive lookup into a {@link buildIdByName} map. */
+function idForName(idByName: ReadonlyMap<string, string>, name: string): string | undefined {
+  return idByName.get(name.toLowerCase())
+}
+
 /** `id → name` for translating outbound wire data (v1 / mothership / CSV export). */
 export function buildNameById(schema: TableSchema): Map<string, string> {
   const map = new Map<string, string>()
@@ -135,13 +201,14 @@ export function buildNameById(schema: TableSchema): Map<string, string> {
 
 /**
  * Remaps a wire row keyed by column **name** to the stored **id** keying. Used
- * at the name-translating boundaries on the way in. Keys not matching a known
+ * at the name-translating boundaries on the way in. Names match
+ * case-insensitively (mirroring `columnMatchesRef`); keys not matching a known
  * column are dropped (validation has already run against the schema).
  */
 export function rowDataNameToId(data: RowData, idByName: Map<string, string>): RowData {
   const out: RowData = {}
   for (const [name, value] of Object.entries(data)) {
-    const id = idByName.get(name)
+    const id = idForName(idByName, name)
     if (id !== undefined) out[id] = value
   }
   return out
@@ -158,7 +225,7 @@ export function filterNamesToIds(filter: Filter, idByName: ReadonlyMap<string, s
     if ((key === '$or' || key === '$and') && Array.isArray(value)) {
       out[key] = (value as Filter[]).map((f) => filterNamesToIds(f, idByName))
     } else {
-      out[idByName.get(key) ?? key] = value
+      out[idForName(idByName, key) ?? key] = value
     }
   }
   return out
@@ -167,7 +234,7 @@ export function filterNamesToIds(filter: Filter, idByName: ReadonlyMap<string, s
 /** Translates a sort's field names → column ids. Unknown fields pass through. */
 export function sortNamesToIds(sort: Sort, idByName: ReadonlyMap<string, string>): Sort {
   const out: Sort = {}
-  for (const [field, dir] of Object.entries(sort)) out[idByName.get(field) ?? field] = dir
+  for (const [field, dir] of Object.entries(sort)) out[idForName(idByName, field) ?? field] = dir
   return out
 }
 
diff --git a/apps/sim/lib/table/column-naming.ts b/apps/sim/lib/table/column-naming.ts
index 1124e8da2df..24c7e6a646c 100644
--- a/apps/sim/lib/table/column-naming.ts
+++ b/apps/sim/lib/table/column-naming.ts
@@ -1,13 +1,36 @@
 /**
- * Shared column-naming helpers used by every path that auto-derives a
- * column name + type from a workflow block output: the table column-sidebar
- * UI, and the Copilot/Mothership `add_workflow_group` op. Keeping one
- * implementation means the AI's auto-named columns match what a user would
- * get from the sidebar.
+ * Shared column-naming helpers: collision-free name claiming (CSV/JSON import,
+ * enrichment output naming) and block-output auto-derivation (the table
+ * column-sidebar UI and the Copilot/Mothership `add_workflow_group` op).
+ * Keeping one implementation means every path — importer, AI, or sidebar —
+ * names columns the same way.
  */
 
+import { TABLE_LIMITS } from '@/lib/table/constants'
 import type { ColumnDefinition } from '@/lib/table/types'
 
+/**
+ * Claims a column name not present in `takenLower` (a set of LOWERCASED names,
+ * matching the schema's case-insensitive uniqueness) by appending `_2`, `_3`, …
+ * to `base` on collision. The base is trimmed so the suffixed result stays
+ * within `MAX_COLUMN_NAME_LENGTH`. The chosen name is added (lowercased) to
+ * `takenLower` before returning, so callers never hand-maintain the set's
+ * casing convention.
+ */
+export function uniqueColumnName(base: string, takenLower: Set<string>): string {
+  const claim = (name: string): string => {
+    takenLower.add(name.toLowerCase())
+    return name
+  }
+  if (!takenLower.has(base.toLowerCase())) return claim(base)
+  for (let suffix = 2; ; suffix++) {
+    const tail = `_${suffix}`
+    const head = base.slice(0, TABLE_LIMITS.MAX_COLUMN_NAME_LENGTH - tail.length).trimEnd()
+    const candidate = `${head}${tail}`
+    if (!takenLower.has(candidate.toLowerCase())) return claim(candidate)
+  }
+}
+
 /**
  * Slugifies a string into a `NAME_PATTERN`-safe column name. Lowercase,
  * non-alphanum runs collapse to `_`, leading digits get a `c_` prefix, empty
@@ -25,16 +48,22 @@ export function slugifyColumnName(value: string): string {
 
 /**
  * Pick a non-colliding column name for a block-output `path`. Uses the bare
- * path slug; on collision, appends `_0`, `_1`, …
+ * path slug; on collision, appends `_0`, `_1`, … Like `uniqueColumnName`,
+ * takes a set of LOWERCASED names and claims the chosen name into it (slugs
+ * are already lowercase), so callers never hand-maintain the set.
  */
-export function deriveOutputColumnName(path: string, taken: Set<string>): string {
+export function deriveOutputColumnName(path: string, takenLower: Set<string>): string {
+  const claim = (name: string): string => {
+    takenLower.add(name)
+    return name
+  }
   const base = slugifyColumnName(path)
-  if (!taken.has(base)) return base
+  if (!takenLower.has(base)) return claim(base)
   for (let i = 0; i < 1000; i++) {
     const candidate = `${base}_${i}`
-    if (!taken.has(candidate)) return candidate
+    if (!takenLower.has(candidate)) return claim(candidate)
   }
-  return `${base}_${Date.now()}`
+  return claim(`${base}_${Date.now()}`)
 }
 
 /**
diff --git a/apps/sim/lib/table/columns/service.ts b/apps/sim/lib/table/columns/service.ts
index 4eafabd456d..aba0fce0550 100644
--- a/apps/sim/lib/table/columns/service.ts
+++ b/apps/sim/lib/table/columns/service.ts
@@ -13,8 +13,13 @@ import { db } from '@sim/db'
 import { userTableDefinitions, userTableRows } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
 import { and, count, eq, sql } from 'drizzle-orm'
-import { columnMatchesRef, generateColumnId, getColumnId } from '@/lib/table/column-keys'
-import { COLUMN_TYPES, NAME_PATTERN, TABLE_LIMITS } from '@/lib/table/constants'
+import {
+  assertValidNewColumnName,
+  columnMatchesRef,
+  generateColumnId,
+  getColumnId,
+} from '@/lib/table/column-keys'
+import { COLUMN_TYPES, TABLE_LIMITS } from '@/lib/table/constants'
 import { stripGroupExecutions } from '@/lib/table/rows/executions'
 import { withLockedTable } from '@/lib/table/service'
 import { scaledStatementTimeoutMs, setTableTxTimeouts } from '@/lib/table/tx'
@@ -54,17 +59,8 @@ export async function addTableColumn(
   requestId: string
 ): Promise<TableDefinition> {
   return withLockedTable(tableId, async (table, trx) => {
-    if (!NAME_PATTERN.test(column.name)) {
-      throw new Error(
-        `Invalid column name "${column.name}". Must start with a letter or underscore and contain only alphanumeric characters and underscores.`
-      )
-    }
-
-    if (column.name.length > TABLE_LIMITS.MAX_COLUMN_NAME_LENGTH) {
-      throw new Error(
-        `Column name exceeds maximum length (${TABLE_LIMITS.MAX_COLUMN_NAME_LENGTH} characters)`
-      )
-    }
+    const schema = table.schema
+    assertValidNewColumnName(column.name, schema.columns)
 
     if (!COLUMN_TYPES.includes(column.type as (typeof COLUMN_TYPES)[number])) {
       throw new Error(
@@ -72,11 +68,6 @@ export async function addTableColumn(
       )
     }
 
-    const schema = table.schema
-    if (schema.columns.some((c) => c.name.toLowerCase() === column.name.toLowerCase())) {
-      throw new Error(`Column "${column.name}" already exists`)
-    }
-
     if (schema.columns.length >= TABLE_LIMITS.MAX_COLUMNS_PER_TABLE) {
       throw new Error(
         `Table has reached maximum column limit (${TABLE_LIMITS.MAX_COLUMNS_PER_TABLE})`
@@ -157,31 +148,12 @@ export async function renameColumn(
   requestId: string
 ): Promise<TableDefinition> {
   return withLockedTable(data.tableId, async (table, trx) => {
-    if (!NAME_PATTERN.test(data.newName)) {
-      throw new Error(
-        `Invalid column name "${data.newName}". Column names must start with a letter or underscore, followed by alphanumeric characters or underscores.`
-      )
-    }
-
-    if (data.newName.length > TABLE_LIMITS.MAX_COLUMN_NAME_LENGTH) {
-      throw new Error(
-        `Column name exceeds maximum length (${TABLE_LIMITS.MAX_COLUMN_NAME_LENGTH} characters)`
-      )
-    }
-
     const schema = table.schema
     const columnIndex = schema.columns.findIndex((c) => columnMatchesRef(c, data.oldName))
     if (columnIndex === -1) {
       throw new Error(`Column "${data.oldName}" not found`)
     }
-
-    if (
-      schema.columns.some(
-        (c, i) => i !== columnIndex && c.name.toLowerCase() === data.newName.toLowerCase()
-      )
-    ) {
-      throw new Error(`Column "${data.newName}" already exists`)
-    }
+    assertValidNewColumnName(data.newName, schema.columns, { excludeIndex: columnIndex })
 
     const targetColumn = schema.columns[columnIndex]
     const actualOldName = targetColumn.name
diff --git a/apps/sim/lib/table/constants.ts b/apps/sim/lib/table/constants.ts
index 4a275abfe54..226ca7e177c 100644
--- a/apps/sim/lib/table/constants.ts
+++ b/apps/sim/lib/table/constants.ts
@@ -135,8 +135,29 @@ export function getTablePlanLimits(): TablePlanLimitsByPlan {
 
 export const COLUMN_TYPES = ['string', 'number', 'boolean', 'date', 'json'] as const
 
+/**
+ * Identifier pattern for table names and column **storage keys** (stable column
+ * ids and legacy name-keys). Storage keys are interpolated into JSONB path SQL
+ * (`data->>'key'`), so this doubles as the injection guard — never relax it.
+ */
 export const NAME_PATTERN = /^[a-z_][a-z0-9_]*$/i
 
+/**
+ * Column **display names** are metadata only (rows key on the stable column id),
+ * so any printable characters are allowed — including spaces and leading digits.
+ * Rejected: the Unicode "Other" category `\p{C}` (control, zero-width/bidi
+ * format, surrogate, private-use — invisible or spoofable, never intentional),
+ * leading/trailing whitespace (visually identical duplicates), and a leading
+ * `$` (collides with the `$or`/`$and` filter-grammar keys on the name-keyed
+ * wire). The CSV import dialog's NUL-prefixed combobox sentinels depend on
+ * `\p{C}` staying rejected — don't relax it without changing those sentinels.
+ */
+export const COLUMN_NAME_PATTERN = /^(?![$\s])(?!.*\s$)[^\p{C}]+$/u
+
+/** Human-readable form of `COLUMN_NAME_PATTERN`, shared by every validation site. */
+export const COLUMN_NAME_RULE =
+  'Column name cannot contain control or invisible characters, start with "$", or start/end with whitespace'
+
 export const USER_TABLE_ROWS_SQL_NAME = 'user_table_rows'
 
 /**
diff --git a/apps/sim/lib/table/import-runner.ts b/apps/sim/lib/table/import-runner.ts
index 64694f97bf3..e582b81b7ab 100644
--- a/apps/sim/lib/table/import-runner.ts
+++ b/apps/sim/lib/table/import-runner.ts
@@ -9,13 +9,16 @@ import {
   CSV_MAX_BATCH_SIZE,
   CSV_SCHEMA_SAMPLE_SIZE,
   type CsvHeaderMapping,
+  CsvImportValidationError,
   coerceRowsForTable,
   createCsvParser,
   inferColumnType,
   inferSchemaFromCsv,
-  sanitizeName,
+  sanitizeColumnName,
   type TableSchema,
+  uniqueColumnName,
   validateMapping,
+  validateTableSchema,
 } from '@/lib/table'
 import { assertRowCapacity, notifyTableRowUsage } from '@/lib/table/billing'
 import { withGeneratedColumnIds } from '@/lib/table/column-keys'
@@ -149,6 +152,15 @@ export async function runTableImport(payload: TableImportPayload): Promise<void>
         // Stamp ids so the imported table is id-native (rows coerce + persist by
         // the same ids).
         schema = withGeneratedColumnIds({ columns: inferred.columns.map(normalizeColumn) })
+        // The sync create route validates via createTable; this path writes the
+        // schema directly, so validate here or an over-wide CSV would persist an
+        // invalid schema (e.g. more than MAX_COLUMNS_PER_TABLE columns).
+        const schemaValidation = validateTableSchema(schema)
+        if (!schemaValidation.valid) {
+          throw new CsvImportValidationError(
+            `Invalid schema: ${schemaValidation.errors.join('; ')}`
+          )
+        }
         headerToColumn = inferred.headerToColumn
         await setTableSchemaForImport(tableId, schema)
         return
@@ -168,14 +180,7 @@ export async function runTableImport(payload: TableImportPayload): Promise<void>
         const additions: { name: string; type: string }[] = []
         const updatedMapping: CsvHeaderMapping = { ...effectiveMapping }
         for (const header of payload.createColumns) {
-          const base = sanitizeName(header)
-          let columnName = base
-          let suffix = 2
-          while (usedNames.has(columnName.toLowerCase())) {
-            columnName = `${base}_${suffix}`
-            suffix++
-          }
-          usedNames.add(columnName.toLowerCase())
+          const columnName = uniqueColumnName(sanitizeColumnName(header), usedNames)
           additions.push({ name: columnName, type: inferColumnType(sample.map((r) => r[header])) })
           updatedMapping[header] = columnName
         }
diff --git a/apps/sim/lib/table/import.test.ts b/apps/sim/lib/table/import.test.ts
index d4997a123fe..c34cb0b876a 100644
--- a/apps/sim/lib/table/import.test.ts
+++ b/apps/sim/lib/table/import.test.ts
@@ -3,6 +3,7 @@
  */
 import { Readable } from 'node:stream'
 import { describe, expect, it } from 'vitest'
+import { uniqueColumnName } from '@/lib/table/column-naming'
 import {
   buildAutoMapping,
   CsvImportValidationError,
@@ -13,6 +14,7 @@ import {
   inferColumnType,
   inferSchemaFromCsv,
   parseCsvBuffer,
+  sanitizeColumnName,
   sanitizeName,
   validateMapping,
 } from '@/lib/table/import'
@@ -92,8 +94,50 @@ describe('import', () => {
     })
   })
 
+  describe('sanitizeColumnName', () => {
+    it('preserves spaces, digits, punctuation, and unicode verbatim', () => {
+      expect(sanitizeColumnName('First Name')).toBe('First Name')
+      expect(sanitizeColumnName('2024 Revenue ($)')).toBe('2024 Revenue ($)')
+      expect(sanitizeColumnName('caf\u00e9')).toBe('caf\u00e9')
+    })
+
+    it('collapses control chars and whitespace runs, strips invisible chars, and trims', () => {
+      expect(sanitizeColumnName('a\u0000b')).toBe('a b')
+      expect(sanitizeColumnName('  a   b  ')).toBe('a b')
+      expect(sanitizeColumnName('zero\u200bwidth')).toBe('zerowidth')
+      expect(sanitizeColumnName('$$or')).toBe('or')
+    })
+
+    it('truncates to the max column-name length', () => {
+      expect(sanitizeColumnName('x'.repeat(80))).toHaveLength(50)
+    })
+
+    it('falls back for empty or invisible-only input', () => {
+      expect(sanitizeColumnName('')).toBe('column')
+      expect(sanitizeColumnName('\u0001\u200b')).toBe('column')
+      expect(sanitizeColumnName('', 'field')).toBe('field')
+    })
+  })
+
+  describe('uniqueColumnName', () => {
+    it('returns the base when free, suffixes case-insensitively when taken, and claims the result', () => {
+      const taken = new Set(['first name'])
+      expect(uniqueColumnName('Email', taken)).toBe('Email')
+      expect(uniqueColumnName('First Name', taken)).toBe('First Name_2')
+      expect(taken).toEqual(new Set(['first name', 'email', 'first name_2']))
+    })
+
+    it('keeps the suffixed result within the max length', () => {
+      const base = 'x'.repeat(50)
+      const taken = new Set([base])
+      const result = uniqueColumnName(base, taken)
+      expect(result).toBe(`${'x'.repeat(48)}_2`)
+      expect(result.length).toBeLessThanOrEqual(50)
+    })
+  })
+
   describe('inferSchemaFromCsv', () => {
-    it('produces sanitized column names and inferred types', () => {
+    it('preserves raw headers as column names and infers types', () => {
       const { columns, headerToColumn } = inferSchemaFromCsv(
         ['First Name', 'Age', 'Active'],
         [
@@ -102,20 +146,20 @@ describe('import', () => {
         ]
       )
       expect(columns).toEqual([
-        { name: 'First_Name', type: 'string' },
+        { name: 'First Name', type: 'string' },
         { name: 'Age', type: 'number' },
         { name: 'Active', type: 'boolean' },
       ])
-      expect(headerToColumn.get('First Name')).toBe('First_Name')
+      expect(headerToColumn.get('First Name')).toBe('First Name')
       expect(headerToColumn.get('Age')).toBe('Age')
     })
 
-    it('disambiguates duplicate sanitized headers', () => {
+    it('disambiguates headers that collide case-insensitively', () => {
       const { columns } = inferSchemaFromCsv(
-        ['a b', 'a-b', 'a.b'],
-        [{ 'a b': '1', 'a-b': '2', 'a.b': '3' }]
+        ['a b', 'A B', 'a b '],
+        [{ 'a b': '1', 'A B': '2', 'a b ': '3' }]
       )
-      expect(columns.map((c) => c.name)).toEqual(['a_b', 'a_b_2', 'a_b_3'])
+      expect(columns.map((c) => c.name)).toEqual(['a b', 'A B_2', 'a b_3'])
     })
   })
 
@@ -169,6 +213,19 @@ describe('import', () => {
       const mapping = buildAutoMapping(['unmatched'], schema)
       expect(mapping).toEqual({ unmatched: null })
     })
+
+    it('exact-matches raw display names, trimming padded headers', () => {
+      const rawSchema: TableSchema = { columns: [{ name: 'First Name', type: 'string' }] }
+      expect(buildAutoMapping(['First Name'], rawSchema)).toEqual({ 'First Name': 'First Name' })
+      expect(buildAutoMapping([' First Name '], rawSchema)).toEqual({
+        ' First Name ': 'First Name',
+      })
+    })
+
+    it('never fuzzy-matches columns whose names collapse to an empty loose key', () => {
+      const nonLatin: TableSchema = { columns: [{ name: '\u540d\u524d', type: 'string' }] }
+      expect(buildAutoMapping(['!!!'], nonLatin)).toEqual({ '!!!': null })
+    })
   })
 
   describe('validateMapping', () => {
diff --git a/apps/sim/lib/table/import.ts b/apps/sim/lib/table/import.ts
index 104186e32e1..9b72f8aea14 100644
--- a/apps/sim/lib/table/import.ts
+++ b/apps/sim/lib/table/import.ts
@@ -13,6 +13,8 @@
 
 import { type Options as CsvParseOptions, type Parser, parse as parseCsvStream } from 'csv-parse'
 import { getColumnId } from '@/lib/table/column-keys'
+import { uniqueColumnName } from '@/lib/table/column-naming'
+import { TABLE_LIMITS } from '@/lib/table/constants'
 import { type NormalizeDateCellOptions, normalizeDateCellValue } from '@/lib/table/dates'
 import type { ColumnDefinition, RowData, TableSchema } from '@/lib/table/types'
 
@@ -160,9 +162,11 @@ export function inferColumnType(values: unknown[]): Exclude<CsvColumnType, 'json
 }
 
 /**
- * Sanitizes a raw header into a valid column/table name. Strips disallowed
- * characters, collapses runs of underscores, and ensures the first character
- * is a letter or underscore (prefixing with `fallbackPrefix` otherwise).
+ * Sanitizes a raw string into a valid **table** name (`NAME_PATTERN`). Strips
+ * disallowed characters, collapses runs of underscores, and ensures the first
+ * character is a letter or underscore (prefixing with `fallbackPrefix`
+ * otherwise). Column display names are not identifiers — use
+ * {@link sanitizeColumnName} for headers.
  */
 export function sanitizeName(raw: string, fallbackPrefix = 'col'): string {
   let name = raw
@@ -178,6 +182,27 @@ export function sanitizeName(raw: string, fallbackPrefix = 'col'): string {
   return name
 }
 
+/**
+ * Sanitizes a raw header into a valid column display name
+ * (`COLUMN_NAME_PATTERN`): control characters collapse to a space, other
+ * `\p{C}` characters (zero-width/bidi format, surrogates) and leading `$`s are
+ * stripped, whitespace runs collapse to one space, and the result is trimmed
+ * and truncated to `MAX_COLUMN_NAME_LENGTH`. Everything else — spaces, digits,
+ * punctuation, unicode — is preserved verbatim. Empty results fall back to
+ * `fallback`.
+ */
+export function sanitizeColumnName(raw: string, fallback = 'column'): string {
+  const name = raw
+    .replace(/\p{Cc}+/gu, ' ')
+    .replace(/\p{C}/gu, '')
+    .replace(/\s+/g, ' ')
+    .trim()
+    .replace(/^\$+\s*/, '')
+    .slice(0, TABLE_LIMITS.MAX_COLUMN_NAME_LENGTH)
+    .trim()
+  return name || fallback
+}
+
 /**
  * Returns column definitions inferred from CSV headers + sample rows. Duplicate
  * sanitized names are suffixed with `_2`, `_3`, etc. Also returns the header ->
@@ -192,14 +217,7 @@ export function inferSchemaFromCsv(
   const headerToColumn = new Map<string, string>()
 
   const columns = headers.map((header) => {
-    const base = sanitizeName(header)
-    let colName = base
-    let suffix = 2
-    while (seen.has(colName.toLowerCase())) {
-      colName = `${base}_${suffix}`
-      suffix++
-    }
-    seen.add(colName.toLowerCase())
+    const colName = uniqueColumnName(sanitizeColumnName(header), seen)
     headerToColumn.set(header, colName)
 
     return {
@@ -362,8 +380,10 @@ export function validateMapping(params: {
 
 /**
  * Builds an auto-mapping from CSV headers to table columns: prefers exact
- * sanitized-name matches and falls back to a case- and punctuation-insensitive
- * comparison. Unmapped headers are set to `null`.
+ * name matches (trimmed header) and falls back to a case- and
+ * punctuation-insensitive comparison, which also covers columns created by the
+ * legacy header slugger (`First Name` → `First_Name`). Unmapped headers are
+ * set to `null`.
  */
 export function buildAutoMapping(csvHeaders: string[], tableSchema: TableSchema): CsvHeaderMapping {
   const mapping: CsvHeaderMapping = {}
@@ -372,14 +392,16 @@ export function buildAutoMapping(csvHeaders: string[], tableSchema: TableSchema)
   const exactByName = new Map(columns.map((c) => [c.name, c.name]))
   const loose = new Map<string, string>()
   for (const col of columns) {
-    loose.set(col.name.toLowerCase().replace(/[^a-z0-9]/g, ''), col.name)
+    // Skip names that collapse to nothing (all-punctuation / non-Latin) — an
+    // empty key would make any unmatched header auto-map to this column.
+    const key = col.name.toLowerCase().replace(/[^a-z0-9]/g, '')
+    if (key) loose.set(key, col.name)
   }
 
   const usedTargets = new Set<string>()
 
   for (const header of csvHeaders) {
-    const sanitized = sanitizeName(header)
-    const exact = exactByName.get(sanitized)
+    const exact = exactByName.get(sanitizeColumnName(header))
     if (exact && !usedTargets.has(exact)) {
       mapping[header] = exact
       usedTargets.add(exact)
@@ -430,8 +452,9 @@ export function coerceRowsForTable(
 /**
  * Sanitizes raw JSON keys so they conform to the same column-name rules as CSV
  * headers, letting `inferSchemaFromCsv` and `coerceRowsForTable` be reused for
- * JSON imports. Collisions after sanitization are disambiguated with a trailing
- * underscore. Returns the headers and rows untouched when no key needs renaming.
+ * JSON imports. Collisions are disambiguated the same way as CSV headers
+ * (case-insensitive `_N` suffixes via `uniqueColumnName`). Returns the headers
+ * and rows untouched when no key needs renaming.
  */
 export function sanitizeJsonHeaders(
   headers: string[],
@@ -441,10 +464,7 @@ export function sanitizeJsonHeaders(
   const seen = new Set<string>()
 
   for (const raw of headers) {
-    let safe = sanitizeName(raw)
-    while (seen.has(safe)) safe = `${safe}_`
-    seen.add(safe)
-    renamed.set(raw, safe)
+    renamed.set(raw, uniqueColumnName(sanitizeColumnName(raw), seen))
   }
 
   const noChange = headers.every((h) => renamed.get(h) === h)
diff --git a/apps/sim/lib/table/index.ts b/apps/sim/lib/table/index.ts
index 58a73f2313f..8076efd710e 100644
--- a/apps/sim/lib/table/index.ts
+++ b/apps/sim/lib/table/index.ts
@@ -7,6 +7,7 @@
 
 export * from '@/lib/table/billing'
 export * from '@/lib/table/column-keys'
+export * from '@/lib/table/column-naming'
 export * from '@/lib/table/columns/service'
 export * from '@/lib/table/constants'
 export * from '@/lib/table/dates'
diff --git a/apps/sim/lib/table/llm/enrichment.ts b/apps/sim/lib/table/llm/enrichment.ts
index 2d9cb7d5d67..9b81bba91b7 100644
--- a/apps/sim/lib/table/llm/enrichment.ts
+++ b/apps/sim/lib/table/llm/enrichment.ts
@@ -44,21 +44,23 @@ export function enrichTableToolDescription(
     const stringCols = table.columns.filter((c) => c.type === 'string')
     const numberCols = table.columns.filter((c) => c.type === 'number')
 
+    // Names are JSON-encoded (not hand-quoted) so a name containing `"` still
+    // produces a well-formed example object.
     let filterExample = ''
     if (stringCols.length > 0 && numberCols.length > 0) {
       filterExample = `
 
-Example filter: {"${stringCols[0].name}": {"$eq": "value"}, "${numberCols[0].name}": {"$lt": 50}}`
+Example filter: {${JSON.stringify(stringCols[0].name)}: {"$eq": "value"}, ${JSON.stringify(numberCols[0].name)}: {"$lt": 50}}`
     } else if (stringCols.length > 0) {
       filterExample = `
 
-Example filter: {"${stringCols[0].name}": {"$eq": "value"}}`
+Example filter: {${JSON.stringify(stringCols[0].name)}: {"$eq": "value"}}`
     }
 
     let sortExample = ''
     if (toolId === 'table_query_rows' && numberCols.length > 0) {
       sortExample = `
-Example sort: {"${numberCols[0].name}": "desc"} for highest first, {"${numberCols[0].name}": "asc"} for lowest first`
+Example sort: {${JSON.stringify(numberCols[0].name)}: "desc"} for highest first, {${JSON.stringify(numberCols[0].name)}: "asc"} for lowest first`
     }
 
     const queryInstructions =
diff --git a/apps/sim/lib/table/rows/service.ts b/apps/sim/lib/table/rows/service.ts
index fe12af429b3..33a382da951 100644
--- a/apps/sim/lib/table/rows/service.ts
+++ b/apps/sim/lib/table/rows/service.ts
@@ -23,7 +23,7 @@ import {
   TableRowLimitError,
   wouldExceedRowLimit,
 } from '@/lib/table/billing'
-import { getColumnId } from '@/lib/table/column-keys'
+import { columnMatchesRef, getColumnId } from '@/lib/table/column-keys'
 import { getMaxPageBytes, TABLE_LIMITS, USER_TABLE_ROWS_SQL_NAME } from '@/lib/table/constants'
 import { nKeysBetween } from '@/lib/table/order-key'
 import { type DbExecutor, type DbTransaction, withSeqscanOff } from '@/lib/table/planner'
@@ -513,12 +513,12 @@ export async function upsertRow(
 
   // Determine the single conflict target column, resolving to its stable
   // storage id (the row-data key). `conflictTarget` may arrive as an id
-  // (first-party) or a name (legacy/internal) — match either.
+  // (first-party) or a name (legacy/internal) — match either, with the same
+  // case-insensitive name semantics as every other column-ref resolver.
   let targetColumnKey: string
   if (data.conflictTarget) {
-    const col = uniqueColumns.find(
-      (c) => getColumnId(c) === data.conflictTarget || c.name === data.conflictTarget
-    )
+    const target = data.conflictTarget
+    const col = uniqueColumns.find((c) => columnMatchesRef(c, target))
     if (!col) {
       throw new Error(
         `Column "${data.conflictTarget}" is not a unique column. Available unique columns: ${uniqueColumns.map((c) => c.name).join(', ')}`
diff --git a/apps/sim/lib/table/service.ts b/apps/sim/lib/table/service.ts
index 90f3c6f6e7a..724976f5fb4 100644
--- a/apps/sim/lib/table/service.ts
+++ b/apps/sim/lib/table/service.ts
@@ -17,8 +17,13 @@ import { and, count, eq, isNull, sql } from 'drizzle-orm'
 import { generateRestoreName } from '@/lib/core/utils/restore-name'
 import type { DbOrTx } from '@/lib/db/types'
 import { assertRowCapacity, notifyTableRowUsage } from '@/lib/table/billing'
-import { generateColumnId, getColumnId, withGeneratedColumnIds } from '@/lib/table/column-keys'
-import { COLUMN_TYPES, NAME_PATTERN, TABLE_LIMITS } from '@/lib/table/constants'
+import {
+  assertValidNewColumnName,
+  generateColumnId,
+  getColumnId,
+  withGeneratedColumnIds,
+} from '@/lib/table/column-keys'
+import { COLUMN_TYPES, TABLE_LIMITS } from '@/lib/table/constants'
 import { EMPTY_JOB_FIELDS, latestJobForTable, latestJobsForTables } from '@/lib/table/jobs/service'
 import { nKeysBetween } from '@/lib/table/order-key'
 import type { DbTransaction } from '@/lib/table/planner'
@@ -421,30 +426,16 @@ export async function addTableColumnsWithTx(
 ): Promise<TableDefinition> {
   if (columns.length === 0) return table
 
-  const usedNames = new Set(table.schema.columns.map((c) => c.name.toLowerCase()))
+  const batchNames = new Set<string>()
   const additions: TableSchema['columns'] = []
 
   for (const column of columns) {
-    if (!NAME_PATTERN.test(column.name)) {
-      throw new Error(
-        `Invalid column name "${column.name}". Must start with a letter or underscore and contain only alphanumeric characters and underscores.`
-      )
-    }
-    if (column.name.length > TABLE_LIMITS.MAX_COLUMN_NAME_LENGTH) {
-      throw new Error(
-        `Column name exceeds maximum length (${TABLE_LIMITS.MAX_COLUMN_NAME_LENGTH} characters)`
-      )
-    }
+    assertValidNewColumnName(column.name, table.schema.columns, { extraTakenLower: batchNames })
     if (!COLUMN_TYPES.includes(column.type as (typeof COLUMN_TYPES)[number])) {
       throw new Error(
         `Invalid column type "${column.type}". Must be one of: ${COLUMN_TYPES.join(', ')}`
       )
     }
-    const lower = column.name.toLowerCase()
-    if (usedNames.has(lower)) {
-      throw new Error(`Column "${column.name}" already exists`)
-    }
-    usedNames.add(lower)
     // Honor a caller-assigned id (the CSV append path pre-assigns so coercion
     // and persistence agree); otherwise mint one.
     const id = column.id ?? generateColumnId()
diff --git a/apps/sim/lib/table/sql.ts b/apps/sim/lib/table/sql.ts
index 5e695d93195..4087a7c2e9e 100644
--- a/apps/sim/lib/table/sql.ts
+++ b/apps/sim/lib/table/sql.ts
@@ -241,8 +241,12 @@ function validateFieldName(field: string): void {
   }
 
   if (!NAME_PATTERN.test(field)) {
+    // Fields reaching this layer are storage keys (column ids) or unmatched
+    // passthrough names — by this point a non-identifier field means the caller
+    // referenced a column that doesn't exist, so say that instead of reciting
+    // the identifier rule (display names are allowed to violate it).
     throw new TableQueryValidationError(
-      `Invalid field name "${field}". Field names must start with a letter or underscore, followed by alphanumeric characters or underscores.`
+      `Unknown field "${field}". Filter and sort fields must reference an existing column.`
     )
   }
 }
diff --git a/apps/sim/lib/table/validation.ts b/apps/sim/lib/table/validation.ts
index df773160513..f4204d435ad 100644
--- a/apps/sim/lib/table/validation.ts
+++ b/apps/sim/lib/table/validation.ts
@@ -6,8 +6,15 @@ import { db } from '@sim/db'
 import { userTableRows } from '@sim/db/schema'
 import { and, eq, or, type SQL, sql } from 'drizzle-orm'
 import { NextResponse } from 'next/server'
-import { getColumnId } from '@/lib/table/column-keys'
-import { COLUMN_TYPES, getMaxRowSizeBytes, NAME_PATTERN, TABLE_LIMITS } from '@/lib/table/constants'
+import { getColumnId, nameShadowsColumnId } from '@/lib/table/column-keys'
+import {
+  COLUMN_NAME_PATTERN,
+  COLUMN_NAME_RULE,
+  COLUMN_TYPES,
+  getMaxRowSizeBytes,
+  NAME_PATTERN,
+  TABLE_LIMITS,
+} from '@/lib/table/constants'
 import { normalizeDateCellValue } from '@/lib/table/dates'
 import { withSeqscanOff } from '@/lib/table/planner'
 import type {
@@ -208,6 +215,13 @@ export function validateTableSchema(schema: TableSchema): ValidationResult {
     errors.push('Duplicate column names found')
   }
 
+  for (let i = 0; i < schema.columns.length; i++) {
+    const column = schema.columns[i]
+    if (nameShadowsColumnId(schema.columns, column.name, i)) {
+      errors.push(`Column name "${column.name}" conflicts with another column's id`)
+    }
+  }
+
   return { valid: errors.length === 0, errors }
 }
 
@@ -662,7 +676,12 @@ export async function checkBatchUniqueConstraintsDb(
   return { valid: rowErrors.length === 0, errors: rowErrors }
 }
 
-/** Validates column definition format and type. */
+/**
+ * Validates column definition format and type, accumulating errors for bulk
+ * schema checks. Write paths use the throwing `assertValidNewColumnName`
+ * (column-keys.ts), which adds uniqueness and id-shadow checks; the
+ * pattern/length rules here mirror it and must stay in sync.
+ */
 export function validateColumnDefinition(column: ColumnDefinition): ValidationResult {
   const errors: string[] = []
 
@@ -677,10 +696,8 @@ export function validateColumnDefinition(column: ColumnDefinition): ValidationRe
     )
   }
 
-  if (!NAME_PATTERN.test(column.name)) {
-    errors.push(
-      `Column name "${column.name}" must start with letter or underscore, followed by alphanumeric or underscore`
-    )
+  if (!COLUMN_NAME_PATTERN.test(column.name)) {
+    errors.push(`Column name "${column.name}" is invalid: ${COLUMN_NAME_RULE}`)
   }
 
   if (!COLUMN_TYPES.includes(column.type)) {
diff --git a/apps/sim/lib/table/workflow-groups/service.ts b/apps/sim/lib/table/workflow-groups/service.ts
index 8479118650d..20e2a950c97 100644
--- a/apps/sim/lib/table/workflow-groups/service.ts
+++ b/apps/sim/lib/table/workflow-groups/service.ts
@@ -13,12 +13,13 @@ import { createLogger } from '@sim/logger'
 import { and, eq, isNull, sql } from 'drizzle-orm'
 import type { DbOrTx } from '@/lib/db/types'
 import {
+  assertValidNewColumnName,
   columnMatchesRef,
   generateColumnId,
   getColumnId,
   remapGroupColumnRefs,
 } from '@/lib/table/column-keys'
-import { NAME_PATTERN, TABLE_LIMITS } from '@/lib/table/constants'
+import { TABLE_LIMITS } from '@/lib/table/constants'
 import { stripGroupExecutions } from '@/lib/table/rows/executions'
 import { getTableById, withLockedTable } from '@/lib/table/service'
 import { setTableTxTimeouts } from '@/lib/table/tx'
@@ -36,6 +37,27 @@ import type {
 import { assertValidSchema, runWorkflowColumn, stripGroupDeps } from '@/lib/table/workflow-columns'
 
 const logger = createLogger('TableWorkflowGroupsService')
+
+/**
+ * Validates the display names of to-be-created output columns (via
+ * `assertValidNewColumnName`, including duplicates within the batch) and the
+ * resulting column count. Shared by all three column-creating group ops so
+ * none of them drifts. No-op for an empty batch so ops that add no columns
+ * (e.g. a mapping-only group update) never trip the count check on a legacy
+ * over-limit table.
+ */
+function assertNewOutputColumns(names: string[], schema: TableSchema): void {
+  if (names.length === 0) return
+  const batchLower = new Set<string>()
+  for (const name of names) {
+    assertValidNewColumnName(name, schema.columns, { extraTakenLower: batchLower })
+  }
+  if (schema.columns.length + names.length > TABLE_LIMITS.MAX_COLUMNS_PER_TABLE) {
+    throw new Error(
+      `Adding ${names.length} column${names.length === 1 ? '' : 's'} would exceed the maximum (${TABLE_LIMITS.MAX_COLUMNS_PER_TABLE}).`
+    )
+  }
+}
 /**
  * Drops references to deleted blocks from every workflow group on every table
  * that targets the just-deployed workflow. Called from the workflow deploy
@@ -126,23 +148,10 @@ export async function addWorkflowGroup(
       throw new Error(`Workflow group "${data.group.id}" already exists`)
     }
 
-    const existingNames = new Set(schema.columns.map((c) => c.name.toLowerCase()))
-    for (const col of data.outputColumns) {
-      if (!NAME_PATTERN.test(col.name)) {
-        throw new Error(
-          `Invalid output column name "${col.name}". Must satisfy ${NAME_PATTERN.source}.`
-        )
-      }
-      if (existingNames.has(col.name.toLowerCase())) {
-        throw new Error(`Column "${col.name}" already exists`)
-      }
-    }
-
-    if (schema.columns.length + data.outputColumns.length > TABLE_LIMITS.MAX_COLUMNS_PER_TABLE) {
-      throw new Error(
-        `Adding ${data.outputColumns.length} columns would exceed the maximum (${TABLE_LIMITS.MAX_COLUMNS_PER_TABLE}).`
-      )
-    }
+    assertNewOutputColumns(
+      data.outputColumns.map((c) => c.name),
+      schema
+    )
 
     // Assign stable ids to the new output columns, then rewrite the group's
     // column refs from name → id so outputs/deps/inputMappings key on ids —
@@ -290,6 +299,16 @@ export async function updateWorkflowGroup(
       }
       const group = groups[groupIndex]
 
+      // New output columns go through the same validation as the other two
+      // column-creating group ops (addWorkflowGroup / addWorkflowGroupOutput) —
+      // this path was the only one that skipped it. Names are checked against
+      // the pre-update schema, so re-using the name of an output removed in
+      // this same update is rejected.
+      assertNewOutputColumns(
+        (data.newOutputColumns ?? []).map((c) => c.name),
+        schema
+      )
+
       // Normalize every caller-supplied column reference to its stable id, so
       // the diff/splice/clear logic below operates uniformly in id-space (the
       // row-data storage key). New output columns get ids first; then output
@@ -674,19 +693,10 @@ export async function addWorkflowGroupOutput(
       )
     }
 
-    const taken = new Set(schema.columns.map((c) => c.name))
-    const columnName = data.columnName ?? deriveOutputColumnName(data.path, taken)
-    if (!NAME_PATTERN.test(columnName)) {
-      throw new Error(`Invalid column name "${columnName}". Must satisfy ${NAME_PATTERN.source}.`)
-    }
-    if (taken.has(columnName)) {
-      throw new Error(`Column "${columnName}" already exists`)
-    }
-    if (schema.columns.length + 1 > TABLE_LIMITS.MAX_COLUMNS_PER_TABLE) {
-      throw new Error(
-        `Adding a column would exceed the maximum (${TABLE_LIMITS.MAX_COLUMNS_PER_TABLE}).`
-      )
-    }
+    const columnName =
+      data.columnName ??
+      deriveOutputColumnName(data.path, new Set(schema.columns.map((c) => c.name.toLowerCase())))
+    assertNewOutputColumns([columnName], schema)
 
     const newColDef: ColumnDefinition = {
       id: generateColumnId(),