DNM: upsert v2 hydration optimizations and payload-separated state

[DAlperin]

Motivation

DNM. Draft benchmark branch only. Upsert v2 hydration spends substantial work allocating and moving wide payloads through its spill and merge path. This branch combines the optimization experiments and payload-separated upsert prototype so CI publishes images for controlled comparisons. Optimization commits remain separate for review.

Description

• Bound serialized column chunks to 2 MiB, except indivisible oversized rows. Size merge reservations and output checks from observed row width.

• Bound pool insertion debt under concurrent enforcement. Encode admission-denied inserts directly to extents.

• Add enable_upsert_async_reads: offload upsert drain reads to the blocking executor, with at most eight queued or running reads per pool and cancellation-safe permit ownership. This flag controls drains and probes independently of merge scheduling.

• Add enable_upsert_async_merges: use Differential's pollable chunk merge pipeline and batcher with a vendored fueled spine for source-stash insertion and sealing, feedback batching, and feedback trace compaction. Pending reads preserve merge phases, remaining fuel, and published input batches. I/O wakeups resume the existing maintenance allowance without granting more fuel, so the arranger can return to input processing before starting another maintenance turn. The arranger prioritizes ready input and drains up to 32 queued events before optional maintenance, sealing the latest observed frontier. This avoids excess maintenance between data and progress events. Materialize continues to use its existing Differential trace interfaces. A private timestamp/frontier adapter connects maintenance to the newer DD/Timely dependency without changing stored timestamps or Timely operators. The DD polling APIs are pinned to 3659671246e0b5635a8128cc27d65f12ecf7dda5 in the fork. Only the spine scheduling implementation remains vendored. DNM-only dependency-policy exceptions allow the two pinned Git sources and duplicate DD/Timely versions. Merge reads yield to Timely and share a 256 MiB decoded-input budget across workers and sources. Trace replacement publishes atomically after completion and respects aggregated reader compaction holds. This path requires chunked upsert with payload separation disabled.

• Add enable_column_chunk_direct_compressed_output: serialize compressed generations directly to extents, avoiding resident admission and the intermediate serialized allocation. Younger identity-coded generations keep the existing path.

• Add enable_upsert_payload_stash: keep row bytes in shared pool-backed blocks and merge columnar keys, times, differences, and payload handles. Manifests retain referenced blocks through compaction and restashing. Feedback compares keys and fingerprints before resolving exact payload bytes, so equal updates cancel across physical allocations without a persistent interner. The bounded comparison cache reuses blocks across argument positions and within-block comparisons. Source encoding publishes only winners after offset consolidation, and output decoding groups reads by block within bounded probe windows.

• Defer optional virtual introductions while the async arranger has queued input. Active merges retain policy-funded maintenance, and the configured consolidation policy resumes when input drains.

• Preserve spilled chunks across timestamp advancement when stored time bounds prove no timestamp changes and a strict following key fence or final input proves the trailing group complete. Both synchronous and asynchronous chunk paths share the predicate. Async checks it before read admission and offload.

The flags default off in production and on in the test configuration. The payload path uses Materialize's existing Timely and Differential dependencies. The standalone local proxy-join harness remains separate and is not required to build these images.

This is a prototype, not a complete out-of-core runtime. Manifests and chunk summaries remain resident. Payload merge comparisons use synchronous reads into a two-block cache outside shared read admission. Probe histories and other temporary metadata still need bounds, payload decoded-read budgets are per dataflow, and partially live blocks retain dead bytes. Oversized rows retain an inline fallback. The design document describes the shared operator model and remaining boundaries.

Verification

Regression coverage exercises byte bounds, oversized rows, concurrent admission with stalled enforcement, resident and spilled round trips, async cancellation and concurrency limits, and direct extent accounting and codec equivalence. Shared payload tests cover ownership, admission, exact equality across independent locators and forced fingerprint collisions, timestamp compaction, block reclamation, and cache reuse across swapped arguments. A deterministic local microbenchmark compares unique keys, cancellation, replacement churn, forced collisions, and timestamp advancement against a canonical-handle control. It verifies exact output and block retirement, and records merge/advance timing and payload decompression counts separately.

