compute: Emit dataflow import outputs as the columnar edge

[antiguru]

Dataflow imports (persist sources, index/trace imports) emit the columnar edge via a vec_to_columnar leaf-encode at the import output. Includes the relations.slt golden update for the resulting VecToColumnar boundary operator.

Columnar dataflow-edge migration. Design doc: doc/developer/design/20260720_columnar_dataflow_edges.md (#37744).

Part of CPU-51.

[linear-code]

CPU-51

[def-]

QA LLM Review

1. MEDIUM -- Import leaf-encode is immediately undone for sink-only dataflows

src/compute/src/render.rs:483

SUBSCRIBE t and pass-through materialized views feed the import's raw collection straight into the sink, so the new unconditional vec_to_columnar at every import is followed immediately by the sink's into_vec() decode. Those dataflows now re-serialize every record into a Column and then re-materialize an owned Row per record, on both the snapshot and every subsequent update, for no benefit.

Details

EXPLAIN PHYSICAL PLAN FOR SUBSCRIBE t is just Source materialize.public.t with no Explained Query (test/sqllogictest/explain/subscribe.slt:101), i.e. no objects_to_build at all. export_sink therefore resolves Id::Global(t) to the import bundle itself and takes the bundle.collection branch at src/compute/src/render/sinks.rs:71, oks.clone().into_vec(). That was a no-op on a CollectionEdge::Vec; it now inserts a ColumnarToVec directly downstream of the VecToColumnar added at render.rs:483, and columnar_to_vec allocates a fresh Row per record via Columnar::into_owned.

The same shape covers SUBSCRIBE (SELECT a, b FROM t WHERE a > 5), where the filter is pushed into the source and the object plan is an identity Get::Collection that as_collection_core hands through unchanged (subscribe.slt:111), and CREATE MATERIALIZED VIEW mv AS SELECT * FROM t, whose Get with PassArrangements returns the import bundle as is. render.rs:678 (SnapshotMode::Exclude) is subscribe-only, since with_snapshot: false is set only by src/adapter/src/optimize/subscribe.rs:193, so a trivial SUBSCRIBE ... WITHOUT SNAPSHOT over an index round-trips too.

The broader point: nothing upstream of an import produces columns yet. This PR leaves the persist/trace decode untouched, persist_source::decode_and_mfp still ships Vec<(Result<Row, ...>, ...)>, and every consumer's Vec arm already reads rows by reference, so the two arms of arrange_collection (src/compute/src/render/context.rs:1251 and :1300) do identical per-record work. The encode is added work at every import rather than work moved earlier, and it exposes imports to the remaining into_vec() seams: temporal bucketing (context.rs:1157, :1193), delta-join update streams (join/delta_join.rs:731), LetRec bindings (render.rs:965, :1026).

Suggested fix: hold imports on CollectionEdge::Vec until the persist/trace decode emits Column directly, so the encode disappears instead of being paid and then undone. CollectionEdge::concat_many already encodes the conditional version of this rule, upgrading a Vec input only when a columnar peer exists; the import boundary has no such peer today. If the flip needs to land now to unblock the stack, gating it on the consumer side (only encode when the edge's reader is columnar-native) keeps the sink-only dataflows at zero conversions.