writer: new RowChunk column-write API; rebuild the encode pool

[rgankema]

Summary

Two independent changes ship together because they share the same FFI seam: a redesigned column-write API on the Julia side, and a rebuilt encode pool on the Rust side.

Column-write API: RowChunk + append! + flush!

Main exposes two parallel column-write paths — the low-level ColumnBatch / ColumnDescriptor surface and the high-level "gathered" surface (GatheredBatch, GatheredColumn, SliceRef, GatheredColumnDescriptor, add_slice!, add_string_slice!) — both eventually calling one of three write_columns overloads. This PR replaces all of that with a new design built around a single user-facing Julia type, RowChunk, and two methods, push! and append!, plus an explicit flush!.

The biggest shape change is who picks batch boundaries. On main, each write_columns(writer, batch) call is one Arrow RecordBatch: the caller's slice cadence and the encoder's row-group cadence are forced to coincide. In this
PR the writer owns an embedded RecordBatchBuilder, callers hand in arbitrary upstream slices via append!(writer, chunk), and the writer auto-flushes a RecordBatch to the encode pool when the accumulated row count reaches its coalesce window. flush!(writer) is the explicit escape hatch for logical boundaries (end of transaction, time tick) where a row-group break is desired sooner than the natural window.

Column types no longer have to be supplied per call. The builder derives them from the table's Arrow schema at writer construction, so callers drop the COLUMN_TYPE_* enum values they previously passed through add_slice!, add_string_slice!, and the ColumnBatch push overloads.

The push! signatures are also broader than add_slice! / add_string_slice!:

• Strings now accept sel for scattered access. On main, add_string_slice! rejected selection vectors and required the caller to pre-build str_ptrs / str_lens arrays with any selection already applied; the new push! does the gather internally.
• Strings now accept any AbstractVector{<:AbstractString} — Arrow's VariableSizeString, SubString{String}, etc. — without materializing through Vector{String}.
• sel accepts any contiguous AbstractVector{Int64}, not just Vector{Int64}, so callers can pass view(sel_buf, 1:n) when their sel buffer carries stale capacity beyond the live region.

Streaming pipelines reuse the same RowChunk across iterations via empty!. That clears the working vectors but retains the chunk's internal string-gather pool, so once the pool is sized to the steady-state slice width, subsequent empty!/push!/append! cycles allocate nothing on the string path. On main the equivalent loop allocated a fresh str_ptrs / str_lens pair per add_string_slice! call.

Source-aligned validity semantics

Validity bitmaps on main were output-aligned: when the caller used sel, they were responsible for gathering the source validity through sel themselves and producing a bitmap whose length matched the output row count.

This PR flips the contract: length(validity) >= length(data), and bit i describes whether data[i] is valid. This matches the natural shape of NullableVector, Vector{Union{T,Missing}}, and Arrow nullable arrays. When sel is also supplied, the library gathers validity alongside the value gather — numerics gather in Rust as part of the existing value gather; strings gather in Julia (where the value gather already lives, since Rust can't walk an AbstractString vector across the FFI) and materialize the output-aligned bitmap into a per-column scratch buffer that's reused across empty!.

Encode pool: per-writer FIFO + work-stealing async tasks

On main the encode pool is N OS threads (N = available_parallelism) pulling from a single MPMC channel, with per-writer ordering enforced by an Arc<Mutex<ConcreteDataFileWriter>> inside WriterState. That design has head-of-line blocking baked in: when two tasks for the same writer land adjacently in the global queue, the second worker takes the task and then blocks on the writer's mutex even when tasks for other idle writers are sitting behind it in the queue.

This PR moves to per-writer FIFO queues drained by a work-stealing pool. Each WriterState now owns a Mutex<VecDeque<RecordBatch>> plus atomic queue_len and busy flag. Workers scan the active-writer snapshot, skip writers with empty queues, CAS busy: false → true to claim one, then drain it. Wakeups use a single shared Notify with two safeguards against collapsed-notification races: the worker that wins a claim cascades a wake to a peer before draining, and after releasing busy it re-checks queue_len and notifies again if a producer pushed during the release window.

The workers themselves moved from OS threads to async tokio tasks. The old loop ran handle.block_on(w.write(batch)), holding the runtime thread for the duration of the S3 PUT. With async tasks the runtime thread is free during the I/O await and can drive another writer's encode in the meantime, hiding S3 latency across writers without disturbing per-writer FIFO ordering. The default worker count moved from available_parallelism to 2 * available_parallelism to keep cores busy with encode CPU from one task while others await I/O.

In-flight encodes can now observe writer shutdown: iceberg_writer_free flips a poisoned flag and the task drops the writer rather than restoring it. The old catch_unwind + manual pending bookkeeping is replaced by a PendingGuard Drop impl that decrements exactly once on normal return or panic.

Other changes

LZ4_RAW = 5 compression codec added as a separate enum value mapping to Compression::LZ4_RAW (modern raw blocks, parquet spec codec ID 7). The existing LZ4 = 3 keeps mapping to Compression::LZ4 — the legacy Hadoop-framed variant whose per-page framing overhead is the usual reason LZ4 underperforms Snappy in parquet benchmarks — for backward compatibility.

🤖 Generated with Claude Code

[gbrgr]

Wondering here: Can users always know the schema order this one expects? They I guess have to use it from the stored arrow schema? I am wondering whether retrieving the table schema multiple times always gives the same schema order, or whether that could potentially be a source of bugs.

[rgankema]

I (and Claude) assumed the schema order is fixed. I have nothing to back that up, but it would be surprising if it weren't, no?

[gbrgr]

Just wondering, because usually schema information is usually returned always as some JSON/dict like object, and not sure whether the iceberg spec specifies a column order?

[gbrgr]

I btw do not think it is a big issue in any case, because callers can just rely on the schema they see in the writer

[gbrgr]

Could you ask Claude whether here it would make sense to use spawn_blocking instead, as encoding is CPU-bound?

[rgankema]

Non-blocking spawn was the point. The encoder tasks don't just encode, they also write to disk and they block on that. One of the things this PR does is creating more tasks than threads so that an encoder task can take over while another encoder task is blocked on fsync.

[gbrgr]

LGTM