FFI: Thread-safe writer

[myrrc]

Replace vx_array_sink with vx_writer interface:

• Allow concurrent *_push calls. File writes are still serial but encoding process can be done on multiple threads in parallel. Since we sit over a thread pool on host threads, user can create a thread pool for maximum performance.
• Make vx_writer_free mandatory even after vx_writer_close. This allows to signal failure from vx_writer_push() instead of being a UB as before.
• Return write summary after finishing the file.
• Add a backpressure control concurrent_array_limit for vx_writer_open to control number of arrays which are allowed to be processed in parallel.
• Adapt cuda-ffi header and docs.

[0ax1]

@claude review in depth, in particular wrt to correctness and write performance

[github-actions]

Claude finished @0ax1's task in 4m 5s —— View job

---

Claude PR Review — vx_array_sink → vx_writer, parallel writes

Reviewed for correctness and write performance as requested. I read the FFI writer (vortex-ffi/src/writer.rs), the runtime it drives (CurrentThreadRuntime/smol), the underlying file writer (vortex-file/src/writer.rs), the C header, C++ wrapper, and the new tests.

Overall: the design is sound. Cloning the Sender under the lock and then blocking on send outside the lock is the right pattern for concurrent producers, and vx_writer is genuinely Send + Sync (both fields behind parking_lot::Mutex, DType immutable). Making vx_writer_free mandatory and letting push surface failure via a closed channel is a real improvement over the previous UB. Below are the issues worth addressing before merge, ordered by importance.

---

Correctness

1. Chunk ordering across concurrent pushes is non-deterministic — but the contract doesn't say so.
All threads feed a single mpsc::channel(32); the order chunks land in the file is the order send happens to win the race, not the order the user called push. For a columnar file this means row order is not preserved under concurrent push. The header (vortex.h:1717-1724) and writer.hpp:16-23 only say "Thread safe", which a caller will reasonably read as "same result as serial." This is the single most important semantic caveat and should be documented explicitly (e.g. "Thread-safe, but chunk/row order is unspecified when pushes are concurrent; serialize pushes if you need deterministic order"). The Concurrent push test only asserts row_count, so it does not exercise/guard ordering. Fix this →

2. push after the writer task has already failed returns a generic "Send error".
When the background task dies (e.g. bad path in File::create), the receiver drops and sender.send fails, so push reports "Send error: {e}" rather than the real cause — the real error is only retrievable at close. That's acceptable and matches test_writer_invalid_path, but the message is misleading; consider noting in the doc that the authoritative error is delivered by vx_writer_close.

3. close concurrent with an in-flight push. close is documented thread-unsafe, which is fine, but note the actual behavior: a push that already cloned its Sender keeps the channel alive, so close's writer.await will block until that push drains. Worth a sentence in the docs since "thread-unsafe" understates that close can block on a racing push rather than just misbehave.

No memory-safety or data-race bug found in the Rust side — the lock scoping (writer.sink.lock().clone() guard dropped before send) is correct, and block_on(executor.run(...)) from multiple threads is exactly the multi-driver model CurrentThreadRuntime documents.

Write performance

4. Be precise about where the parallelism comes from. The PR says "encoding process can be done on multiple threads." Mechanically, push itself only does an Arc clone + channel send — cheap. The real parallelism is that each push thread calls RUNTIME.block_on → smol::Executor::run, adding a driver thread to the shared executor. The actual CPU-bound serialization is parallel only because the layout strategy inside write_internal spawns sub-tasks (spawn_nested, see the comment at vortex-file/src/writer.rs:160-164). Consequences worth knowing:

• A single push in flight ≈ one driver; encoding throughput scales with how many threads are simultaneously inside push/close, bounded by the strategy's own task granularity — not by the number of chunks buffered.
• The final flush → file is a kanal::bounded_async(1) single-consumer path (correctly, "file writes are still serial").
  So the speedup is real but comes from extra executor drivers + parallel user-side array construction, not from the channel depth. A short note in the PR description would set correct expectations.

5. Channel capacity 32 governs peak memory. With large chunks (the perf test uses 300k × u64 = 2.4 MB each), 32 buffered ≈ 76 MB in flight per writer, plus one extra slot per live Sender clone (futures mpsc reserves a guaranteed slot per producer). Fine, but the "chosen arbitrarily" comment undersells that this is the backpressure/memory knob.

Tests

6. Concurrent push is faster than serial is a hard timing assertion and will likely flake in CI.
REQUIRE(parallel < serial) (test/writer.cpp:196) compares wall-clock once, with no margin or repetition. On a CPU-limited/loaded runner — or where std::thread::hardware_concurrency() over-reports vs. the cgroup CPU quota — parallel can lose. Recommend either moving this to a benchmark (not a pass/fail test), requiring a margin with a best-of-N, or gating it behind an env flag. Also its tag is "[sink]" while the other cases use "[writer]" — leftover from the rename. Fix this →

7. Concurrency test doesn't verify data, only count. Concurrent push checks row_count == NUM_THREADS * CHUNK_LEN but never reads the values back. Since chunks carry a deterministic pattern (start+i)%997, reading back and checking the multiset of values (order-independent) would catch corruption/lost-chunk bugs that a count check misses. Worth adding given the whole point is concurrent correctness.

