storage: write the persist source's decode straight into columns

[antiguru]

decode_and_mfp wrote both sides of a persist read into one container as Results, so every
consumer of persist_source ran an ok_err demux, and a consumer that wanted columns paid a
second pass to re-encode the ok side. The decode operator gets a second output instead, with the
ok side generic over a container builder. The compute import asks for columns and gets them
straight from the decode; the storage sink and the materialized-view read-back ask for row
vectors and get the containers they had before, without the demux.

There is one decode implementation. record_time picks what a record stores for its time.
Compute passes |time| time.0, dropping the Subtime coordinate: it refines the capabilities so
flow control can pace parts within a millisecond, it stays on the capabilities either way, and no
reader of those streams distinguishes times that finely. persist_source_core passes
|time| time and keeps it, because upsert builds its collection inside the refined scope, then
joins the two sides back with map and concat. That is a move per record on a path that had
none, and it is the reason there is one decode operator rather than two.

Three commits, separable if you would rather review them apart:

• txn-wal: txns_progress_remap and txns_progress_frontiers are exposed separately, so the
  two streams share one subscription to the txns shard. txns_progress_frontiers becomes generic
  over the container it passes through.
• timely-util: an Enter impl for Column<(D, T, R)>, which a columnar collection needs to
  enter the iterative scope of a WITH MUTUALLY RECURSIVE dataflow.
• storage-operators and compute: the split itself.

On the table-import dataflow this removes the OkErr and VecToColumnar operators and four
Vec<(Result<Row, E>, (Timestamp, Subtime), Diff)> channels, visible in the change to
test/sqllogictest/introspection/relations.slt.

Not measured. The map/concat on the upsert rehydration read is the one place this adds work
rather than removing it, and delay_sources_past_rehydration exists because that path is
latency-sensitive, so it is worth a look.

🤖 Generated with Claude Code

[linear-code]

CPU-252

[antiguru]

This should go column-by-column, avoiding the reconstruction of the unchanged D and R columns.

[antiguru]

Done. A Column::Typed now moves its data and diff columns across whole and rebuilds only the times. A serialized column owns no typed sub-containers, so all three are rebuilt there, but one column at a time rather than record by record.

Worth noting which variant this path actually sees: the compute import's columns come from ConsolidatingColumnBuilder, whose flush_aligned mints Column::Align, so the serialized arm is the common one and the move only applies to the finish() case.

Posted by Claude Code.

[DAlperin]

why not just return serialized here and just pay the cost of materializing the time column

[antiguru]

Done, that works and it is better than what I had. The serialized arm now materializes only the time column and hands the data and diff columns to the output as borrowed views:

let (borrowed_data, borrowed_times, borrowed_diffs) = serialized.borrow();
let mut times = T2::Container::default();
for time in borrowed_times.into_index_iter() {
    times.push(&T2::to_inner(T1::into_owned(time)));
}
let view = (borrowed_data, times.borrow(), borrowed_diffs);
let words = indexed::length_in_words(&view);
let mut alloc: Vec<u64> = Vec::with_capacity(words);
indexed::encode(&mut alloc, &view);
Column::Align(alloc)

The result is Column::Align rather than Column::Typed, and data and diff cost one copy per column into the output allocation instead of a decode-and-re-encode per record. <(D, T2, R)>::Borrowed is just the tuple of the three borrowed sub-containers, so the mixed view encodes without any extra bounds.

That matters for this path in particular: the compute import's columns come from ConsolidatingColumnBuilder::flush_aligned, so Align is the variant that actually arrives here, and the Typed arm is the rarer finish() case. I kept the Typed arm because there data and diff move across with no copy at all.

Posted by Claude Code.