Files
sim/packages/db/migrations/0224_table_import_columns.sql
T
Theodore LiandClaude Opus 4.8 3518b999af feat(tables): background import for large CSVs with live progress (#4861)
* feat(tables): background import for large CSVs with live progress

* fix(tables): address review — import heartbeat, overlap guard, column/empty validation

* fix(tables): guard sync import overlap, scope fileKey to workspace, delete-on-replace after download

* fix(tables): stream large CSV imports from storage instead of buffering the whole file

* test(tables): fix async-import route tests for workspace-scoped fileKey + name uniquification

* fix(tables): append imports start after existing rows; reconcile missed import failures in the tray

* fix(tables): delete the uploaded CSV from storage after the import finishes

* fix(tables): validate replace before deleting rows; ignore stale replayed import events by importId

* fix(tables): bind import worker to its importId (no stale-worker clobber/overlap) and destroy storage stream on failure

* feat(tables): byte-based import progress, cancel support, and a start toast that opens the import view

* fix(tables): don't emit ready after cancel; honor cancel during the upload phase

* improvement(tables): use a stop (square) icon for canceling an active import

* fix(tables): make markTableImporting an atomic claim to close the concurrent-import TOCTOU race

* improvement(tables): preview CSV import from a slice, drop client row-count warning

The import dialog parsed the entire file in the browser to show an exact row
count and a row-limit warning. That holds the whole file in memory, blocks the
main thread, and hits V8's ~512MB string ceiling — so the dialog capped the
effective import size well below what the streaming importer handles.

Parse only the first 512KB (headers + sample for the mapping); drop the exact
count and the "would exceed the row limit by N" gate. The DB row-count trigger
already enforces max_rows server-side, so an over-limit import fails fast during
the run with a clear message instead of being blocked by an expensive parse.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* fix(tables): gate import ownership every batch and stop canceled imports reappearing

- Worker checked run ownership only at the progress cadence (~every 5k rows), so
  a canceled/superseded import could insert several more batches (incl. the final
  partial batch) before stopping. Move the updateImportProgress ownership gate to
  the top of every flush — a run that lost the table stops within one batch.
- A list/dialog import canceled mid-upload left the server row `importing` until
  the in-flight server cancel landed; hydration re-seeded it from useTablesList,
  so the dismissed import flickered back. Flag the real table id canceled on the
  mid-upload cancel path, skip re-seeding flagged tables in hydration, and clear
  the flag once the server import is terminal.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* refactor(tables): drive import tray by polling derived from server, not SSE

Import progress no longer holds an SSE connection per importing table. The tray
now derives its importing rows live from the table list (React Query), polled
only while an import is in flight; the table detail page keeps its own
cell-state SSE for grid refresh.

- store holds only client-only state now: optimistic uploads, which terminal
  completions to surface this session, canceled ids, menu open — no copied
  importStatus/rowsProcessed.
- useWorkspaceImports is the single source: polls via a data-predicate
  refetchInterval, derives rows, and fires completion toasts on the
  importing -> terminal transition.
- kickoff handlers use startUpload/setUploadPercent/endUpload; the invalidated
  list refetch surfaces the server row and polling takes over.
- removes use-hydrate-import-tray + use-import-progress-tracker (folded in).
- trims over-verbose comments across the import paths.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* fix(tables): ignore superseded-run import events in the detail SSE cache

applyImport applied every replayed import payload to the detail cache. The SSE
buffer can replay a prior import's terminal event for the same table, stomping a
newer in-flight import's UI. Lock to the active run's importId (and ignore a
replayed terminal before the id is known), matching the guard the header tracker
used to have.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* fix(tables): close sync-import TOCTOU by claiming the atomic import gate

The sync import route checked importStatus from a checkAccess snapshot, then
parsed/validated/wrote seconds later without taking the atomic claim. A
concurrent async kickoff (markTableImporting) could slip into that window and
both writers would run together — for replace mode, two delete+insert passes
leave the table indeterminate.

Claim the same atomic gate (markTableImporting) right before the write and
release it in the finally (before the response returns, so a client refetch
never sees the transient status). A row-level FOR UPDATE was avoided on purpose:
it would invert lock order against the position advisory lock / row-count
trigger and risk a deadlock — markTableImporting is the established gate.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* fix(multipart): keep abort wired after resolve so a mid-upload disconnect tears down the stream

readMultipart resolves on the file-part header and hands the caller an un-drained
stream, but settle() ran cleanup() and detached the abort listener on that path
too. A client disconnect mid-upload then destroyed nothing — busboy never saw EOF,
the file stream stalled, and the route's `for await` held a request slot until
maxDuration (300s). Re-arm an abort handler scoped to the file stream on resolve,
detached when the stream closes.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-03 21:26:06 -04:00

93 lines
3.6 KiB
PL/PgSQL

ALTER TABLE "user_table_definitions" ADD COLUMN "import_status" text;--> statement-breakpoint
ALTER TABLE "user_table_definitions" ADD COLUMN "import_id" text;--> statement-breakpoint
ALTER TABLE "user_table_definitions" ADD COLUMN "import_error" text;--> statement-breakpoint
ALTER TABLE "user_table_definitions" ADD COLUMN "import_rows_processed" integer DEFAULT 0 NOT NULL;--> statement-breakpoint
ALTER TABLE "user_table_definitions" ADD COLUMN "import_started_at" timestamp;--> statement-breakpoint
-- ============================================================
-- Statement-level row-count maintenance for user_table_rows.
--
-- Replaces the per-row BEFORE INSERT / AFTER DELETE triggers from migration 0158
-- (whose increment function body was rewritten race-free in 0198, but which still
-- fired FOR EACH ROW). Per-row firing serialized a row-level lock on the single
-- user_table_definitions row once per inserted/deleted row -- the dominant cost and
-- contention point for bulk operations (e.g. a 1M-row import = 1M lock cycles).
--
-- The statement-level versions use transition tables to bump row_count by the
-- per-table count of affected rows in ONE UPDATE per statement, preserving the
-- atomic cap check. Transition tables require AFTER triggers, so the insert trigger
-- moves BEFORE -> AFTER: rows are inserted, then the count is bumped with the cap
-- check; an over-cap batch RAISEs and rolls back the whole statement.
-- ============================================================
CREATE OR REPLACE FUNCTION increment_user_table_row_count_stmt()
RETURNS TRIGGER AS $$
DECLARE
over_cap text;
BEGIN
-- Per-table counts within this statement; one capped UPDATE per affected table.
-- A table_id present in the inserted rows always exists (FK), so any table the
-- UPDATE did not touch was rejected by the `row_count + n <= max_rows` guard.
WITH counts AS (
SELECT table_id, count(*)::int AS n
FROM new_rows
GROUP BY table_id
),
updated AS (
UPDATE user_table_definitions d
SET row_count = d.row_count + c.n,
updated_at = now()
FROM counts c
WHERE d.id = c.table_id
AND d.row_count + c.n <= d.max_rows
RETURNING d.id
)
SELECT string_agg(c.table_id, ', ')
INTO over_cap
FROM counts c
WHERE c.table_id NOT IN (SELECT id FROM updated);
IF over_cap IS NOT NULL THEN
RAISE EXCEPTION 'Maximum row limit reached for table(s) %', over_cap
USING ERRCODE = 'check_violation';
END IF;
RETURN NULL;
END;
$$ LANGUAGE plpgsql;
--> statement-breakpoint
CREATE OR REPLACE FUNCTION decrement_user_table_row_count_stmt()
RETURNS TRIGGER AS $$
BEGIN
UPDATE user_table_definitions d
SET row_count = GREATEST(d.row_count - c.n, 0),
updated_at = now()
FROM (
SELECT table_id, count(*)::int AS n
FROM old_rows
GROUP BY table_id
) c
WHERE d.id = c.table_id;
RETURN NULL;
END;
$$ LANGUAGE plpgsql;
--> statement-breakpoint
DROP TRIGGER IF EXISTS user_table_rows_insert_trigger ON user_table_rows;--> statement-breakpoint
DROP TRIGGER IF EXISTS user_table_rows_delete_trigger ON user_table_rows;--> statement-breakpoint
CREATE TRIGGER user_table_rows_insert_stmt_trigger
AFTER INSERT ON user_table_rows
REFERENCING NEW TABLE AS new_rows
FOR EACH STATEMENT
EXECUTE FUNCTION increment_user_table_row_count_stmt();
--> statement-breakpoint
CREATE TRIGGER user_table_rows_delete_stmt_trigger
AFTER DELETE ON user_table_rows
REFERENCING OLD TABLE AS old_rows
FOR EACH STATEMENT
EXECUTE FUNCTION decrement_user_table_row_count_stmt();