Minor / nits

• Duplicated TempPath. The new test/temp_path.hpp reintroduces a TempPath that already exists verbatim in test/scan.cpp:31-54. scan.cpp should include the new header instead of keeping its own copy.
• vx_writer_free(sink: *mut vx_writer) — parameter still named sink (in both writer.rs:153 and vortex.h:1750), a leftover from sink.rs. Cosmetic.
• #[expect(non_camel_case_types)] at writer.rs:29 sits above the doc comment with a blank line between; it still applies to vx_writer, but placing it directly adjacent to the struct (below the doc comment) reads more clearly.
• Good: write_sample.c and the C++ wrapper correctly call vx_writer_free on every path now, matching the new mandatory-free contract.

---

I did not run cargo/clippy or the C++ tests (review-only). None of the above are blocking soundness bugs; items 1 (document ordering) and 6 (flaky timing test) are the ones I'd most want resolved before merge.

[gatesn]

This seems like a weird API? The user pushes chunks from multiple threads, so they're not deterministically ordered?

I think you might just want to expose functions for driving the CurrentThreadRuntime. Like "runtime.drive(2 seconds)" and then you just drive the runtime until your 2 seconds timer is up.

[myrrc]

I don't think we should expose timers and driving the runtime in general. Presumably user should care about ordering on their side if they want parallel processing.

If we want to be on par with Parquet, fastparquet/dask allows partition-based parallelism, and this PR does the same.

[github-actions]

Polar Signals Profiling Results

Latest Run

Status	Commit	Job	Attempt	Link
🟢 Done	5d54925		1	Explore Profiling Data

Previous Runs (1)

Status	Commit	Job	Attempt	Link
🟢 Done	e6df2b4		1	Explore Profiling Data

---

Powered by Polar Signals Cloud

[github-actions]

Benchmarks: Vortex queries 📖

Verdict: Likely regression (medium confidence)
Attributed Vortex impact: +11.5%
Engines: DataFusion Likely regression (+22.0%, high confidence) · DuckDB No clear signal (+1.8%, low confidence)
Vortex (geomean): 1.024x ➖
Parquet (geomean): 0.970x ➖
Shifts: Parquet (control) -3.0% · Median polish +0.8%

How to read Verdict and Engines

• Verdict: Overall PR-level signal after subtracting baseline drift estimated from Parquet control rows. It can be Likely improvement, Likely regression, or No clear signal.
• Engines: Per-engine attribution. DataFusion is compared against DataFusion/Parquet controls; DuckDB is compared against DuckDB/Parquet controls. This answers whether each engine improved or regressed independently.
• Confidence: Based on directional consistency, share of rows above the noise floor, and control-run noise.

---

datafusion / vortex-file-compressed (1.148x ❌, 0↑ 2↓)

name	PR 5d54925 (ns)	base 30091f2 (ns)	ratio (PR/base)
vortex_q00/datafusion:vortex-file-compressed 🚨	10071672	8782896	1.15
vortex_q01/datafusion:vortex-file-compressed 🚨	6227306	5418053	1.15

datafusion / parquet (0.941x ➖, 0↑ 0↓)

name	PR 5d54925 (ns)	base 30091f2 (ns)	ratio (PR/base)
vortex_q00/datafusion:parquet	20646641	21877618	0.94
vortex_q01/datafusion:parquet	4829729	5149794	0.94

duckdb / vortex-file-compressed (1.017x ➖, 0↑ 0↓)

name	PR 5d54925 (ns)	base 30091f2 (ns)	ratio (PR/base)
vortex_q00/duckdb:vortex-file-compressed	10140254	10052416	1.01
vortex_q01/duckdb:vortex-file-compressed	6084550	5929380	1.03

duckdb / parquet (1.000x ➖, 0↑ 0↓)

name	PR 5d54925 (ns)	base 30091f2 (ns)	ratio (PR/base)
vortex_q00/duckdb:parquet	23572247	23618785	1.00
vortex_q01/duckdb:parquet	9429780	9419889	1.00

---

No file size changes detected.

[myrrc]

I think this may need another iteration since we also want ordering in some sense but with parallel decoding

[onursatici]

Is there a race here with finish below? A concurrent finish could reset the handle after we call get but before vx_writer_push has finished using it, which would be UB, right?

[onursatici]

afaik this clone would create a slot for this new sender, increasing the channel capacity by 1. We do this clone on every push so even if I set concurrent_array_limit to 1 then call push 5 times I would have 6 in-flight arrays on the channel. So when cloning the sender we kind of defeat the intended backpressure

[onursatici]

so this means one push thread would consume the error, so all other threads that call push would get channel closed instead. I think that is fine, but also the thread that calls close would also not get the actual error, and get channel closed. I wonder if we should use Arc<Err> here to persist and report the same error for all push threads and whoever is calling close

[onursatici]

am I right in understanding that this method and the push method is the only place we do work on the runtime? If so I think we should change this. If we have a channel on push, the sender.send call would complete near immediately because it is not blocking as long as we have capacity on the channel. For any actual work to happen on the runtime, we need to call close, and that is only one thread.

The only time we can progress on multiple threads is if the sender.send blocks on push, which is super confusing.