The upsert operator comparison exercises paged, chunked, resumable, and payload-separated execution under forced spilling. Resumable batching is also compared against synchronous sealing across retractions, times, and re-stashing. Trace tests verify reader holds and that pending compaction keeps both input batches visible. Spine tests pause irregular fueled rollups to check visibility and continuation state. A maintenance microbenchmark polls the actual arranger maintenance future with controlled read completions and checks that ready and pending reads use the same work allowance, retain wakeups, and preserve published contents through eventual completion. A live operator comparison measures synchronous and async arrangements with independent sources sharing a pool, including paced and queued input. A correctness test checks complete output tuples, cancellations, future timestamps, closed input, and queues longer than the drain limit. DD owns the chunk and merger suspension tests. Timestamp-adapter tests cover incomparable frontiers, lattice operations, and path-summary composition and overflow. It covers inserts, highest-offset selection, deletes, persisted feedback, resume filtering, ineligible restashing, errors, multiple payload blocks, and oversized rows.

Hydration comparisons use a skewed 192.36 million row seed across 105 sources on eight workers, with 1,900 byte payloads and a 0.2 pool budget fraction. Each run retains its image identity, replica restart evidence, CPU, memory, extent traffic, and published-row work counters. Fixed observation windows distinguish work amplification from full hydration latency. The local operator fixture also measures output-frontier latency separately from completion at an equal final trace shape, using storage's maintenance proportionality of 1337 by default.

Release note: None (draft performance experiment).

Merge-work diagnostics and read admission

Instrument columnar merge and advancement work, commits, published batch sizes,
async input sizes, initial settlement, and native maintenance grants. Row-size
histograms and work counters distinguish changes in batch formation from extra
trace compaction and pool writes. Upsert records published batches while draining
its existing wakeup input, without introducing another dataflow operator.

The diagnostic AWS comparison uses the retained skewed seed and the same image
with async merges disabled and enabled. Each observation window is capped at
20 minutes and is expected to stop before full hydration. The metrics registry
has a registration/scrape test covering all stage labels and size buckets.

Async read admission now lasts for one loaded chunk primitive rather than across merger fuel yields. The merge kernel preserves pool-backed handles for untouched survivors while consuming temporary decoded inputs. Reader callbacks and external exertion queue maintenance for the owning arranger, preventing them from holding admission while that arranger waits for its batcher. Regression tests cover release at fuel yields, untouched pool-body preservation, and callback deferral. Shared-budget operator probes exercise many concurrent sources through complete output and final compaction.

storage_arrangement_exert_proportionality exposes the actual storage startup policy and preserves its existing default of 1337. The compute setting remains separate. Diagnostic batteries vary the actual storage policy and verify each pod's startup arguments before observation. The combined scheduling and unchanged-advancement candidate compares async and synchronous execution at the normal storage policy of 1337 using the same retained seed and image. Zero is an experimental control for optional maintenance, not a proposed production default or an equal-final-compaction performance claim.

Resident pool reads now copy inline when their state lock is available. Evicted,
oversize, and contended reads retain blocking-executor dispatch. Native merges
poll once before spawning their cancellation owner, so an immediately ready read
also avoids the outer task hop. Tests cover completion without a runtime,
contended-lock yielding, retained admission, and cancellation of offloaded reads.
An opt-in resident-read benchmark isolates dispatch overhead from full-operator
work. This change leaves storage maintenance policy and merge fuel unchanged.

Regression coverage for the scheduling guard checks that merge-only maintenance preserves separate batches, idle maintenance completes consolidation, and read readiness does not change the work allowance. Unchanged-advancement tests cover preserved pool handles, no offload for eligible cold chunks, shared boundary cancellation, incomplete trailing groups, and timestamp changes that require reconsolidation. Negative controls exercise the corresponding previous behavior.