diff --git a/Cargo.lock b/Cargo.lock index 48f37d004ce8b..9dd7b1ac5a72e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6784,6 +6784,7 @@ dependencies = [ "scopeguard", "serde", "smallvec", + "static_assertions", "timely", "tokio", "tracing", diff --git a/doc/developer/design/20260720_two_runtime_compute/implementation-plan.md b/doc/developer/design/20260720_two_runtime_compute/implementation-plan.md new file mode 100644 index 0000000000000..2d05e4fb74528 --- /dev/null +++ b/doc/developer/design/20260720_two_runtime_compute/implementation-plan.md @@ -0,0 +1,375 @@ +# Two-runtime compute implementation plan + +> **For agentic workers:** REQUIRED SUB-SKILL: use `superpowers:subagent-driven-development` or `superpowers:executing-plans` to execute this plan task by task. +> Steps use checkbox (`- [ ]`) syntax for tracking. +> This is a *stacked-PR roadmap*, not a single bite-sized task list. +> The near-term slices (primitive fixes, Stage 1) are planned to executable granularity. +> The Stage 2 slices are planned to PR boundary and definition-of-done granularity, and each requires its own detailed sub-plan before execution, because several depend on open questions the design leaves to measurement. + +**Goal:** isolate latency-sensitive reads from CPU-bound maintenance in a compute replica, first by offloading the index-peek walk off the maintenance worker (Stage 1), then by running a second in-process timely runtime that serves reads against shared arrangements (Stage 2). + +**Architecture:** build on the Arc-batches branch so arrangement batches are `Arc`-backed and their contents are `Send + Sync`. +Stage 1 takes a `Send` point-in-time snapshot of a local arrangement on the maintenance worker and runs the cursor walk on an async task, mirroring the existing persist-peek offload. +Stage 2 adds a second timely runtime, a per-process publication registry, and a process-level command/response multiplexer, so reads execute on threads that are not maintenance workers. + +**Tech stack:** Rust, timely-dataflow, differential-dataflow (Materialize fork), `mz-compute`, `mz-compute-client`, `mz-clusterd`, `mz-row-spine`. + +## Global constraints + +* **Base branch.** All Materialize work stacks on `upstream/claude/spines-differential-arc-j93mho` (the Arc-batches branch, materialize #37743), *not* `main`. + All differential-fork work stacks on `antiguru/differential-dataflow@claude/spines-differential-arc-j93mho`. + Every file and line reference in this plan is to those branches. +* **Design source.** The design is PR #37747, `doc/developer/design/20260720_two_runtime_compute/README.md` and `doc/developer/design/20260719_shared_arrangements_across_runtimes/README.md`. + Where this plan and the design prose disagree, this plan wins, because it is grounded in the current fork code (see "Corrections to the design" below). +* **No `as_conversions`.** Use `mz_ore::cast::CastFrom` / `CastLossy`. +* **No `std::HashMap`.** Use `BTreeMap`. +* **No `unsafe`** in the sharing layer or the compute integration. + Any unsafe block requires a `SAFETY` comment, and the design forbids it here. +* **Feature-flag new behavior.** Every new peek or runtime path is gated behind a compute dyncfg flag, defaulted off in production and enabled in CI and test configs, following the project convention for new optimizer and compute flags. +* **Formatting and checks.** Run `bin/fmt` and `cargo check` before reporting any task done, and `bin/lint` plus `cargo clippy` before committing. +* **Squash-merge, sequential landing.** PRs squash-merge. + Land the stack bottom-up: primitive fixes, then Stage 1, then Stage 2 in order. + +## Corrections to the design + +Grounding against the current fork code found that the design's "Compaction: controller authority and required primitive changes" section is partly stale. +Two of the four primitive changes it lists are already implemented on the fork branch. + +* **Already done, verify only.** `snapshot_at` already enforces the `since <= t` gate (`sharing.rs:205-207`). + `batches_through` already fail-stops on a straddling batch with an assertion (`sharing.rs:286-289`), rather than silently including it. +* **Still open, this plan implements them.** The publisher still pins compaction (`sharing.rs:449`, forwarded at `499-500`, TODO at `430-438`). + `import` still does not assert equal peers (only a TODO at `sharing.rs:548-553`). +* **API differs from the design prose.** There is no bare `snapshot`, only `snapshot_at` (`sharing.rs:190`). + There is no `ReaderWakeup` enum. + Reader wakeup is a `Condvar upper_changed` field on `SharedTrace` (`sharing.rs:117`), with import wakeups driven through the replay queue's `SyncActivator`. + Plan against the real types. + +--- + +## Task graph and sequencing + +```mermaid +graph TD + P1[P1 publisher-no-pin fork] --> S2a[2a publication at export_index] + P2[P2 import equal-peers fork] --> S2f[2f interactive import path] + S1a[1a Send snapshot + assertion] --> S1b[1b index-peek offload] + S1a --> S2e[2e interactive peek path] + S2a --> S2b[2b registry + block-until-published] + S2b --> S2c[2c second runtime + globals] + S2c --> S2d[2d multiplexer] + S2d --> S2e + S2d --> S2f + S2c --> S2g[2g failure model] +``` + +Stage 1 (`1a`, `1b`) is independent of the fork fixes and of Stage 2. +It is the recommended first landing because it delivers a real latency win with no second runtime. +The fork fixes (`P1`, `P2`) gate the Stage 2 slices that consume the shared trace (`2a`, `2f`). + +--- + +## P1: publisher must not pin compaction (differential fork) + +**Repo:** `antiguru/differential-dataflow`, branch stacked on `claude/spines-differential-arc-j93mho`. +Clone into a worktree and repoint the Materialize `[patch.crates-io]` entry at the new branch when integrating. + +**Files:** +* Modify: `src/operators/arrange/sharing.rs:430-500` (the publisher's compaction-forwarding block). +* Test: add to the existing `sharing` test module (the module already runs a main and a query runtime as two `timely::execute` instances). + +**Interfaces:** +* Consumes: `TraceAgent::{get_logical_compaction, set_logical_compaction, set_physical_compaction}`, `compaction_target` (`sharing.rs:94`), `SharedTraceState.{logical_holds, physical_holds, since}` (`sharing.rs:60`). +* Produces: unchanged public API. + Behavior change only: publishing with no registered import holds no longer freezes the trace's `since`. + +**Rationale (from the design and the in-code TODO at `sharing.rs:430-438`):** +the publisher owns a `TraceAgent` clone whose counted hold in `TraceBox` currently sits at the publish-time `since` and never advances, because with no readers `compaction_target` returns that same `since` and `set_logical_compaction` re-applies it as a no-op. +So merely publishing an index freezes its compaction for the life of the index. +The fix: publishing carries no independent compaction floor. +The trace's writer and the controller drive `since`. +Only a live import's own `as_of` hold (released on drop) may hold the trace back, and the publisher forwards the meet of those holds when any exist. +The existing safety rule stays: never forward the destructive empty meet, because that permanently releases the publisher's read capability and makes `batches_through` panic. +With no holds, the publisher must let its agent follow the writer-driven `since` rather than re-pinning at publish time. + +- [ ] **Step 1: Write the failing test** — publish, no importers, writer advances `since`, assert the trace compacts. + +```rust +#[test] +fn publish_without_readers_does_not_pin_compaction() { + // Main runtime: arrange a collection, publish it, no importers. + // Advance the input and let the arrange operator seal past an initial time. + // Drive the writer's AllowCompaction (set_logical_compaction) forward on the + // source TraceAgent to `t`. + // Assert: after the publisher's next activation, the published `since` + // (state.since) has advanced to `t`, and a `snapshot_at(t0)` for a time + // `t0 < t` returns `None` (compacted away), where before the fix it stayed + // pinned at the publish-time since and `snapshot_at(t0)` still succeeded. +} +``` + +- [ ] **Step 2: Run it, verify it fails** on the current pinned behavior (`snapshot_at(t0)` still succeeds, `since` stuck at publish time). +- [ ] **Step 3: Implement.** In the compaction-forwarding block (`sharing.rs:449` onward), stop sourcing the floor from the publisher agent's own `get_logical_compaction`. + Compute `logical_target` as the meet of `state.logical_holds` when non-empty, else the writer-driven `since` the agent already reflects (do not re-introduce the publish-time floor). + Keep the "never forward the empty meet" guard. + Forward `logical_target` / `physical_target` to the agent at `499-500` as today. + Update the `430-438` TODO into a doc comment stating the contract: publishing pins nothing, holds come only from live imports. +- [ ] **Step 4: Run the full `sharing` test suite,** including the existing "read holds keep the main trace from compacting, and release on drop" test, and verify the new test passes and none regress. +- [ ] **Step 5: Commit** (`fix(sharing): publishing no longer pins compaction`). + +**DoD:** with no importers, publishing an index does not hold its `since`. +A registered import still holds its `as_of` and releases on drop. +The empty-meet guard is intact. +No test regresses. + +--- + +## P2: `import` asserts equal peers (differential fork) + +**Files:** +* Modify: `src/operators/arrange/sharing.rs:543-556` (`import`), and `SharedTrace` / `publish_named` (`sharing.rs:115-118, 373`) to record the publisher's peer count at publish time. +* Test: `sharing` test module. + +**Interfaces:** +* Consumes: `scope.peers()` on the importing scope, timely worker peer count at publish. +* Produces: `import` panics with a clear message when `scope.peers()` differs from the publisher's recorded peer count. + +**Rationale:** pairwise import (importer worker `i` reads publisher worker `i`) is sound only when both sides shard by `key.hashed() % peers` with equal *total* peers, which is `workers_per_process * num_processes`, not just workers-per-process. +The current code only has a TODO (`sharing.rs:548-553`). + +- [ ] **Step 1: Write the failing test** — publish on a runtime with `peers = A`, import on a runtime with `peers = B != A`, expect a panic. + +```rust +#[test] +#[should_panic(expected = "peers")] +fn import_asserts_equal_peers() { /* two timely::execute with different worker counts */ } +``` + +- [ ] **Step 2: Run it, verify it fails** (no panic today, silent wrong-shard read). +- [ ] **Step 3: Implement.** Record `peers` in `SharedTrace` at `publish_named`. + In `import`, `assert_eq!(scope.peers(), recorded_peers, "shared-trace import requires equal total peers")` before registration. +- [ ] **Step 4: Run the `sharing` suite,** verify the new test panics as expected and the equal-peer import tests still pass. +- [ ] **Step 5: Commit** (`fix(sharing): import asserts equal peers`). + +**DoD:** a mismatched-peer import fails loudly at import time. +Equal-peer imports are unaffected. + +--- + +## Stage 1: offload the index-peek walk (Materialize, no second runtime) + +Stage 1 mirrors the existing persist-peek async offload for fast-path index peeks. +The maintenance worker still takes the snapshot on its command loop (this is the design's acknowledged smaller-win first increment, the "offload only the walk" shape, which the design rejects *as a substitute for* Stage 2 but endorses as the Stage 1 stepping stone). +The win is removing the cursor walk (dominant for full scans and large results) from the worker's critical path. + +### Task 1a: `Send` point-in-time snapshot of a local arrangement + compile-time assertion + +**Files:** +* Create: `src/compute/src/compute_state/local_snapshot.rs` (new module for the owned, `Send` snapshot cursor over `Arc` batches taken from a local trace via `cursor_through`). +* Modify: `src/compute/src/compute_state.rs` (module declaration) and `src/compute/src/compute_state/peek_result_iterator.rs:25-39` (allow `PeekResultIterator` to be built over the owned snapshot cursor). +* Modify: `src/compute/src/compute_state/peek_stash.rs:45-47` (update the `!Send` note once the assertion holds). +* Test: unit test in `local_snapshot.rs` plus a `static_assertions::assert_impl_all!` for `Send`. + +**Interfaces:** +* Consumes: the local `TraceAgent`'s `cursor_through(upper)` returning `Arc`-backed batches (available on the Arc branch), `mz-row-spine` batch contents asserted `Send + Sync` (added by #37743). +* Produces: + * `pub struct LocalSnapshot { cursor: CursorList>, since: Antichain, upper: Antichain }`, `Send`. + * `pub fn snapshot_local(trace: &mut TraceBundle-or-handle, as_of: &Antichain) -> Result, SnapshotError>` taking the `Arc` batch chain via `cursor_through` and building an owned `CursorList`. + * `PeekResultIterator::new_over_snapshot(snapshot, mfp, timestamp, ...)` constructing the iterator over a `LocalSnapshot` instead of the borrowed `TraceCursor`. + +**Rationale:** the walk can only move to another thread if the whole iterator is `Send`. +Today it is `!Send` because the trace reader is `Rc` (`peek_stash.rs:45-47`). +On the Arc branch, batches are `Arc` and their contents are `Send + Sync`, so an *owned* `CursorList` over `Arc` batch cursors (not the `Rc` `TraceAgent`) is `Send`. +The snapshot must own the `Arc` batches so the maintenance runtime may merge and compact freely while the walk proceeds, with no torn read. + +- [ ] **Step 1: Write the failing compile-time assertion** in `local_snapshot.rs`. + +```rust +#[cfg(test)] +mod tests { + use static_assertions::assert_impl_all; + // Over the production Row/Timestamp/Diff layout. + assert_impl_all!(super::LocalSnapshot: Send); + assert_impl_all!(crate::compute_state::peek_result_iterator::PeekResultIterator: Send); +} +``` + +- [ ] **Step 2: Run it, verify it fails to compile** (`PeekResultIterator` is `!Send` today). +- [ ] **Step 3: Implement `LocalSnapshot` and `snapshot_local`.** + Take `cursor_through(as_of)` to obtain the covering `Arc` batches, build an owned `CursorList` of their cursors, capture `since`/`upper`. + Add `PeekResultIterator::new_over_snapshot` that holds the owned cursor and storage. + The assertion covers the batch cursor and the columnation containers, not just the batch wrapper, per the design's requirement. +- [ ] **Step 4: Run `cargo check -p mz-compute` and the assertion test,** verify `Send` holds. +- [ ] **Step 5: Commit** (`compute: Send point-in-time snapshot over Arc batches`). + +**DoD:** `LocalSnapshot` and a `PeekResultIterator` built over it are statically `Send`. +The snapshot owns its `Arc` batches, so concurrent maintenance merge and compaction cannot tear the read. + +**Risk:** if any columnation container in the production batch layout is not yet `Send + Sync` on the branch, the assertion fails and the gap must be closed in `mz-row-spine` first. +Fail loud at compile time, do not paper over with a wrapper. + +### Task 1b: index-peek async offload + +**Files:** +* Modify: `src/compute/src/compute_state.rs`: `handle_peek:674`, `process_peek:934` and the `seek_fulfillment` call site `981-987`, the `PendingPeek` enum, mirroring the persist offload at `1254-1326`. +* Modify: the compute dyncfg definitions (new flag `enable_index_peek_offload`). +* Test: `src/compute/src/compute_state.rs` unit coverage plus an integration test under `test/` exercising a fast-path index peek while the worker is busy. + +**Interfaces:** +* Consumes: `snapshot_local` and `PeekResultIterator::new_over_snapshot` from Task 1a, the `PendingPeek::persist` offload pattern (oneshot `result_tx`/`result_rx`, `mz_ore::task::spawn`, `activator.activate()`), the since-gate at `1536-1542`. +* Produces: a `PendingPeek::Index` async variant that takes the snapshot on the worker loop, spawns a task running the walk, and returns the `PeekResponse` over a oneshot, waking the worker via the activator. + +**Rationale:** the persist path (`compute_state.rs:1254-1326`) already spawns an async task, sends `PeekResponse` over a oneshot, and activates the worker on completion. +Index peeks take the fast path but run the walk synchronously on the worker (`process_peek` -> `seek_fulfillment`). +Stage 1 gives index peeks the same offload, now possible because the snapshot is `Send`. + +- [ ] **Step 1: Write the failing test.** Create an index, issue a fast-path peek, assert correct rows, and assert (via a busy-worker fixture or a timing/counter assertion) the walk ran off the worker step. Reuse the persist-peek test harness shape. +- [ ] **Step 2: Run it, verify it fails** (flag off / variant absent, walk still inline). +- [ ] **Step 3: Add the `enable_index_peek_offload` dyncfg,** default off in prod, on in CI/test config. +- [ ] **Step 4: Implement `PendingPeek::Index`.** + On the worker loop, apply the since-gate (`1536-1542`), take `snapshot_local`, spawn a task that runs `PeekResultIterator::new_over_snapshot` and produces `PeekResponse::Rows` / `PeekResponse::Error`, send over the oneshot, `activator.activate()`. + Hold an `abort_on_drop` handle so `CancelPeek` and drop abort the task, mirroring `PendingPeek::Persist`. + Gate the whole path on the flag, falling back to the current synchronous walk when off. +- [ ] **Step 5: Run the peek test suite** (`bin/sqllogictest --optimized` for the relevant files, plus the new integration test), verify correctness with the flag on and off. +- [ ] **Step 6: Commit** (`compute: offload fast-path index peek walk to an async task`). + +**DoD:** with the flag on, a fast-path index peek's walk runs off the worker step and returns the same rows as with the flag off. +The since-gate error is unchanged. +Cancel and drop abort the task. +The point-lookup latency is unchanged (correctly, per the design's non-criterion). + +**Open item for 1b:** whether the async task runs on the tokio pool (as persist does) or a dedicated reader pool is an implementation choice deferred to measurement. +Start with the tokio pool to match persist, and revisit under Stage 2. + +--- + +## Stage 2: second timely runtime serving reads + +Stage 2 slices are planned to PR boundary and DoD. +Each requires its own detailed sub-plan (via `superpowers:writing-plans`) before execution, because they are large and several touch open questions the design defers to measurement. +They land in the order below. + +### Task 2a: publication at `export_index` + +**Depends on:** P1. +**Files:** `src/compute/src/render.rs`: `export_index:688`, `ArrangementFlavor::Local` arm `713`, `traces.set` `734-737`, `ArrangementFlavor::Trace` re-export arm `739-743`, and `export_index_iterative:767` (Local arm `793`, `traces.set` `836-838`, Trace arm `841-845`). + +**Behavior:** in the `Local` arm, alongside `traces.set`, sink the arrange output stream into a `sharing::publish`-backed publication point and record a `Send` handle in the per-process registry (Task 2b) keyed by `GlobalId`, one entry per worker ordinal. +The `Trace` re-export arm re-registers the same published handle under the new id. +`export_index_iterative` mirrors this. +Publishing does not change what `traces.set` stores, so the local trace and its compaction are exactly as today. + +**DoD:** every maintained index publishes. +Local trace behavior and compaction are byte-for-byte unchanged. +Publication cost is one `Arc` clone plus a mutex push per batch, no per-record work. +With no importers, publishing does not pin compaction (relies on P1). + +### Task 2b: per-process publication registry + block-until-published handshake + +**Depends on:** 2a. +**Files:** new module `src/compute/src/sharing_registry.rs` (or similar), wired through `ComputeState::new` (`compute_state.rs:179-187`) and `server::serve` (`server.rs:85`) the way the persist client cache is shared (created once, handed to both `serve` calls). + +**Behavior:** a first-class per-process object keyed by `GlobalId`, one entry per worker ordinal, shared into both runtimes. +Interactive worker `i` finds maintenance worker `i`'s handle. +A read for an unregistered `GlobalId` *blocks* until the publication point appears, woken when it registers, rather than erroring or reading an empty snapshot. +This closes the reconciliation and replica-restart race where a `Peek` reaches the interactive side before the maintenance runtime has re-rendered and re-published. + +**DoD:** a peek for a not-yet-published id blocks and then resolves once published. +Reconciliation replay (`CreateDataflow` then `Peek` from history) is safe. +The handshake has no lost-wakeup window. + +### Task 2c: second "interactive" timely runtime + process-global reconciliation + +**Depends on:** 2b. +**Files:** `src/clusterd/src/lib.rs` (second `serve`/`ClusterSpec` instance, equal-peers assertion near `425-429`), `src/compute/src/compute_state.rs` (`DICTIONARY_COMPRESSION:497`, lgalloc/pager init `255-361`, metrics `538-541`). + +**Behavior:** one replica process runs two compute timely runtimes sharing per-process resources (persist client cache, tracing handle, metrics), exactly as storage and compute share them today. +Reconcile the three process-globals the design names. +`DICTIONARY_COMPRESSION`: both runtimes must agree, or it moves to per-runtime state. +lgalloc and pager config: initialized once per process, the second runtime must not re-initialize. +Metrics: add per-runtime labels to tell maintenance and interactive apart. +Assert *equal peers* (`workers_per_process * num_processes`), stronger than the current workers-per-process assertion at `clusterd:425-429`. + +**DoD:** two runtimes boot in one process. +Process-globals are initialized exactly once and agree. +Equal peers is asserted at configuration and fails loudly otherwise. +Metrics distinguish the two runtimes. + +### Task 2d: process-level command/response multiplexer + +**Depends on:** 2c. +**Files:** new component under `src/compute-client/src/` (explicitly *not* `PartitionedComputeState:79`, which merges homogeneous partitions with a meet). + +**Behavior:** a genuinely new component between the controller connection and the two runtimes, doing ownership-based demux and pass-through. +Route by the bespoke policy "maintained work to maintenance, one-shot reads to interactive": `Peek`, `CancelPeek` (keyed by `uuid`), and transient `CreateDataflow` / `AllowCompaction` to interactive; maintained `CreateDataflow` / `Schedule` / `AllowWrites` / `AllowCompaction` to maintenance; lifecycle (`CreateInstance`, `UpdateConfiguration`, `InitializationComplete`) to both. +Distinguish a subscribe (which exports transient ids but stays on maintenance) by the subscribe sink in its `DataflowDescription` (`dataflows.rs:348-356`). +Learn transient-id ownership by observing `CreateDataflow`. +Merge responses: pass `PeekResponse` through from whichever side produced it, report maintained `Frontiers` from maintenance, and guarantee exactly one `PeekResponse` per peek including for point-lookups (the cancel-race open question). + +**DoD:** the controller sees one endpoint and one protocol. +Every command reaches the correct runtime by the policy above. +Exactly one `PeekResponse` per peek, cancel races included. + +### Task 2e: interactive peek path off the registry + +**Depends on:** 2d, 1a. +**Files:** `src/compute/src/compute_state.rs`: replace the local `TraceManager` lookup in `handle_peek` (`674`, lookup at `678`) with a registry lookup for the target id on the interactive side. + +**Behavior:** the interactive side looks up the published handle, gates on published `upper`/`since`, takes a `Send` snapshot (a handful of `Arc` clones plus frontiers), and runs `PeekResultIterator` over it, reusing Stage 1's offload. +A snapshot peek holds nothing on the trace. + +**DoD:** a fast-path peek is served on the interactive runtime and never touches a maintenance worker thread. +Correctness and the since-gate are unchanged. +The snapshot pins no compaction. + +### Task 2f: interactive temporary-dataflow import path + +**Depends on:** 2d, P2. +**Files:** `src/compute/src/render.rs`: a new import path analogous to `import_index:591` that sources the imported `Arranged` from the registry handle via `sharing::import` (change-stream replay from the dataflow's `as_of`), instead of the local `TraceManager`. + +**Behavior:** a slow-path peek or ad-hoc query renders as a temporary dataflow on the interactive runtime, importing maintenance arrangements as ordinary `Arranged` collections, then dropping. +The import registers a real read hold at its `as_of`, released on drop. +Add the import-queue backpressure bound the design flags as an open question, because across runtimes the replay queue is no longer bounded by a shared worker step. + +**DoD:** a slow-path peek renders on the interactive runtime importing a maintenance arrangement, and drops cleanly. +The import holds only its own `as_of`, released on drop, and never below the controller's hold (relies on P1 and criterion 1). +The replay queue is bounded. + +### Task 2g: failure model confirmation + +**Depends on:** 2c. +**Files:** the process abort path in `clusterd` / `compute` server setup. + +**Behavior:** confirm a panic on any worker or reader thread of *either* runtime aborts the whole process, so an import's read hold and a half-served read cannot leak. +Shared fate is what makes the import hold safe without a lease-expiry mechanism. + +**DoD:** a panic on either runtime's worker or reader thread aborts the process. +There is no partial-failure path and no fallback that reroutes interactive reads onto the maintenance runtime. + +--- + +## Deferred to measurement or follow-up design (open questions) + +These are explicitly out of the executable plan and must not be silently resolved by an implementer. + +* **Core-sharing policy** between the two runtimes (pinning, priorities, oversubscription). + Worker counts are fixed by equal peers. +* **Reader pool vs tokio pool** for the walk (Task 1b / 2e), deferred to measurement. +* **Import-queue backpressure** bound and publisher backpressure (Task 2f), grows in importance if long-lived imports such as subscribes move to the interactive runtime. +* **Introspection and per-runtime memory attribution.** + Under one endpoint the controller cannot see which runtime holds what memory, and a shared arrangement reachable from both runtimes' size loggers can be double-counted. + `mz_arrangement_sizes` and friends need a per-runtime view eventually. +* **Cancellation and exactly-one response** semantics for point-lookups (folded into 2d, but the general contract may need its own design). +* **`batches_through` straddle:** already fail-stops (`sharing.rs:286-289`). + Confirm every `cursor_through` frontier that reaches a `SharedTraceHandle` is batch-aligned, so the assertion never trips on a legitimate read. +* **Lease expiry:** the design assumes same-process trust (RAII holds, shared fate). + Revisit only if a query thread stuck in a long computation holding compaction back becomes a real problem. + +## Self-review notes + +* **Spec coverage.** Every "New components this design requires" bullet maps to a task: multiplexer (2d), registry plus handshake (2b), interactive peek path (2e), interactive import path (2f), the two primitive changes (P1, P2 plus the verify-only note on `snapshot_at` and `batches_through`), and the `Send` assertion (1a). + The equal-peers requirement is 2c plus P2. + The process-global reconciliation is 2c. + The failure model is 2g. +* **Stale-design corrections** are called out up front so an implementer does not re-add an already-present gate or straddle guard. +* **Granularity.** P1, P2, 1a, and 1b are executable now. + 2a through 2g are PR boundaries with DoD and need per-task sub-plans before execution. diff --git a/src/compute-types/src/dyncfgs.rs b/src/compute-types/src/dyncfgs.rs index e1e8a96349d18..d17e89fe5f031 100644 --- a/src/compute-types/src/dyncfgs.rs +++ b/src/compute-types/src/dyncfgs.rs @@ -486,6 +486,18 @@ pub const MV_SINK_ADVANCE_PERSIST_FRONTIERS: Config = Config::new( "Whether the MV sink's write operator advances its internal persist frontiers to the as_of.", ); +/// Whether to offload a fast-path index peek's cursor walk to an async task, instead of running +/// it synchronously on the maintenance timely worker. +/// +/// Disabled by default while the offloaded path is validated; the synchronous walk remains the +/// default and is behavior-unchanged. +pub const ENABLE_INDEX_PEEK_OFFLOAD: Config = Config::new( + "enable_index_peek_offload", + false, + "Whether to offload a fast-path index peek's cursor walk to an async task, instead of \ + running it synchronously on the maintenance timely worker.", +); + /// Adds the full set of all compute `Config`s. pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet { configs @@ -535,6 +547,7 @@ pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet { .add(&COMPUTE_PROMETHEUS_INTROSPECTION_SCRAPE_INTERVAL) .add(&SUBSCRIBE_SNAPSHOT_OPTIMIZATION) .add(&MV_SINK_ADVANCE_PERSIST_FRONTIERS) + .add(&ENABLE_INDEX_PEEK_OFFLOAD) .add(&ENABLE_COLUMN_PAGED_BATCHER) .add(&ENABLE_COLUMN_PAGED_BATCHER_SPILL) .add(&COLUMN_PAGED_BATCHER_BUDGET_FRACTION) diff --git a/src/compute/Cargo.toml b/src/compute/Cargo.toml index eaed8bd235694..1bc6ecb345c67 100644 --- a/src/compute/Cargo.toml +++ b/src/compute/Cargo.toml @@ -46,6 +46,7 @@ prost.workspace = true scopeguard.workspace = true serde.workspace = true smallvec = { workspace = true, features = ["serde", "union"] } +static_assertions.workspace = true timely.workspace = true tokio.workspace = true tracing.workspace = true diff --git a/src/compute/src/compute_state.rs b/src/compute/src/compute_state.rs index c370724b0a622..008a33008510d 100644 --- a/src/compute/src/compute_state.rs +++ b/src/compute/src/compute_state.rs @@ -30,7 +30,7 @@ use mz_compute_client::protocol::response::{ }; use mz_compute_types::dataflows::DataflowDescription; use mz_compute_types::dyncfgs::{ - ENABLE_PEEK_RESPONSE_STASH, PEEK_RESPONSE_STASH_BATCH_MAX_RUNS, + ENABLE_INDEX_PEEK_OFFLOAD, ENABLE_PEEK_RESPONSE_STASH, PEEK_RESPONSE_STASH_BATCH_MAX_RUNS, PEEK_RESPONSE_STASH_THRESHOLD_BYTES, PEEK_STASH_BATCH_SIZE, PEEK_STASH_NUM_BATCHES, }; use mz_compute_types::plan::render_plan::RenderPlan; @@ -68,14 +68,16 @@ use tokio::sync::{oneshot, watch}; use tracing::{Level, debug, error, info, span, trace, warn}; use uuid::Uuid; -use crate::arrangement::manager::{TraceBundle, TraceManager}; +use crate::arrangement::manager::{PaddedTrace, TraceBundle, TraceManager}; use crate::logging; use crate::logging::compute::{CollectionLogging, ComputeEvent, PeekEvent}; use crate::logging::initialize::LoggingTraces; use crate::metrics::{CollectionMetrics, WorkerMetrics}; use crate::render::{LinearJoinSpec, StartSignal}; use crate::server::{ComputeInstanceContext, ResponseSender}; +use crate::typedefs::ErrAgent; +mod local_snapshot; mod peek_result_iterator; mod peek_stash; @@ -954,6 +956,14 @@ impl<'a> ActiveComputeState<'a> { let peek_stash_threshold_bytes = PEEK_RESPONSE_STASH_THRESHOLD_BYTES.get(&self.compute_state.worker_config); + // The peek stash's size-based diversion is only known once the walk has already + // produced enough rows to exceed its threshold, so a peek that's eligible for + // (and has) the stash always takes the synchronous path, which already knows how + // to divert to it; offload is only for peeks the stash would never apply to. + let enable_index_peek_offload = ENABLE_INDEX_PEEK_OFFLOAD + .get(&self.compute_state.worker_config) + && !(peek_stash_enabled && peek_stash_eligible); + let metrics = IndexPeekMetrics { seek_fulfillment_seconds: &self .compute_state @@ -984,17 +994,31 @@ impl<'a> ActiveComputeState<'a> { self.compute_state.max_result_size, peek_stash_enabled && peek_stash_eligible, peek_stash_threshold_bytes, + enable_index_peek_offload, + self.timely_worker, &metrics, ); - self.compute_state - .metrics - .index_peek_total_seconds - .observe(start.elapsed().as_secs_f64()); + // The offloaded walk records its own task-duration observation into this + // histogram once the offload completes (see `PendingPeek::IndexOffload` below), + // so skip it here to avoid double-counting offloaded peeks. + if !matches!(status, PeekStatus::Offloaded(_)) { + self.compute_state + .metrics + .index_peek_total_seconds + .observe(start.elapsed().as_secs_f64()); + } match status { PeekStatus::Ready(result) => Some(result), PeekStatus::NotReady => None, + PeekStatus::Offloaded(offload_peek) => { + let uuid = offload_peek.peek.uuid; + self.compute_state + .pending_peeks + .insert(uuid, PendingPeek::IndexOffload(offload_peek)); + return; + } PeekStatus::UsePeekStash => { let _span = span!(parent: &peek.span, Level::DEBUG, "process_stash_peek").entered(); @@ -1027,6 +1051,15 @@ impl<'a> ActiveComputeState<'a> { .observe(duration.as_secs_f64()); result }), + PendingPeek::IndexOffload(peek) => { + peek.result.try_recv().ok().map(|(result, duration)| { + self.compute_state + .metrics + .index_peek_total_seconds + .observe(duration.as_secs_f64()); + result + }) + } PendingPeek::Stash(stashing_peek) => { let num_batches = PEEK_STASH_NUM_BATCHES.get(&self.compute_state.worker_config); let batch_size = PEEK_STASH_BATCH_SIZE.get(&self.compute_state.worker_config); @@ -1208,6 +1241,8 @@ pub enum PendingPeek { /// A peek against an index that is being stashed in the peek stash by an /// async background task. Stash(peek_stash::StashingPeek), + /// A peek against an index whose cursor walk was offloaded to an async task. + IndexOffload(IndexOffloadPeek), } impl PendingPeek { @@ -1329,6 +1364,7 @@ impl PendingPeek { PendingPeek::Index(p) => &p.span, PendingPeek::Persist(p) => &p.span, PendingPeek::Stash(p) => &p.span, + PendingPeek::IndexOffload(p) => &p.span, } } @@ -1337,6 +1373,7 @@ impl PendingPeek { PendingPeek::Index(p) => &p.peek, PendingPeek::Persist(p) => &p.peek, PendingPeek::Stash(p) => &p.peek, + PendingPeek::IndexOffload(p) => &p.peek, } } } @@ -1487,6 +1524,27 @@ pub struct IndexPeek { span: tracing::Span, } +/// An in-progress index peek whose cursor walk was offloaded to an async task. +/// +/// Gated by [`ENABLE_INDEX_PEEK_OFFLOAD`]. Only constructed once `IndexPeek::seek_fulfillment` +/// has confirmed the peek is fulfillable (frontiers past `peek.timestamp`, compaction frontier +/// not past it) and it is not eligible for the peek response stash. Mirrors `PersistPeek`: the +/// walk runs inside a spawned task over an owned, `Send` snapshot (see `local_snapshot`), and +/// the result comes back over a oneshot channel. +/// +/// Note that `PendingPeek` intentionally does not implement or derive `Clone`, as each +/// `PendingPeek` is meant to be dropped after it's responded to. +pub struct IndexOffloadPeek { + pub(crate) peek: Peek, + /// A background task that walks the snapshot and produces the peek result. If we're no + /// longer interested in the results, we abort the task. + _abort_handle: AbortOnDropHandle<()>, + /// The result of the background task, eventually. + result: oneshot::Receiver<(PeekResponse, Duration)>, + /// The `tracing::Span` tracking this peek's operation. + span: tracing::Span, +} + /// Histogram metrics for index peek phases. /// /// This struct bundles references to the various histogram metrics used to @@ -1520,6 +1578,8 @@ impl IndexPeek { max_result_size: u64, peek_stash_eligible: bool, peek_stash_threshold_bytes: usize, + enable_offload: bool, + timely_worker: &TimelyWorker, metrics: &IndexPeekMetrics<'_>, ) -> PeekStatus { let method_start = Instant::now(); @@ -1547,12 +1607,20 @@ impl IndexPeek { .frontier_check_seconds .observe(method_start.elapsed().as_secs_f64()); - let result = self.collect_finished_data( - max_result_size, - peek_stash_eligible, - peek_stash_threshold_bytes, - metrics, - ); + // Once we reach this point the peek is known-fulfillable: the frontier and + // compaction-frontier checks above have passed. `enable_offload` hands the walk to an + // async task instead of running it inline (`collect_finished_data`); see + // `spawn_offloaded_walk`. + let result = if enable_offload { + PeekStatus::Offloaded(self.spawn_offloaded_walk(max_result_size, timely_worker)) + } else { + self.collect_finished_data( + max_result_size, + peek_stash_eligible, + peek_stash_threshold_bytes, + metrics, + ) + }; metrics .seek_fulfillment_seconds @@ -1573,44 +1641,76 @@ impl IndexPeek { // Check if there exist any errors and, if so, return whatever one we // find first. - let (mut cursor, storage) = self.trace_bundle.errs_mut().cursor(); + let (cursor, storage) = self.trace_bundle.errs_mut().cursor(); + if let Some(error) = Self::scan_errs_for_error::>>( + self.peek.target.id(), + self.peek.timestamp, + cursor, + storage, + ) { + return PeekStatus::Ready(error); + } + + metrics + .error_scan_seconds + .observe(error_scan_start.elapsed().as_secs_f64()); + + Self::collect_ok_finished_data( + &self.peek, + self.trace_bundle.oks_mut(), + max_result_size, + peek_stash_eligible, + peek_stash_threshold_bytes, + metrics, + ) + } + + /// Scans an errs cursor for any error at or before `peek_timestamp`, returning the first one + /// found (or `None`). + /// + /// Shared between the synchronous errs scan in `collect_finished_data` (cursor borrowed live + /// off the trace via `TraceReader::cursor`) and the offloaded walk in `offloaded_response` + /// (cursor from an owned, `Send` snapshot; see `local_snapshot`): a `(cursor, storage)` pair + /// looks the same to this scan either way. + fn scan_errs_for_error( + target_id: GlobalId, + peek_timestamp: Timestamp, + mut cursor: peek_result_iterator::TraceCursor, + storage: peek_result_iterator::TraceStorage, + ) -> Option + where + Tr: TraceReader, + for<'a> BatchCursor: Cursor< + Key<'a>: std::fmt::Display, + TimeGat<'a>: PartialOrder, + DiffGat<'a> = &'a Diff, + >, + { while cursor.key_valid(&storage) { let mut copies = Diff::ZERO; cursor.map_times(&storage, |time, diff| { - if time.less_equal(&self.peek.timestamp) { + if time.less_equal(&peek_timestamp) { copies += diff; } }); if copies.is_negative() { let error = cursor.key(&storage); error!( - target = %self.peek.target.id(), diff = %copies, %error, + target = %target_id, diff = %copies, %error, "index peek encountered negative multiplicities in error trace", ); - return PeekStatus::Ready(PeekResponse::Error(format!( + return Some(PeekResponse::Error(format!( "Invalid data in source errors, \ saw retractions ({}) for row that does not exist: {}", -copies, error, ))); } if copies.is_positive() { - return PeekStatus::Ready(PeekResponse::Error(cursor.key(&storage).to_string())); + return Some(PeekResponse::Error(cursor.key(&storage).to_string())); } cursor.step_key(&storage); } - - metrics - .error_scan_seconds - .observe(error_scan_start.elapsed().as_secs_f64()); - - Self::collect_ok_finished_data( - &self.peek, - self.trace_bundle.oks_mut(), - max_result_size, - peek_stash_eligible, - peek_stash_threshold_bytes, - metrics, - ) + None } /// Collects data for a known-complete peek from the ok stream. @@ -1632,15 +1732,12 @@ impl IndexPeek { DiffGat<'a> = &'a Diff, >, { - let max_result_size = usize::cast_from(max_result_size); - let count_byte_size = size_of::(); - // Cursor setup timing let cursor_setup_start = Instant::now(); // We clone `literal_constraints` here because we don't want to move the constraints // out of the peek struct, and don't want to modify in-place. - let mut peek_iterator = peek_result_iterator::PeekResultIterator::new( + let peek_iterator = peek_result_iterator::PeekResultIterator::new( peek.target.id().clone(), peek.map_filter_project.clone(), peek.timestamp, @@ -1652,6 +1749,47 @@ impl IndexPeek { .cursor_setup_seconds .observe(cursor_setup_start.elapsed().as_secs_f64()); + Self::drain_ok_iterator( + peek_iterator, + peek, + max_result_size, + peek_stash_eligible, + peek_stash_threshold_bytes, + Some(metrics), + ) + } + + /// Drains a [`peek_result_iterator::PeekResultIterator`] into a [`PeekStatus`], sorting and + /// truncating per `peek.finishing`. + /// + /// Shared between the synchronous walk (`collect_ok_finished_data`, iterator borrowed live + /// off the trace) and the offloaded walk (`offloaded_response`, iterator built over an + /// owned, `Send` snapshot; see `local_snapshot`): the accumulation logic is identical either + /// way. Only the synchronous walk records per-phase metrics: the offloaded walk runs inside + /// a spawned task, and threading `'static` histogram handles across that boundary isn't + /// worth it for phases this granular, so `metrics` is `None` there and only the overall + /// duration is recorded, at the call site that receives the task's result. + fn drain_ok_iterator( + mut peek_iterator: peek_result_iterator::PeekResultIterator, + peek: &Peek, + max_result_size: u64, + peek_stash_eligible: bool, + peek_stash_threshold_bytes: usize, + metrics: Option<&IndexPeekMetrics<'_>>, + ) -> PeekStatus + where + Tr: TraceReader, + for<'a> BatchCursor: Cursor< + Key<'a>: ExtendDatums + Eq, + KeyContainer: BatchContainer, + Val<'a>: ExtendDatums, + TimeGat<'a>: PartialOrder, + DiffGat<'a> = &'a Diff, + >, + { + let max_result_size = usize::cast_from(max_result_size); + let count_byte_size = size_of::(); + // Accumulated `Vec<(row, count)>` results that we are likely to return. let mut results = Vec::new(); let mut total_size: usize = 0; @@ -1701,17 +1839,21 @@ impl IndexPeek { if results.len() >= 2 * max_results { if peek.finishing.order_by.is_empty() { results.truncate(max_results); - metrics - .row_iteration_seconds - .observe(row_iteration_start.elapsed().as_secs_f64()); - metrics - .result_sort_seconds - .observe(sort_time_accum.as_secs_f64()); + if let Some(metrics) = metrics { + metrics + .row_iteration_seconds + .observe(row_iteration_start.elapsed().as_secs_f64()); + metrics + .result_sort_seconds + .observe(sort_time_accum.as_secs_f64()); + } let row_collection_start = Instant::now(); let collection = RowCollection::new(results, &peek.finishing.order_by); - metrics - .row_collection_seconds - .observe(row_collection_start.elapsed().as_secs_f64()); + if let Some(metrics) = metrics { + metrics + .row_collection_seconds + .observe(row_collection_start.elapsed().as_secs_f64()); + } return PeekStatus::Ready(PeekResponse::Rows(vec![collection])); } else { // We can sort `results` and then truncate to `max_results`. @@ -1742,20 +1884,441 @@ impl IndexPeek { } } - metrics - .row_iteration_seconds - .observe(row_iteration_start.elapsed().as_secs_f64()); - metrics - .result_sort_seconds - .observe(sort_time_accum.as_secs_f64()); + if let Some(metrics) = metrics { + metrics + .row_iteration_seconds + .observe(row_iteration_start.elapsed().as_secs_f64()); + metrics + .result_sort_seconds + .observe(sort_time_accum.as_secs_f64()); + } let row_collection_start = Instant::now(); let collection = RowCollection::new(results, &peek.finishing.order_by); - metrics - .row_collection_seconds - .observe(row_collection_start.elapsed().as_secs_f64()); + if let Some(metrics) = metrics { + metrics + .row_collection_seconds + .observe(row_collection_start.elapsed().as_secs_f64()); + } PeekStatus::Ready(PeekResponse::Rows(vec![collection])) } + + /// Takes an owned snapshot of the oks/errs traces and spawns a task to walk it off the + /// timely worker thread, mirroring `PendingPeek::persist`'s offload (oneshot + activator + + /// abort-on-drop). Only called once `seek_fulfillment`'s frontier and compaction-frontier + /// checks have already confirmed the traces are known-complete as of `self.peek.timestamp`, + /// so `snapshot_local`'s own since-gate can only fail here if that invariant were violated + /// between the two checks; on a single-threaded worker nothing runs in between, so this + /// cannot happen in practice, but we still handle it (rather than unwrap) defensively. + fn spawn_offloaded_walk( + &mut self, + max_result_size: u64, + timely_worker: &TimelyWorker, + ) -> IndexOffloadPeek { + // `snapshot_local` requires a batch-aligned `as_of`, i.e. one the trace has actually + // reached (see its doc comment); each trace's own current upper qualifies; `[]` (as used + // by the live cursor the synchronous path walks) does not, since it would make the + // snapshot's own since-gate vacuously true. The per-record filtering down to + // `peek.timestamp` still happens during the walk, exactly as it does synchronously. + let mut oks_upper = Antichain::new(); + self.trace_bundle.oks_mut().read_upper(&mut oks_upper); + let mut errs_upper = Antichain::new(); + self.trace_bundle.errs_mut().read_upper(&mut errs_upper); + + let oks_snapshot = local_snapshot::snapshot_local(self.trace_bundle.oks_mut(), &oks_upper); + let errs_snapshot = + local_snapshot::snapshot_local(self.trace_bundle.errs_mut(), &errs_upper); + + let activator = timely_worker.sync_activator_for([].into()); + let peek = self.peek.clone(); + let peek_uuid = peek.uuid; + let (result_tx, result_rx) = oneshot::channel(); + + let task_handle = mz_ore::task::spawn(|| "index_peek::offload", async move { + let start = Instant::now(); + let response = + Self::offloaded_response(&peek, max_result_size, oks_snapshot, errs_snapshot); + match result_tx.send((response, start.elapsed())) { + Ok(()) => {} + Err((_response, elapsed)) => { + debug!(duration =? elapsed, "dropping result for cancelled peek {peek_uuid}") + } + } + match activator.activate() { + Ok(()) => {} + Err(_) => { + debug!( + "unable to wake timely after completed offloaded index peek {peek_uuid}" + ); + } + } + }); + + IndexOffloadPeek { + peek: self.peek.clone(), + _abort_handle: task_handle.abort_on_drop(), + result: result_rx, + span: self.span.clone(), + } + } + + /// Builds the peek response from a pair of owned snapshots, off the timely worker thread. + /// + /// Mirrors `collect_finished_data` (errs scan, then the ok walk), but drains + /// `PeekResultIterator::new_over_snapshot` instead of a cursor borrowed live off the trace. + fn offloaded_response( + peek: &Peek, + max_result_size: u64, + oks_snapshot: Result, local_snapshot::SnapshotError>, + errs_snapshot: Result, local_snapshot::SnapshotError>, + ) -> PeekResponse + where + OksTr: TraceReader, + for<'a> BatchCursor: Cursor< + Key<'a>: ExtendDatums + Eq, + KeyContainer: BatchContainer, + Val<'a>: ExtendDatums, + TimeGat<'a>: PartialOrder, + DiffGat<'a> = &'a Diff, + >, + ErrsTr: TraceReader, + for<'a> BatchCursor: Cursor< + Key<'a>: std::fmt::Display, + TimeGat<'a>: PartialOrder, + DiffGat<'a> = &'a Diff, + >, + { + let target_id = peek.target.id(); + + let since_gate_error = || { + soft_panic_or_log!( + "index peek offload: snapshot compacted past readiness-checked timestamp {}", + peek.timestamp, + ); + PeekResponse::Error(format!( + "Arrangement compaction frontier is beyond the time of the attempted read ({})", + peek.timestamp, + )) + }; + let oks_snapshot = match oks_snapshot { + Ok(snapshot) => snapshot, + Err(_) => return since_gate_error(), + }; + let errs_snapshot = match errs_snapshot { + Ok(snapshot) => snapshot, + Err(_) => return since_gate_error(), + }; + + let (errs_cursor, errs_storage) = errs_snapshot.into_cursor(); + if let Some(error) = Self::scan_errs_for_error::( + target_id, + peek.timestamp, + errs_cursor, + errs_storage, + ) { + return error; + } + + let peek_iterator = peek_result_iterator::PeekResultIterator::new_over_snapshot( + target_id, + peek.map_filter_project.clone(), + peek.timestamp, + peek.literal_constraints.clone().as_deref_mut(), + oks_snapshot, + ); + + // `process_peek` only sets `enable_offload` for peeks that are neither peek-stash + // eligible nor peek-stash enabled, so `drain_ok_iterator`'s stash diversion never + // triggers here, and it never returns `NotReady` (that only comes from + // `seek_fulfillment`'s own frontier checks, not from draining an already-taken + // snapshot). + match Self::drain_ok_iterator(peek_iterator, peek, max_result_size, false, 0, None) { + PeekStatus::Ready(response) => response, + PeekStatus::UsePeekStash | PeekStatus::NotReady | PeekStatus::Offloaded(_) => { + unreachable!("offloaded index peeks always resolve to `PeekStatus::Ready`") + } + } + } +} + +#[cfg(test)] +mod index_peek_offload_tests { + use std::rc::Rc; + + use differential_dataflow::operators::arrange::TraceAgent; + use differential_dataflow::trace::{Builder, Description, Trace}; + use mz_expr::{MapFilterProject, RowSetFinishing}; + use mz_repr::{Datum, RelationDesc, SqlScalarType}; + use mz_row_spine::RowRowBuilder; + use mz_timely_util::columnation::ColumnationStack; + use timely::container::PushInto; + use timely::dataflow::operators::generic::OperatorInfo; + use timely::progress::Timestamp as _; + use uuid::Uuid; + + use super::*; + use crate::typedefs::{ErrAgent, ErrBuilder, ErrSpine, RowRowAgent, RowRowSpine}; + + fn row(x: i64) -> Row { + Row::pack_slice(&[Datum::Int64(x)]) + } + + /// A standalone, single-thread timely `Worker`, for its `sync_activator_for` alone (no + /// dataflow is ever built on it). Built directly rather than via `timely::execute_directly`, + /// since that helper requires its whole closure (and thus anything it captures, like our + /// `Rc`-based `IndexPeek`/`TraceBundle`) to be `Send + Sync`, which peek state never is. + fn standalone_worker() -> TimelyWorker { + let alloc = timely::communication::Allocator::Thread(Default::default()); + TimelyWorker::new(timely::WorkerConfig::default(), alloc, None) + } + + /// Builds a one-batch `[0, upper)` oks trace with `rows`, wrapped exactly like a real + /// index's `TraceBundle.oks` (a `PaddedTrace>`), but constructed directly + /// (bypassing rendering a dataflow) for test purposes. + /// + /// The batch is inserted through the `TraceWriter` (not `Trace::insert` on the bare spine + /// directly), because the writer tracks its own idea of the trace's current upper and + /// asserts new batches are contiguous with it; inserting straight into the spine before + /// wrapping desyncs that bookkeeping, and the writer's `Drop` (which seals the trace to the + /// empty frontier) then panics. Closing the trace this way is fine for a test snapshot: an + /// empty (fully closed) upper is readable at any finite peek timestamp. + fn oks_trace_with_rows( + upper: Timestamp, + rows: Vec<((Row, Row), Timestamp, Diff)>, + ) -> PaddedTrace> { + let spine: RowRowSpine = + Trace::new(OperatorInfo::new(0, 0, Rc::from(vec![0])), None, None); + let (agent, mut writer) = + TraceAgent::new(spine, OperatorInfo::new(1, 0, Rc::from(vec![0])), None); + + let description = Description::new( + Antichain::from_elem(Timestamp::minimum()), + Antichain::from_elem(upper), + Antichain::from_elem(Timestamp::minimum()), + ); + let mut chunk = ColumnationStack::default(); + for row in rows { + chunk.push_into(row); + } + let batch = RowRowBuilder::::seal(&mut vec![chunk], description); + writer.insert(batch, Some(Timestamp::minimum())); + + agent.into() + } + + /// Builds a one-batch `[0, upper)` errs trace with no errors, wrapped like a real index's + /// `TraceBundle.errs`. + fn errs_trace_empty(upper: Timestamp) -> PaddedTrace> { + let spine: ErrSpine = + Trace::new(OperatorInfo::new(2, 0, Rc::from(vec![0])), None, None); + let (agent, mut writer) = + TraceAgent::new(spine, OperatorInfo::new(3, 0, Rc::from(vec![0])), None); + + let description = Description::new( + Antichain::from_elem(Timestamp::minimum()), + Antichain::from_elem(upper), + Antichain::from_elem(Timestamp::minimum()), + ); + let chunk = ColumnationStack::default(); + let batch = ErrBuilder::::seal(&mut vec![chunk], description); + writer.insert(batch, Some(Timestamp::minimum())); + + agent.into() + } + + fn test_metrics(registry: &mz_ore::metrics::MetricsRegistry) -> crate::metrics::ComputeMetrics { + crate::metrics::ComputeMetrics::register_with(registry) + } + + fn make_peek(timestamp: Timestamp) -> Peek { + let result_desc = RelationDesc::builder() + .with_column("k", SqlScalarType::Int64.nullable(false)) + .with_column("v", SqlScalarType::Int64.nullable(false)) + .finish(); + Peek { + target: PeekTarget::Index { + id: GlobalId::User(1), + }, + result_desc, + literal_constraints: None, + uuid: Uuid::new_v4(), + timestamp, + finishing: RowSetFinishing::trivial(2), + map_filter_project: MapFilterProject::new(2) + .into_plan() + .expect("identity MFP plans") + .into_nontemporal() + .expect("identity MFP has no temporal filters"), + otel_ctx: OpenTelemetryContext::empty(), + } + } + + /// A fast-path index peek's cursor walk, offloaded via `ENABLE_INDEX_PEEK_OFFLOAD`, returns + /// the same rows as the synchronous walk, and `seek_fulfillment` hands the walk off to a + /// task rather than draining it inline: with the flag on, `seek_fulfillment` returns + /// `PeekStatus::Offloaded` without ever calling `collect_finished_data`, so the row walk + /// provably did not run during that (synchronous, worker-step) call; it only runs once the + /// spawned task is polled below. + #[mz_ore::test(tokio::test)] + #[cfg_attr(miri, ignore)] // differential-dataflow's Columnation isn't miri-clean + async fn offload_matches_synchronous_walk() { + let registry = mz_ore::metrics::MetricsRegistry::new(); + let metrics = test_metrics(®istry).for_worker(0); + let index_metrics = IndexPeekMetrics { + seek_fulfillment_seconds: &metrics.index_peek_seek_fulfillment_seconds, + frontier_check_seconds: &metrics.index_peek_frontier_check_seconds, + error_scan_seconds: &metrics.index_peek_error_scan_seconds, + cursor_setup_seconds: &metrics.index_peek_cursor_setup_seconds, + row_iteration_seconds: &metrics.index_peek_row_iteration_seconds, + result_sort_seconds: &metrics.index_peek_result_sort_seconds, + row_collection_seconds: &metrics.index_peek_row_collection_seconds, + }; + + let rows = vec![ + ((row(1), row(10)), Timestamp::new(1), Diff::ONE), + ((row(2), row(20)), Timestamp::new(2), Diff::ONE), + ((row(3), row(30)), Timestamp::new(3), Diff::ONE), + ]; + let peek_timestamp = Timestamp::new(5); + let trace_upper = Timestamp::new(10); + + // Flag off: the synchronous walk, run directly. + let mut off_peek = IndexPeek { + peek: make_peek(peek_timestamp), + trace_bundle: TraceBundle::new( + oks_trace_with_rows(trace_upper, rows.clone()), + errs_trace_empty(trace_upper), + ), + span: tracing::Span::none(), + }; + let mut upper = Antichain::new(); + let worker = standalone_worker(); + let off_status = off_peek.seek_fulfillment( + &mut upper, + u64::MAX, + false, + 0, + false, + &worker, + &index_metrics, + ); + let off_response = match off_status { + PeekStatus::Ready(response) => response, + _ => panic!("synchronous walk must resolve directly"), + }; + + // Flag on: `seek_fulfillment` hands the walk off instead of draining it inline. + let mut on_peek = IndexPeek { + peek: make_peek(peek_timestamp), + trace_bundle: TraceBundle::new( + oks_trace_with_rows(trace_upper, rows.clone()), + errs_trace_empty(trace_upper), + ), + span: tracing::Span::none(), + }; + let mut upper2 = Antichain::new(); + let on_status = on_peek.seek_fulfillment( + &mut upper2, + u64::MAX, + false, + 0, + true, + &worker, + &index_metrics, + ); + let offload_peek = match on_status { + PeekStatus::Offloaded(offload_peek) => offload_peek, + _ => panic!("offloaded walk must hand off to a task, not resolve directly"), + }; + let (on_response, _duration) = offload_peek + .result + .await + .expect("offloaded task sends a result before the sender is dropped"); + + assert_eq!( + off_response, on_response, + "flag on/off must return identical rows" + ); + } + + /// The since-gate error (a read whose timestamp has been compacted past) is identical + /// whether or not the offload flag is set: the check runs before `seek_fulfillment` decides + /// whether to offload, so both paths share the exact same error text. + #[mz_ore::test(tokio::test)] + #[cfg_attr(miri, ignore)] + async fn since_gate_error_unchanged_by_offload() { + let registry = mz_ore::metrics::MetricsRegistry::new(); + let metrics = test_metrics(®istry).for_worker(0); + let index_metrics = IndexPeekMetrics { + seek_fulfillment_seconds: &metrics.index_peek_seek_fulfillment_seconds, + frontier_check_seconds: &metrics.index_peek_frontier_check_seconds, + error_scan_seconds: &metrics.index_peek_error_scan_seconds, + cursor_setup_seconds: &metrics.index_peek_cursor_setup_seconds, + row_iteration_seconds: &metrics.index_peek_row_iteration_seconds, + result_sort_seconds: &metrics.index_peek_result_sort_seconds, + row_collection_seconds: &metrics.index_peek_row_collection_seconds, + }; + + // A peek at time 1, against a trace that has already compacted its logical frontier to + // time 5: the read is beyond the trace's compaction frontier. + let peek_timestamp = Timestamp::new(1); + let trace_upper = Timestamp::new(10); + + let make_compacted_bundle = || { + let mut bundle = TraceBundle::new( + oks_trace_with_rows(trace_upper, vec![]), + errs_trace_empty(trace_upper), + ); + let compacted = Antichain::from_elem(Timestamp::new(5)); + bundle.oks_mut().set_logical_compaction(compacted.borrow()); + bundle.errs_mut().set_logical_compaction(compacted.borrow()); + bundle + }; + + let mut off_peek = IndexPeek { + peek: make_peek(peek_timestamp), + trace_bundle: make_compacted_bundle(), + span: tracing::Span::none(), + }; + let mut upper = Antichain::new(); + let worker = standalone_worker(); + let off_status = off_peek.seek_fulfillment( + &mut upper, + u64::MAX, + false, + 0, + false, + &worker, + &index_metrics, + ); + + let mut on_peek = IndexPeek { + peek: make_peek(peek_timestamp), + trace_bundle: make_compacted_bundle(), + span: tracing::Span::none(), + }; + let mut upper2 = Antichain::new(); + let on_status = on_peek.seek_fulfillment( + &mut upper2, + u64::MAX, + false, + 0, + true, + &worker, + &index_metrics, + ); + + let (PeekStatus::Ready(off_response), PeekStatus::Ready(on_response)) = + (off_status, on_status) + else { + panic!("a compacted-past read must resolve directly (never offloaded) either way"); + }; + assert_eq!(off_response, on_response); + assert!( + matches!(&off_response, PeekResponse::Error(msg) if msg.contains("compaction frontier")), + "expected a compaction-frontier error, got {off_response:?}", + ); + } } /// For keeping track of the state of pending or ready peeks, and managing @@ -1769,6 +2332,9 @@ enum PeekStatus { UsePeekStash, /// The peek result is ready. Ready(PeekResponse), + /// The peek is fulfillable, and its cursor walk was handed off to an async task; see + /// `IndexPeek::spawn_offloaded_walk`. + Offloaded(IndexOffloadPeek), } /// The frontiers we have reported to the controller for a collection. diff --git a/src/compute/src/compute_state/local_snapshot.rs b/src/compute/src/compute_state/local_snapshot.rs new file mode 100644 index 0000000000000..2ff6b72e2d146 --- /dev/null +++ b/src/compute/src/compute_state/local_snapshot.rs @@ -0,0 +1,232 @@ +// Copyright Materialize, Inc. and contributors. All rights reserved. +// +// Use of this software is governed by the Business Source License +// included in the LICENSE file. + +//! An owned, `Send` point-in-time snapshot of a local arrangement. +//! +//! [`snapshot_local`] and [`LocalSnapshot::into_cursor`] back the offloaded fast-path index +//! peek's cursor walk (`crate::compute_state::IndexOffloadPeek::spawn_offloaded_walk`), which +//! runs on its own async task off the maintenance worker's thread. `#[allow(dead_code)]` remains +//! for `since`/`upper`, which today are exercised only by this module's `Send` assertion tests. +#![allow(dead_code)] + +use differential_dataflow::trace::{Navigable, TraceReader}; +use timely::order::PartialOrder; +use timely::progress::Antichain; + +use crate::compute_state::peek_result_iterator::{TraceCursor, TraceStorage}; + +/// An owned, `Send` point-in-time snapshot of a local arrangement, taken from a trace's +/// `Arc`-backed batches via [`TraceReader::cursor_through`]. +/// +/// A cursor borrowed straight off a live [`TraceReader`] stays tied to that reader for its +/// whole walk; for the traces this crate maintains, the reader is `Rc`-based and so `!Send`. +/// This snapshot instead owns the covering batches outright (an owned [`CursorList`] over their +/// `Arc` cursors), so it does not borrow from the trace at all and is `Send`. The +/// maintenance worker can keep merging and compacting the trace while the snapshot's cursor is +/// walked elsewhere: the owned `Arc` batches keep their contents alive and unchanged regardless +/// of what the trace does to its own bookkeeping afterwards, so there is no torn read. +/// +/// [`CursorList`]: differential_dataflow::trace::cursor::CursorList +pub struct LocalSnapshot +where + Tr: TraceReader, +{ + cursor: TraceCursor, + storage: TraceStorage, + since: Antichain, + upper: Antichain, +} + +impl LocalSnapshot +where + Tr: TraceReader, +{ + /// The trace's logical compaction frontier at the time the snapshot was taken. + /// + /// Times not beyond this frontier may have been coalesced together; see + /// [`TraceReader::get_logical_compaction`]. + pub fn since(&self) -> &Antichain { + &self.since + } + + /// The frontier through which the snapshot's batches are known to be complete. + /// + /// This is the `as_of` passed to [`snapshot_local`]: the snapshot's cursor covers exactly + /// the updates at times not beyond it. + pub fn upper(&self) -> &Antichain { + &self.upper + } + + /// Consumes the snapshot, returning its owned cursor and backing storage. + /// + /// For building a [`PeekResultIterator`] directly over the snapshot's already-owned + /// batches, instead of a cursor borrowed live from the trace. + /// + /// [`PeekResultIterator`]: crate::compute_state::peek_result_iterator::PeekResultIterator + pub(crate) fn into_cursor(self) -> (TraceCursor, TraceStorage) { + (self.cursor, self.storage) + } +} + +/// Errors that can occur when taking an owned snapshot of a local trace at a point in time. +#[derive(Debug, PartialEq, Eq)] +pub enum SnapshotError { + /// The trace's logical compaction frontier has advanced past `as_of`, so the batches + /// backing a snapshot at that time could no longer distinguish updates at or before it. + CompactedPast, + /// The trace could not produce a covering, batch-aligned cursor through `as_of`. Callers + /// should only pass an `as_of` that is a frontier the trace has actually reached, e.g. one + /// obtained from [`TraceReader::read_upper`]. + NotYetAvailable, +} + +/// Takes an owned, `Send` snapshot of `trace`'s contents as of `as_of`. +/// +/// `as_of` must be a frontier the trace has actually reached (for example one obtained from +/// [`TraceReader::read_upper`]); it is passed directly to [`TraceReader::cursor_through`], which +/// requires a batch-aligned upper and panics otherwise. The returned snapshot owns the +/// `Arc`-backed batches covering `as_of`, so it stays valid and `Send` independent of `trace` +/// and can be handed to another thread to walk. +pub fn snapshot_local( + trace: &mut Tr, + as_of: &Antichain, +) -> Result, SnapshotError> +where + Tr: TraceReader, +{ + let since = trace.get_logical_compaction().to_owned(); + if !PartialOrder::less_equal(&since, as_of) { + return Err(SnapshotError::CompactedPast); + } + + let (cursor, storage) = trace + .cursor_through(as_of.borrow()) + .ok_or(SnapshotError::NotYetAvailable)?; + + Ok(LocalSnapshot { + cursor, + storage, + since, + upper: as_of.clone(), + }) +} + +#[cfg(test)] +mod tests { + use std::rc::Rc; + + use differential_dataflow::trace::{Builder, Cursor, Description, Trace}; + use mz_repr::{Datum, Diff, Row, Timestamp}; + use mz_row_spine::RowRowBuilder; + use mz_timely_util::columnation::ColumnationStack; + use static_assertions::assert_impl_all; + use timely::PartialOrder; + use timely::container::PushInto; + use timely::dataflow::operators::generic::OperatorInfo; + use timely::progress::{Antichain, Timestamp as _}; + + use super::snapshot_local; + use crate::typedefs::RowRowSpine; + + // Over the production Row/Timestamp/Diff layout. Proves that an owned snapshot cursor over + // `Arc` batches is `Send`, and that a `PeekResultIterator` built over it is too, so the walk + // can move to a thread other than the maintenance worker (task 1b). + assert_impl_all!(super::LocalSnapshot>: Send); + assert_impl_all!( + crate::compute_state::peek_result_iterator::PeekResultIterator< + RowRowSpine, + >: Send + ); + + fn row(x: i64) -> Row { + Row::pack_slice(&[Datum::Int64(x)]) + } + + /// Builds a one-batch `RowRowSpine` with two rows at different times, `[0, upper)`. + fn spine_with_rows( + upper: Timestamp, + rows: Vec<((Row, Row), Timestamp, Diff)>, + ) -> RowRowSpine { + let operator = OperatorInfo::new(0, 0, Rc::from(vec![0])); + let mut trace: RowRowSpine = Trace::new(operator, None, None); + + let lower = Antichain::from_elem(Timestamp::minimum()); + let upper = Antichain::from_elem(upper); + let description = Description::new( + lower, + upper.clone(), + Antichain::from_elem(Timestamp::minimum()), + ); + + let mut chunk = ColumnationStack::default(); + for row in rows { + chunk.push_into(row); + } + let batch = RowRowBuilder::::seal(&mut vec![chunk], description); + Trace::insert(&mut trace, batch); + + trace + } + + #[mz_ore::test] + fn snapshot_reports_since_and_upper() { + let mut trace = spine_with_rows( + Timestamp::new(10), + vec![((row(1), row(2)), Timestamp::new(1), Diff::ONE)], + ); + let as_of = Antichain::from_elem(Timestamp::new(10)); + + let snapshot = snapshot_local(&mut trace, &as_of).expect("snapshot should succeed"); + + assert_eq!(snapshot.upper(), &as_of); + assert_eq!( + snapshot.since(), + &Antichain::from_elem(Timestamp::minimum()) + ); + } + + /// A snapshot taken with `as_of` covering the whole batch owns updates at every time in the + /// batch; filtering those updates down to a chosen time (as `PeekResultIterator` does via + /// `map_times`) yields only the rows visible as of that time. + #[mz_ore::test] + fn snapshot_returns_expected_rows_at_chosen_time() { + let mut trace = spine_with_rows( + Timestamp::new(10), + vec![ + ((row(1), row(2)), Timestamp::new(1), Diff::ONE), + ((row(3), row(4)), Timestamp::new(5), Diff::ONE), + ], + ); + let as_of = Antichain::from_elem(Timestamp::new(10)); + + let snapshot = snapshot_local(&mut trace, &as_of).expect("snapshot should succeed"); + let (mut cursor, storage) = snapshot.into_cursor(); + + let peek_time = Timestamp::new(3); + let mut visible = Vec::new(); + while cursor.key_valid(&storage) { + while cursor.val_valid(&storage) { + let key: Vec = cursor.key(&storage).into_iter().collect(); + let val: Vec = cursor.val(&storage).into_iter().collect(); + let mut copies = Diff::ZERO; + cursor.map_times(&storage, |time, diff| { + if time.less_equal(&peek_time) { + copies += diff; + } + }); + if !copies.is_zero() { + visible.push((key, val, copies)); + } + cursor.step_val(&storage); + } + cursor.step_key(&storage); + } + + assert_eq!( + visible, + vec![(vec![Datum::Int64(1)], vec![Datum::Int64(2)], Diff::ONE)] + ); + } +} diff --git a/src/compute/src/compute_state/peek_result_iterator.rs b/src/compute/src/compute_state/peek_result_iterator.rs index fc38f024e0c79..cd41e65682e6b 100644 --- a/src/compute/src/compute_state/peek_result_iterator.rs +++ b/src/compute/src/compute_state/peek_result_iterator.rs @@ -15,9 +15,15 @@ use differential_dataflow::trace::{Cursor, Navigable, TraceReader}; /// The merged cursor a [`TraceReader::cursor`] hands out over all of a trace's batches: a /// [`CursorList`] over the per-batch cursors. -type TraceCursor = CursorList>; +/// +/// Shared with [`crate::compute_state::local_snapshot`], whose owned [`LocalSnapshot`] holds the +/// same cursor/storage shape, but over batches taken once via `cursor_through` rather than +/// borrowed live from the trace. +/// +/// [`LocalSnapshot`]: crate::compute_state::local_snapshot::LocalSnapshot +pub(crate) type TraceCursor = CursorList>; /// Backing storage for a [`TraceCursor`]: the batches the cursor borrows from. -type TraceStorage = Vec<::Batch>; +pub(crate) type TraceStorage = Vec<::Batch>; use mz_ore::result::ResultExt; use mz_repr::fixed_length::ExtendDatums; use mz_repr::{DatumVec, Diff, GlobalId, Row, RowArena}; @@ -143,6 +149,37 @@ where } } + /// Builds a [`PeekResultIterator`] over an owned [`LocalSnapshot`], instead of a cursor + /// borrowed live from the trace. + /// + /// Unlike [`Self::new`], this does not need a `&mut Tr` for the walk: the snapshot already + /// owns its cursor and backing `Arc` batches, so the returned iterator is `Send` and can + /// run on a thread other than the one maintaining the trace. + /// + /// [`LocalSnapshot`]: crate::compute_state::local_snapshot::LocalSnapshot + pub fn new_over_snapshot( + target_id: GlobalId, + map_filter_project: mz_expr::SafeMfpPlan, + peek_timestamp: mz_repr::Timestamp, + literal_constraints: Option<&mut [Row]>, + snapshot: crate::compute_state::local_snapshot::LocalSnapshot, + ) -> Self { + let (mut cursor, storage) = snapshot.into_cursor(); + let literals = literal_constraints + .map(|constraints| Literals::new(constraints, &mut cursor, &storage)); + + Self { + target_id, + cursor, + storage, + map_filter_project, + peek_timestamp, + row_builder: Row::default(), + datum_vec: DatumVec::new(), + literals, + } + } + /// Returns `true` if the iterator has no more literals to process, or if there are no literals at all. fn literals_exhausted(&self) -> bool { self.literals.as_ref().map_or(false, Literals::is_exhausted) diff --git a/src/compute/src/compute_state/peek_stash.rs b/src/compute/src/compute_state/peek_stash.rs index 214d9f371f64f..b2b5fd51c3dd9 100644 --- a/src/compute/src/compute_state/peek_stash.rs +++ b/src/compute/src/compute_state/peek_stash.rs @@ -45,6 +45,14 @@ pub struct StashingPeek { /// We can't give a PeekResultIterator to our async upload task because the /// underlying trace reader is not Send/Sync. So we need to use a channel to /// send result rows from the worker thread to the async background task. + /// + /// The `peek_iterator` above is built with `PeekResultIterator::new`, which borrows the + /// cursor live off `trace_bundle` (a `PaddedTrace>`, `Rc`-based and + /// `!Send`), so `pump_rows` has to keep walking it on this worker thread. On this branch + /// batches are `Arc`-backed, so an owned snapshot cursor over them (see + /// `crate::compute_state::local_snapshot::LocalSnapshot`, taken via `snapshot_local` and + /// consumed by `PeekResultIterator::new_over_snapshot`) IS `Send`: a walk built that way + /// could move into the async task directly, without this channel. rows_tx: Option, String>>>, /// The result of the background task, eventually. pub result: oneshot::Receiver<(PeekResponse, Duration)>, diff --git a/src/compute/src/typedefs.rs b/src/compute/src/typedefs.rs index af9c867ec0586..7910a901d582c 100644 --- a/src/compute/src/typedefs.rs +++ b/src/compute/src/typedefs.rs @@ -32,12 +32,12 @@ pub(crate) mod spines { use std::sync::Arc; use columnation::Columnation; + use differential_dataflow::trace::arc_blanket_impls::ArcBuilder; use differential_dataflow::trace::implementations::ord_neu::{ OrdKeyBatch, OrdKeyBuilder, OrdValBatch, OrdValBuilder, }; use differential_dataflow::trace::implementations::spine_fueled::Spine; use differential_dataflow::trace::implementations::{Layout, Update}; - use differential_dataflow::trace::arc_blanket_impls::ArcBuilder; use mz_timely_util::columnation::ColumnationStack; use mz_row_spine::OffsetOptimized;