[fcollman]

Makes the bulk upload workflow practical for very large files. On 2026-10-22 a 337M-row (25.8 GB) synapse CSV for minnie spent 21.5 h in process_csv. That ran past the 6 h visibility timeout, so the task was redelivered and imported twice; this helped fill the minnie Cloud SQL disk on Oct 24. The spatial lookup that follows was the next bottleneck.

Tested end to end on ltv7 (vFastCsv1–vFastCsv3) with a 4,795,306-row slice of the same file. CSV processing took 2 min, against an estimated ~25 min single-threaded at the old rate. The spatial lookup took 15 min, against 62 min before this change.

Changes

1. Vectorized point geometry (processor.py)
SchemaProcessor.process_chunk built every point through iterrows() and a Shapely Point, three points per row. encode_points_wkb_hex now writes the EWKB POINT Z hex for a whole column with NumPy. The output is byte-identical to the old path, including rows dropped for missing coordinates. It is 49× faster on 1M real rows.

2. Parallel byte-range processing (gcs_processor.py, tasks.py)

• Planning: process_csv splits the source into ~128 MB byte ranges that each start on a line (UPLOAD_CSV_PART_SIZE_MB).
• Parts: each range runs as process:process_csv_part, about a minute per part, well inside the process queue's 300 s visibility timeout. A part overwrites its own output, so a retry or redelivery is safe.
• Join: GCS compose joins the parts in order into the single CSV that upload_to_database imports.
• Generated IDs: they continue across parts, using per-part line counts from process:count_csv_part.
• Fallback: a file that fits in one part, or that can't be split by lines, uses the old single-stream path. A part detects an unsplittable file by comparing rows read with lines read; this catches a quoted field holding a newline, or blank lines.
• Cancel and failure: cancelling revokes the waiting parts, and a failed or cancelled run deletes the parts it wrote.
• Pre-pass removed: the full row-count pre-pass is gone; progress is now parts done out of parts total.

3. Read the CSV in place from gs:// or public s3:// (source.py, api.py, wizard step 1)

• Wizard: step 1 can take a cloud location instead of a browser upload.
• Inspect endpoint: POST /materialize/upload/api/source/inspect/<datastack> checks that the file can be read and returns its size and first rows for the column mapping. It needs edit permission on the datastack, the same as the browser upload.
• Start endpoint: /api/process/start takes sourceUri.
• GCS: read with the deployment's storage client, so any object its service account can read works, which includes every public object.
• S3: read with unsigned requests to the bucket's region (boto3 is already a dependency).
• Output: processed output is still written to the upload bucket.

4. Spatial lookup: one download per storage chunk, and connection pools kept between tasks (spatial_lookup.py, database.py, app.py, celery_worker.py)

Profiled on ltv7 from the logs of 806 sub-batches:

Per point	Before	After
Supervoxel lookup	11.6 ms	2.06 ms
Setup before get_roots	15.4 ms	0.07 ms
get_roots	1.9 ms	1.18 ms
Upsert	0.8 ms	0.09 ms
Total	30.1 ms	3.65 ms

• Supervoxels. scattered_points downloads and gunzips the whole 256×256×32 chunk again for every point. A 50-point sub-batch was spread over a 4201-unit grid cell, so 48–50 of its 50 points hit different chunks: 9.6M downloads (~2.9 TB) for 1.3M distinct chunks. Decompression itself is cheap (gunzip 1.5 ms, voxel read 0.03 ms per chunk).
  • process_chunk now orders a spatial chunk's points, pre and post together, by storage chunk, and builds sub-batches only from whole chunks: a target of 500 points, about 45 chunks.
  • lookup_supervoxel_ids fetches each chunk once and reads every point in it. The labels are identical to scattered_points.
  • Sharded volumes still use scattered_points.
• Connections. Every Celery task runs in an app context whose teardown_appcontext disposed all engines. So every ~1.5 s sub-batch opened new connections: ~60 per second, which held the 2-vCPU ltv Cloud SQL instance at ~100% CPU, and the existing-roots query took 0.77 s instead of ~15 ms.
  • Tasks declared with keep_db_connections=True now keep their pools. These are the two spatial lookup tasks, which only use the staging and aligned-volume databases.
  • Everything else is unchanged. Idle pooled connections to a version database would block its DROP DATABASE when the version expires, so this is deliberately narrow.
  • After the change the SQL instance ran at 20–34% CPU with 2–4 new connections per second.

Verification on ltv7

• CSV processing: 3 parts of 20–27 s each read exactly 4,795,306 rows, the file's row count. The joined output imported with no duplicate IDs, and spot-checked rows match the CSV.
• Cloud location: the same file, read in place from gs://cave-ltv-materialization-upload/... using ltv7's service account.
• Spatial lookup: the vFastCsv3 run's segmentation table was compared with the vFastCsv2 run's in 364 buckets of 20,000 IDs. Row counts and supervoxel ID sums are identical, and the rows with no root ID match (18 pre, 12 post).

Tests

New:

• test_upload_processor.py: byte-identical geometry, NaN and string edge cases, the combined [x,y,z] format.
• test_upload_csv_parts.py: range planning, compose order beyond 32 parts, parts against one stream with explicit and generated IDs, the fallback, and cancel and failure handling.
• test_upload_source.py: URI parsing, S3 and GCS sources, the inspect endpoint, request validation.
• test_spatial_supervoxel_lookup.py: on local volumes in compressed_segmentation and raw encodings, labels match scattered_points, a missing chunk reads as 0, each chunk is downloaded once, chunks are never split across sub-batches, and the keep_db_connections flag works.

127 spatial and upload tests pass, and ruff is clean.

Defaults changed

• SUPERVOXEL_BATCH_SIZE default: 50 → 500. It's now a target number of points; a sub-batch takes whole storage chunks.
• New setting UPLOAD_CSV_PART_SIZE_MB (default 128).

Known follow-ups (not in this PR)

• Progress counter: the spatial lookup's rows_processed counts rows touched per upsert. Now that a synapse's pre and post points usually share a sub-batch, it stops near half of the point estimate (5.46M of 9.59M when this run finished). It should count points.
• Dispatch: on this test slice the spatial workers were busy only ~17% of the time. The grid was 96% empty chunks, and the dispatcher waits for the slowest chunk in each batch of 50. That may matter less on a full, dense table.
• Boundary points: points lying exactly on a grid-chunk boundary are looked up by both chunks (0.08% of points here). This is harmless, because the upsert writes the same IDs.
• Log noise: composite source objects make google-cloud-storage log a CRC32C INFO line on every ranged read.
• drop_upload_tables: it fails on staging databases without combined_table_metadata (seen on minnie).

🤖 Generated with Claude Code