Skip to content

feat(precompute): build stored summary panes from Remote Write samples - #820

Draft
zzylol wants to merge 1 commit into
feat/sdsfrom
feat/precompute-engine
Draft

zzylol wants to merge 1 commit into
feat/sdsfrom
feat/precompute-engine

Conversation

@zzylol

@zzylol zzylol commented Oct 7, 2026 •

Copy link
Copy Markdown
Contributor

Stack 3/5. Base: #819. Next: #821.

Why

Remote Write samples were accepted and dropped, so no stored summary was ever built.

What

A precompute engine on the engine thread builds stored panes from Remote Write:

  • Selection: samples are buffered per definition and pane. The plan's own Scan selects them, so its label matchers apply.
  • Building: a pane (end − width, end] (left-open, like PromQL ranges and the executor's windows) is built once the watermark (newest sample − allowed lateness) passes end. The build runs the definition's builder, compiled from the DAG with asap_executor::physical_planner::compile, under ingestion scope, and stores the summary batch. Panes without samples are stored empty.
  • Lateness and startup: a sample arriving after its pane was built is dropped and counted. The first pane after startup is skipped because it may be partial.
  • Eviction: panes older than the largest offset plus two widths are evicted.
  • /api/v1/write returns 503 when the engine queue is full, so senders retry.
  • data_plane --allowed-lateness-ms (default 15 s).

How

The Planner IR holds Rcs, so one engine thread owns the plan, the precompute engine and the store; HTTP handlers send it commands over a bounded channel.

Key code interfaces added / modified

// data_plane/src/precompute.rs
pub struct RawSample {
    /// The series' full label set, including `__name__`.
    pub labels: Arc<BTreeMap<String, String>>,
    /// Sample time.
    pub timestamp_ms: i64,
    /// Sample value.
    pub value: f64,
}

pub struct PrecomputeStats {
    /// Samples buffered for at least one pane.
    pub accepted: u64,
    /// Samples for a pane already built or skipped.
    pub dropped: u64,
    /// Panes built and stored.
    pub panes_built: u64,
    /// Panes whose build failed; they are skipped, not retried.
    pub panes_failed: u64,
}

pub struct Precompute {
    /// Built panes.
    pub store: SummaryStore<Batch>,
    /// What happened to incoming samples.
    pub stats: PrecomputeStats,
    /* private: the DAG, one compiled build per definition, open panes */
}
impl Precompute {
    /// Compiles one build per definition of `sds`, a description of `dag`.
    pub fn new(dag: PhysicalASAPDAG, sds: &Sds, allowed_lateness_ms: i64) -> Result<Self, PrecomputeError>;
    /// Newest sample time minus the allowed lateness.
    pub fn watermark_ms(&self) -> Option<i64>;
    /// Buffers `samples`, then builds every pane the watermark has passed.
    pub fn ingest(&mut self, samples: &[RawSample]);
}
/// End of the pane of width `pane_ms` containing `timestamp_ms`; panes are `(end - pane_ms, end]`.
pub fn pane_end(timestamp_ms: i64, pane_ms: i64) -> i64;

// data_plane/src/engine.rs
/// The HTTP side of the engine thread.
pub struct EngineHandle { /* private */ }
impl EngineHandle {
    /// Starts the engine for the serialized plan; fails if it cannot be executed.
    pub fn spawn(plan_json: String, allowed_lateness_ms: i64) -> anyhow::Result<Self>;
    /// Queues samples for precompute, or `Busy` if the engine is behind.
    pub fn try_ingest(&self, samples: Vec<RawSample>) -> Result<(), Busy>;
    /// Precompute counters once every command queued before has run.
    pub async fn stats(&self) -> PrecomputeStats;
}

DataPlane::new(plan_json, prometheus_url, allowed_lateness_ms) now starts the engine.

Before this PR

Remote Write samples were decoded and dropped.

After this PR

Example 4b without raw data, one series sampled every 15 s from 1 to 75 min: the pane ending at 10 min is skipped as possibly partial, and panes ending at 20..70 min are built.

Usage

// The plan and SDS of #819's usage (Example 4b, no raw data).
// Build panes with zero allowed lateness.
let mut precompute = Precompute::new(dag.clone(), &sds, 0)?;
// One series, a sample every 15 s from 1 min to 75 min, values cycling 0..240.
let labels = Arc::new(BTreeMap::from([
    ("__name__".to_string(), "latency_ms".to_string()),
    ("service".to_string(), "a".to_string()),
]));
let samples: Vec<RawSample> = (60_000..4_500_000).step_by(15_000)
    .map(|t| RawSample { labels: labels.clone(), timestamp_ms: t, value: ((t / 15_000) % 240) as f64 })
    .collect();
precompute.ingest(&samples);
// PrecomputeStats { accepted: 259, dropped: 37, panes_built: 6, panes_failed: 0 }, 6 panes stored

Verification

  • data_plane/tests/precompute.rs (5 tests): the watermark and skipped first pane, ignored metrics and dropped late samples, empty panes, label matching, left-open boundaries.
  • cargo test -p control_plane -p data_plane, fmt, clippy on this PR's tree.
  • Usage output above is from running it on the feat(query): answer planned instant queries by running the plan's roots #821 tree, whose precompute API is this PR's.

An engine thread owns the installed plan (Planner IR holds Rcs) and receives
Remote Write samples from the HTTP server. For each stored summary
definition it buffers its Scan metric's samples by tumbling pane
(end - width, end], like PromQL ranges. When the watermark (newest sample
minus --allowed-lateness-ms) passes a pane's end, it runs the plan's Scan
over them (applying label matchers) and the definition's build from its
TimeRange input, then stores the output. Panes without samples are stored
empty; the first pane seen is skipped because it may be partial; samples for
built panes are dropped and counted. A full engine queue answers Remote
Write with 503 so senders retry.
@zzylol
zzylol force-pushed the feat/precompute-engine branch from eb8a314 to f9c3343 Compare October 18, 2026 15:40
@zzylol zzylol added area: data plane Runtime ingestion, storage, and query-serving data paths area: summary execution Warm-summary binding, readout, composition, and result decoding enhancement New feature or request feature: PromQL PromQL compatibility and execution semantics labels Oct 18, 2026
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

area: data plane Runtime ingestion, storage, and query-serving data paths area: summary execution Warm-summary binding, readout, composition, and result decoding enhancement New feature or request feature: PromQL PromQL compatibility and execution semantics

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant