diff --git a/Cargo.lock b/Cargo.lock index 4f977a85..441160ea 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -389,6 +389,7 @@ dependencies = [ name = "asap-gorilla" version = "0.1.0" dependencies = [ + "crc32fast", "serde", "serde_json", "thiserror 1.0.69", diff --git a/asap-query-engine/Cargo.toml b/asap-query-engine/Cargo.toml index 08d53de7..672988df 100644 --- a/asap-query-engine/Cargo.toml +++ b/asap-query-engine/Cargo.toml @@ -95,6 +95,14 @@ crc32fast = "1.4" # rustls backend already pulled in by `reqwest`. `lru` powers the # Phase 3 `ChunkCache` keyed on chunk object key. asap-gorilla = { path = "../../ASAPCollector/asap-gorilla" } +# mvp/v5 NOTE FOR REVIEWERS: this PR depends on the matching +# `mvp/v5-postings-compactor` PR in ASAPCollector — specifically the +# `asap_gorilla::Postings` type and the `IndexEntry::byte_offset / +# byte_length / object_key` fields. CI machines that build from a +# `/home/.../ASAPCollector` checkout pinned to `origin/main` will +# fail compilation with `no Postings in the root` until the +# collector PR lands. Local dev: `git fetch && git checkout +# mvp/v5-postings-compactor` in the sibling ASAPCollector repo. s3 = { version = "0.37", package = "rust-s3", default-features = false, features = ["tokio-rustls-tls"] } lru = "0.12" diff --git a/asap-query-engine/src/drivers/query/fallback/cold_store/gorilla_s3.rs b/asap-query-engine/src/drivers/query/fallback/cold_store/gorilla_s3.rs index 02368662..cae16163 100644 --- a/asap-query-engine/src/drivers/query/fallback/cold_store/gorilla_s3.rs +++ b/asap-query-engine/src/drivers/query/fallback/cold_store/gorilla_s3.rs @@ -50,9 +50,9 @@ use thiserror::Error; use tokio::sync::Mutex; use tracing::debug; -use asap_gorilla::{GorillaDecoder, IndexFile}; +use asap_gorilla::{GorillaDecoder, IndexFile, Postings}; -use super::{ChunkRef, ColdStore, ColdStoreError, RawSample}; +use super::{ChunkRef, ColdStore, ColdStoreError, PostingsHits, RawSample}; // ───────────────────────────────────────────────────────────────────── // Public config @@ -310,6 +310,15 @@ pub use rust_s3_backend::S3ObjectStore; /// chunk skip the Gorilla decode pass entirely. type ChunkCache = Mutex>>>; +/// **mvp/v5**: LRU cache for parsed postings sidecars. Keyed by +/// the postings-v1.json S3 key (one per `(metric, hour)`). 256 +/// entries by default → ≈ 256 MiB at 1 MiB per postings file. +type PostingsCache = Mutex>>; + +/// **mvp/v5**: LRU cache for parsed `index.json` files (per +/// `(metric, hour)`). Same capacity tier as the postings cache. +type IndexCache = Mutex>>; + /// `ColdStore` adapter that reads `GORILLA1`-format chunks out of /// an S3-compatible bucket. See module docs for layout + S3 client /// notes. @@ -317,6 +326,16 @@ pub struct GorillaS3ColdStore { object_store: Arc, config: GorillaS3Config, cache: ChunkCache, + /// **mvp/v5**: postings sidecar cache. + postings_cache: PostingsCache, + /// **mvp/v5**: index.json cache. Reserved for the upcoming + /// `Range:`-based partial-read path that fetches chunk bytes + /// out of compactor-merged objects — the cache is wired now to + /// match the backend's hot-path layout but the caller doesn't + /// yet route partial reads through it. The current `read_chunk` + /// already uses an LRU on samples, which is the dominant cost. + #[allow(dead_code)] + index_cache: IndexCache, } impl GorillaS3ColdStore { @@ -326,18 +345,32 @@ impl GorillaS3ColdStore { pub fn new(object_store: Arc, config: GorillaS3Config) -> Self { let cap = NonZeroUsize::new(config.cache_capacity.max(1)) .unwrap_or(NonZeroUsize::new(1).unwrap()); + // mvp/v5: postings + index caches scale with the chunk + // cache (one entry per hour-bucket, mirrors typical query + // cardinality). + let pc_cap = NonZeroUsize::new(cap.get().max(64)) + .unwrap_or(NonZeroUsize::new(64).unwrap()); Self { object_store, config, cache: Mutex::new(LruCache::new(cap)), + postings_cache: Mutex::new(LruCache::new(pc_cap)), + index_cache: Mutex::new(LruCache::new(pc_cap)), } } /// Build from a [`GorillaS3Config`] using the default /// `rust-s3`-backed [`ObjectStore`]. + /// + /// **mvp/v5**: the underlying `S3ObjectStore` is wrapped in an + /// [`super::S3CostTrackingObjectStore`] tied to the global + /// counter set, so the HTTP server's `/internal/s3_cost.csv` + /// + `/metrics` endpoints report measured PUT/GET/etc counts. pub fn with_default_backend(config: GorillaS3Config) -> Result { - let backend = Arc::new(S3ObjectStore::new(&config)?); - Ok(Self::new(backend, config)) + let backend: Arc = Arc::new(S3ObjectStore::new(&config)?); + let counters = super::s3_cost_tracker::global_s3_cost_counters(); + let tracked = super::S3CostTrackingObjectStore::new(backend, counters); + Ok(Self::new(Arc::new(tracked), config)) } /// Borrow the active config — useful for diagnostics. @@ -348,6 +381,22 @@ impl GorillaS3ColdStore { /// Render the configured `prefix_template` for one /// `(metric, hour)` bucket and append `index.json`. fn index_key(&self, metric: &str, ts_ms: i64) -> String { + let mut key = self.bucket_prefix(metric, ts_ms); + key.push_str("index.json"); + key + } + + /// **mvp/v5**: derive the postings-v1.json key for the same + /// `(metric, hour)` bucket as [`Self::index_key`]. + fn postings_key(&self, metric: &str, ts_ms: i64) -> String { + let mut key = self.bucket_prefix(metric, ts_ms); + key.push_str("postings-v1.json"); + key + } + + /// Shared prefix-rendering helper used by [`Self::index_key`] / + /// [`Self::postings_key`]. Always ends with `/`. + fn bucket_prefix(&self, metric: &str, ts_ms: i64) -> String { let dt: DateTime = DateTime::::from_timestamp_millis(ts_ms) .unwrap_or_else(|| DateTime::::from_timestamp(0, 0).unwrap()); let prefix = self @@ -363,7 +412,6 @@ impl GorillaS3ColdStore { if !key.ends_with('/') { key.push('/'); } - key.push_str("index.json"); key } @@ -486,6 +534,102 @@ impl ColdStore for GorillaS3ColdStore { } Ok(samples) } + + /// **mvp/v5**: postings-aware chunk pruning. + /// + /// Walks the per-hour buckets covering `[start_ms, end_ms)`, + /// fetches each `postings-v1.json` (LRU-cached), and intersects + /// the per-matcher series-id lists across every bucket. + /// Missing-postings buckets are noted (caller-visible quirk). + /// + /// Empty `matchers` ⇒ returns the union of all postings' + /// series_ids in range — this is the "no predicate" + /// short-circuit and the engine usually skips calling us in + /// that case. + async fn list_postings_for( + &self, + metric: &str, + start_ms: i64, + end_ms: i64, + matchers: &[(String, String)], + ) -> Result { + let buckets = Self::hour_starts(start_ms, end_ms); + let mut hits = PostingsHits { + series_ids: Vec::new(), + buckets_in_range: buckets.len(), + buckets_with_postings: 0, + }; + // Per-bucket: load postings, intersect across matchers, + // union into the running result. Cross-bucket join is a + // UNION (a series might exist in one hour but not the + // next); intra-bucket intersection across matchers is an + // AND. + let mut union_set: std::collections::BTreeSet = + std::collections::BTreeSet::new(); + for hour_ms in buckets { + let key = self.postings_key(metric, hour_ms); + // LRU short-circuit. + let postings = { + let mut guard = self.postings_cache.lock().await; + guard.get(&key).cloned() + }; + let postings = match postings { + Some(p) => Some(p), + None => match self.object_store.get_object(&key).await { + Ok(bytes) => match Postings::read(bytes.as_slice()) { + Ok(p) => { + let arc = Arc::new(p); + let mut guard = self.postings_cache.lock().await; + guard.put(key.clone(), arc.clone()); + Some(arc) + } + Err(e) => { + // Treat a corrupt postings file as + // "missing" — the engine then falls + // through to the scan-all path with + // the postings_missing quirk. + debug!(key = %key, error = %e, "gorilla-s3: postings parse failed; treating as missing"); + None + } + }, + Err(e) if self.object_store.object_missing(&e) => { + debug!(key = %key, "gorilla-s3: postings missing for hour bucket"); + None + } + Err(e) => return Err(e), + }, + }; + let Some(postings) = postings else { continue }; + hits.buckets_with_postings += 1; + + // Intersect across matchers within this bucket. + let bucket_set: std::collections::BTreeSet = if matchers.is_empty() { + // Union of every series_id across every label. + let mut set = std::collections::BTreeSet::new(); + for by_value in postings.by_label.values() { + for ids in by_value.values() { + set.extend(ids.iter().copied()); + } + } + set + } else { + let first = + postings.lookup(&matchers[0].0, &matchers[0].1); + let mut acc: std::collections::BTreeSet = + first.iter().copied().collect(); + for (label_name, label_value) in &matchers[1..] { + let next = postings.lookup(label_name, label_value); + let next_set: std::collections::BTreeSet = + next.iter().copied().collect(); + acc = acc.intersection(&next_set).copied().collect(); + } + acc + }; + union_set.extend(bucket_set); + } + hits.series_ids = union_set.into_iter().collect(); + Ok(hits) + } } // ───────────────────────────────────────────────────────────────────── @@ -666,6 +810,9 @@ mod tests { sample_count: 10, label_hash: 0xAAAA, size_bytes: 100, + object_key: None, + byte_offset: None, + byte_length: None, }, IndexEntry { key: key_b.clone(), @@ -673,6 +820,9 @@ mod tests { sample_count: 11, label_hash: 0xBBBB, size_bytes: 110, + object_key: None, + byte_offset: None, + byte_length: None, }, IndexEntry { key: key_c.clone(), @@ -680,6 +830,9 @@ mod tests { sample_count: 12, label_hash: 0xCCCC, size_bytes: 120, + object_key: None, + byte_offset: None, + byte_length: None, }, ]; @@ -731,6 +884,9 @@ mod tests { sample_count: 3, label_hash: 0x1234, size_bytes: block.len() as u32, + object_key: None, + byte_offset: None, + byte_length: None, }]), ) .await; @@ -772,6 +928,9 @@ mod tests { sample_count: 2, label_hash: 0, size_bytes: block.len() as u32, + object_key: None, + byte_offset: None, + byte_length: None, }]), ) .await; @@ -818,6 +977,9 @@ mod tests { sample_count: 2, label_hash: i as u64, size_bytes: block.len() as u32, + object_key: None, + byte_offset: None, + byte_length: None, }); chunk_refs.push(ChunkRef { key, @@ -923,6 +1085,9 @@ mod tests { sample_count: 2, label_hash: 0, size_bytes: block.len() as u32, + object_key: None, + byte_offset: None, + byte_length: None, }]), ) .await; @@ -954,6 +1119,9 @@ mod tests { sample_count: 1, label_hash: 0, size_bytes: 50, + object_key: None, + byte_offset: None, + byte_length: None, }]), ) .await; @@ -969,6 +1137,9 @@ mod tests { sample_count: 1, label_hash: 0, size_bytes: 50, + object_key: None, + byte_offset: None, + byte_length: None, }]), ) .await; diff --git a/asap-query-engine/src/drivers/query/fallback/cold_store/mod.rs b/asap-query-engine/src/drivers/query/fallback/cold_store/mod.rs index 77bd8e6c..6e5cd89a 100644 --- a/asap-query-engine/src/drivers/query/fallback/cold_store/mod.rs +++ b/asap-query-engine/src/drivers/query/fallback/cold_store/mod.rs @@ -36,10 +36,14 @@ use thiserror::Error; pub mod format; pub mod gorilla_s3; pub mod local_fs; +pub mod s3_cost_tracker; pub use format::{part_path_prefix, RawSample}; pub use gorilla_s3::{GorillaS3ColdStore, GorillaS3Config, GorillaS3ConfigError}; pub use local_fs::LocalFsColdStore; +pub use s3_cost_tracker::{ + global_s3_cost_counters, S3CostCounters, S3CostSnapshot, S3CostTrackingObjectStore, +}; /// Error surface for cold-store scans. #[derive(Debug, Error)] @@ -95,6 +99,39 @@ pub struct ChunkRef { pub size_bytes: u32, } +/// **mvp/v5**: postings result returned by [`ColdStore::list_postings_for`]. +/// +/// `series_ids` is the union of `series_id` lists across all hour +/// buckets in the requested time range, deduped and sorted ascending. +/// `postings_present_buckets` counts how many hour buckets actually +/// had a postings sidecar — used by the engine to decide whether to +/// emit a `data_source_quirk: postings_missing` annotation. +#[derive(Debug, Clone, PartialEq, Eq, Default)] +pub struct PostingsHits { + /// Series ids matching all label predicates, sorted ascending, + /// deduplicated. + pub series_ids: Vec, + /// Hour buckets in the request window. + pub buckets_in_range: usize, + /// Buckets that actually had a `postings-v1.json` sidecar. + pub buckets_with_postings: usize, +} + +impl PostingsHits { + /// `true` iff at least one hour bucket carried postings — + /// indicates the postings-aware filter ran on real data and the + /// caller should trust [`Self::series_ids`] as a complete answer. + pub fn fully_covered(&self) -> bool { + self.buckets_in_range > 0 && self.buckets_with_postings == self.buckets_in_range + } + + /// `true` iff postings were present for every bucket AND at + /// least one matched series. + pub fn nonempty_and_complete(&self) -> bool { + self.fully_covered() && !self.series_ids.is_empty() + } +} + /// Read-only view over a cold raw-sample store. /// /// Scans are `(metric, [start_ms, end_ms))` — inclusive start, @@ -157,6 +194,27 @@ pub trait ColdStore: Send + Sync { ) -> Result, ColdStoreError> { Err(ColdStoreError::Unsupported("read_chunk")) } + + /// **mvp/v5**: load + intersect per-bucket postings under + /// `(metric, time_range)` for the supplied `(label_name, + /// label_value)` matchers. The result's `series_ids` is the + /// intersection across all matchers — i.e. only series_ids + /// that match every predicate. With zero matchers this returns + /// the union of every series_id in range (rare; the engine + /// short-circuits the postings-aware path before calling). + /// + /// Default impl returns [`ColdStoreError::Unsupported`] so + /// JSONL-only stores keep compiling. The Gorilla-S3 cold + /// store overrides. + async fn list_postings_for( + &self, + _metric: &str, + _start_ms: i64, + _end_ms: i64, + _matchers: &[(String, String)], + ) -> Result { + Err(ColdStoreError::Unsupported("list_postings_for")) + } } /// Convenience alias: a label set as stored in a [`RawSample`]. diff --git a/asap-query-engine/src/drivers/query/fallback/cold_store/s3_cost_tracker.rs b/asap-query-engine/src/drivers/query/fallback/cold_store/s3_cost_tracker.rs new file mode 100644 index 00000000..a40e2d56 --- /dev/null +++ b/asap-query-engine/src/drivers/query/fallback/cold_store/s3_cost_tracker.rs @@ -0,0 +1,250 @@ +//! mvp/v5 — instrumented S3 client wrapper. +//! +//! The compaction story for the MVP demo wants a measured (not +//! fabricated) S3 cost picture: per-baseline counts of PUT / GET / +//! HEAD / LIST / DELETE plus bytes-out per request, dumped to CSV +//! at end-of-run. +//! +//! This module provides a thin wrapper that delegates to the +//! existing `rust-s3` [`s3::Bucket`] but ticks a small counter set +//! before / after every operation. Live values are exposed via a +//! Prometheus gauge (`asap_backend_s3__total`) so dashboards +//! see them in real time, AND a CSV / JSON dump on demand. +//! +//! ## Boundary +//! +//! The wrapper sits at the lowest level — between the +//! `GorillaS3ColdStore::ObjectStore` impl and the actual `Bucket`. +//! Tests that don't need S3 (the in-memory mock path) never touch +//! it; production deployments wire `S3CostTrackingObjectStore` +//! around `S3ObjectStore`. + +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, OnceLock}; + +use async_trait::async_trait; + +/// Process-wide S3 cost counters. The HTTP server's +/// `/internal/s3_cost.csv` endpoint reads this; the +/// `GorillaS3ColdStore` constructor opts in via +/// [`S3CostTrackingObjectStore`]. Lazy-initialised on first access. +static GLOBAL_S3_COST: OnceLock> = OnceLock::new(); + +/// Access (and lazily create) the process-wide S3 cost counters. +pub fn global_s3_cost_counters() -> Arc { + GLOBAL_S3_COST + .get_or_init(|| Arc::new(S3CostCounters::new())) + .clone() +} + +use super::gorilla_s3::ObjectStore; +use super::ColdStoreError; + +/// Per-operation counter set + cumulative bytes. +#[derive(Debug, Default)] +pub struct S3CostCounters { + /// PUT operations issued. + pub put_count: AtomicU64, + /// GET operations issued (full + range). + pub get_count: AtomicU64, + /// HEAD operations issued. + pub head_count: AtomicU64, + /// LIST operations issued. + pub list_count: AtomicU64, + /// DELETE operations issued. + pub delete_count: AtomicU64, + /// Bytes uploaded (PUT request bodies). + pub bytes_put: AtomicU64, + /// Bytes downloaded (GET response bodies). + pub bytes_got: AtomicU64, +} + +impl S3CostCounters { + /// Build a fresh zeroed counter set. + pub fn new() -> Self { + Self::default() + } + + /// Plain-old-data snapshot. + pub fn snapshot(&self) -> S3CostSnapshot { + S3CostSnapshot { + put_count: self.put_count.load(Ordering::Relaxed), + get_count: self.get_count.load(Ordering::Relaxed), + head_count: self.head_count.load(Ordering::Relaxed), + list_count: self.list_count.load(Ordering::Relaxed), + delete_count: self.delete_count.load(Ordering::Relaxed), + bytes_put: self.bytes_put.load(Ordering::Relaxed), + bytes_got: self.bytes_got.load(Ordering::Relaxed), + } + } + + /// Render Prometheus text-exposition lines for `/metrics`. + pub fn render_prometheus(&self) -> String { + let s = self.snapshot(); + format!( + concat!( + "# HELP asap_backend_s3_put_total S3 PUT count.\n", + "# TYPE asap_backend_s3_put_total counter\n", + "asap_backend_s3_put_total {}\n", + "asap_backend_s3_get_total {}\n", + "asap_backend_s3_head_total {}\n", + "asap_backend_s3_list_total {}\n", + "asap_backend_s3_delete_total {}\n", + "asap_backend_s3_bytes_put {}\n", + "asap_backend_s3_bytes_got {}\n", + ), + s.put_count, s.get_count, s.head_count, s.list_count, s.delete_count, + s.bytes_put, s.bytes_got, + ) + } + + /// Render a CSV summary suitable for the demo's `s3_cost.csv`. + /// Single header + single data row. + pub fn render_csv(&self) -> String { + let s = self.snapshot(); + format!( + "put_count,get_count,head_count,list_count,delete_count,bytes_put,bytes_got\n\ + {},{},{},{},{},{},{}\n", + s.put_count, s.get_count, s.head_count, s.list_count, s.delete_count, + s.bytes_put, s.bytes_got, + ) + } +} + +/// Plain-old-data snapshot returned by [`S3CostCounters::snapshot`]. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct S3CostSnapshot { + /// PUT count. + pub put_count: u64, + /// GET count. + pub get_count: u64, + /// HEAD count. + pub head_count: u64, + /// LIST count. + pub list_count: u64, + /// DELETE count. + pub delete_count: u64, + /// Bytes uploaded. + pub bytes_put: u64, + /// Bytes downloaded. + pub bytes_got: u64, +} + +/// `ObjectStore` wrapper that ticks the supplied counters. +/// +/// Note: only `get_object` is in the cold-store hot path today +/// (Phase 3 + Phase 4). PUT / LIST / DELETE / HEAD are recorded +/// even though current callers never go through them — having the +/// counter live makes follow-up MVP cost work additive. +pub struct S3CostTrackingObjectStore { + inner: Arc, + counters: Arc, +} + +impl S3CostTrackingObjectStore { + /// Wrap `inner` and a counter-set for the wrapper to update. + pub fn new( + inner: Arc, + counters: Arc, + ) -> Self { + Self { inner, counters } + } + + /// Borrow the live counters — handy for HTTP exposition. + pub fn counters(&self) -> &Arc { + &self.counters + } +} + +#[async_trait] +impl ObjectStore for S3CostTrackingObjectStore { + async fn get_object(&self, key: &str) -> Result, ColdStoreError> { + self.counters.get_count.fetch_add(1, Ordering::Relaxed); + let body = self.inner.get_object(key).await?; + self.counters + .bytes_got + .fetch_add(body.len() as u64, Ordering::Relaxed); + Ok(body) + } + + fn object_missing(&self, err: &ColdStoreError) -> bool { + self.inner.object_missing(err) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::drivers::query::fallback::cold_store::gorilla_s3::ObjectStore as _; + use std::collections::HashMap; + use tokio::sync::Mutex; + + /// Minimal in-memory ObjectStore stand-in for the wrapper test — + /// we don't pull in `InMemoryObjectStore` because it's a + /// `#[cfg(test)]` type local to its own module. + #[derive(Default)] + struct StubStore { + inner: Mutex>>, + } + + #[async_trait] + impl ObjectStore for StubStore { + async fn get_object(&self, key: &str) -> Result, ColdStoreError> { + let g = self.inner.lock().await; + match g.get(key) { + Some(b) => Ok(b.clone()), + None => Err(ColdStoreError::Backend(format!("get {key}: not found"))), + } + } + } + + #[tokio::test] + async fn counters_increment_on_get() { + let inner = Arc::new(StubStore::default()); + inner + .inner + .lock() + .await + .insert("k1".to_string(), vec![0u8; 100]); + let counters = Arc::new(S3CostCounters::new()); + let wrapped = S3CostTrackingObjectStore::new(inner, counters.clone()); + + let _ = wrapped.get_object("k1").await.unwrap(); + let _ = wrapped.get_object("k1").await.unwrap(); + let snap = counters.snapshot(); + assert_eq!(snap.get_count, 2); + assert_eq!(snap.bytes_got, 200); + assert_eq!(snap.put_count, 0); + assert_eq!(snap.head_count, 0); + } + + #[tokio::test] + async fn missing_object_does_not_count_bytes() { + let inner = Arc::new(StubStore::default()); + let counters = Arc::new(S3CostCounters::new()); + let wrapped = S3CostTrackingObjectStore::new(inner, counters.clone()); + let _ = wrapped.get_object("missing").await; + let snap = counters.snapshot(); + assert_eq!(snap.get_count, 1); + assert_eq!(snap.bytes_got, 0); + } + + #[test] + fn render_csv_has_expected_columns() { + let c = S3CostCounters::new(); + c.put_count.store(3, Ordering::Relaxed); + c.get_count.store(7, Ordering::Relaxed); + c.bytes_got.store(1024, Ordering::Relaxed); + let csv = c.render_csv(); + assert!(csv.starts_with("put_count,get_count,head_count,list_count,delete_count,bytes_put,bytes_got\n")); + assert!(csv.contains("3,7,0,0,0,0,1024")); + } + + #[test] + fn render_prometheus_has_help_line() { + let c = S3CostCounters::new(); + let p = c.render_prometheus(); + assert!(p.contains("asap_backend_s3_put_total")); + assert!(p.contains("# TYPE asap_backend_s3_put_total counter")); + } +} diff --git a/asap-query-engine/src/drivers/query/servers/http.rs b/asap-query-engine/src/drivers/query/servers/http.rs index bc0f9c51..76351a57 100644 --- a/asap-query-engine/src/drivers/query/servers/http.rs +++ b/asap-query-engine/src/drivers/query/servers/http.rs @@ -260,6 +260,10 @@ impl HttpServer { .route(runtime_info_path, get(handle_runtime_info)) .route(runtime_info_path, post(handle_runtime_info)) .route("/metrics", get(handle_metrics)) + // mvp/v5: dump the S3 cost-tracking counters as CSV. + // The demo's `run_mvp_demo.sh` curls this for each + // baseline; missing counters render as zeros. + .route("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/internal/s3_cost.csv", get(handle_s3_cost_csv)) // Controller integration endpoints .route("/api/v1/precompute", post(handle_precompute_job)) .route("/api/v1/health", get(handle_health)) @@ -999,6 +1003,12 @@ async fn handle_metrics() -> impl IntoResponse { let mut buffer = Vec::new(); prometheus::Encoder::encode(&encoder, &metric_families, &mut buffer) .unwrap_or_else(|e| tracing::error!("Failed to encode metrics: {}", e)); + // mvp/v5: append the S3 cost counters in Prometheus text + // exposition. Mirrors `/internal/s3_cost.csv` — the CSV is for + // the demo, this is for live dashboards. + let counters = + crate::drivers::query::fallback::cold_store::global_s3_cost_counters(); + buffer.extend_from_slice(counters.render_prometheus().as_bytes()); ( [( axum::http::header::CONTENT_TYPE, @@ -1008,6 +1018,23 @@ async fn handle_metrics() -> impl IntoResponse { ) } +/// mvp/v5: CSV dump of the S3 cost counters. +/// +/// Renders ONE header row + ONE data row. Empty when no S3 +/// operations have been issued (the counters default to zero, so +/// the CSV is still well-formed). +async fn handle_s3_cost_csv() -> impl IntoResponse { + let counters = + crate::drivers::query::fallback::cold_store::global_s3_cost_counters(); + ( + [( + axum::http::header::CONTENT_TYPE, + "text/csv; charset=utf-8", + )], + counters.render_csv(), + ) +} + // ============================================================ // Range Query Handlers // ============================================================ diff --git a/asap-query-engine/src/engines/gorilla_engine/exact_executor.rs b/asap-query-engine/src/engines/gorilla_engine/exact_executor.rs index 6398b0dd..60ef04b6 100644 --- a/asap-query-engine/src/engines/gorilla_engine/exact_executor.rs +++ b/asap-query-engine/src/engines/gorilla_engine/exact_executor.rs @@ -16,9 +16,11 @@ use std::sync::Arc; use tracing::debug; -use crate::drivers::query::fallback::cold_store::{ColdStore, RawSample}; +use crate::drivers::query::fallback::cold_store::{ + ChunkRef, ColdStore, ColdStoreError, RawSample, +}; -use super::query_planner::{QueryPlan, QueryStatistic}; +use super::query_planner::{LabelMatcher, QueryPlan, QueryStatistic}; use super::{EngineError, ExecutionOutcome, GorillaEngineConfig}; /// Streaming-additive operation tag — what the per-sample fold @@ -85,6 +87,12 @@ impl ExactExecutor { /// the per-sample fold, drops the decoded chunk before fetching /// the next one. Memory is O(1) per query, regardless of how /// many samples the time range covers. + /// + /// **mvp/v5**: when the plan carries label matchers, the executor + /// first reads the postings sidecar to compute the matching + /// `series_ids`, then prunes the chunk list down to chunks + /// whose `label_hash` appears in that set. Falls back to the + /// scan-all path when postings are missing. pub async fn execute_streaming_additive( &self, plan: &QueryPlan, @@ -95,20 +103,28 @@ impl ExactExecutor { .cold_store .list_chunks(&plan.metric, start_ms, end_ms) .await?; - let chunks_fetched = chunks.len(); + let total_chunks = chunks.len(); debug!( metric = plan.metric.as_str(), - chunks = chunks_fetched, + chunks = total_chunks, op = ?op, + label_matchers = plan.label_matchers.len(), "gorilla-engine: streaming additive over chunks" ); + let (filtered_chunks, postings_outcome) = + self.apply_postings_filter(plan, &chunks).await; + let mut acc = AdditiveAccumulator::new(op); let mut samples_scanned: usize = 0; - for chunk in chunks { + let chunks_fetched = filtered_chunks.len(); + for chunk in filtered_chunks { let samples = self.cold_store.read_chunk(&chunk).await?; for s in samples { if s.ts_ms >= start_ms && s.ts_ms < end_ms { + if !self.sample_matches(plan, &s) { + continue; + } acc.observe(&s); samples_scanned += 1; } @@ -120,9 +136,123 @@ impl ExactExecutor { value, samples_scanned, chunks_fetched, + chunks_skipped_via_postings: total_chunks - chunks_fetched, + postings_filtered_series_count: postings_outcome.matched_series, + postings_missing: postings_outcome.postings_missing, }) } + /// **mvp/v5**: apply the postings-aware filter to a chunk list. + /// Returns `(filtered_chunks, postings_outcome)` where the + /// outcome captures `(matched_series, postings_missing)` so the + /// caller can populate [`super::ExecutionOutcome`] without + /// re-querying. + async fn apply_postings_filter( + &self, + plan: &QueryPlan, + chunks: &[ChunkRef], + ) -> (Vec, PostingsOutcome) { + if plan.label_matchers.is_empty() { + // No predicate — the postings filter is a no-op. The + // postings file isn't consulted at all in this path. + return ( + chunks.to_vec(), + PostingsOutcome { + matched_series: 0, + postings_missing: false, + }, + ); + } + let matchers: Vec<(String, String)> = plan + .label_matchers + .iter() + .map(|LabelMatcher { name, value }| (name.clone(), value.clone())) + .collect(); + let (start_ms, end_ms) = plan.time_range_ms; + let hits = match self + .cold_store + .list_postings_for(&plan.metric, start_ms, end_ms, &matchers) + .await + { + Ok(h) => h, + Err(ColdStoreError::Unsupported(_)) => { + // Backend doesn't support postings at all (legacy + // cold store). Surface as missing and fall through. + debug!("gorilla-engine: cold store does not support postings; falling back to scan-all"); + return ( + chunks.to_vec(), + PostingsOutcome { + matched_series: 0, + postings_missing: true, + }, + ); + } + Err(e) => { + // Transport/parse failure — log + fall through. We + // don't propagate the error because the scan-all + // path is still correct, just slower. + debug!(error = %e, "gorilla-engine: postings fetch failed; falling back to scan-all"); + return ( + chunks.to_vec(), + PostingsOutcome { + matched_series: 0, + postings_missing: true, + }, + ); + } + }; + + // If any bucket in range was missing postings we can't trust + // the filter to be complete; scan everything (correctness + // first, postings are a perf optimization). + if !hits.fully_covered() { + return ( + chunks.to_vec(), + PostingsOutcome { + matched_series: hits.series_ids.len(), + postings_missing: true, + }, + ); + } + + // Postings → series_ids → keep only chunks whose + // `label_hash` is in the set. Chunks with `label_hash == 0` + // are pre-mvp/v5 multi-series chunks that don't pin a + // single series — keep them (they may carry matching + // series; correctness > pruning). + let series_set: std::collections::BTreeSet = + hits.series_ids.iter().copied().collect(); + let filtered: Vec = chunks + .iter() + .filter(|c| c.label_hash == 0 || series_set.contains(&c.label_hash)) + .cloned() + .collect(); + ( + filtered, + PostingsOutcome { + matched_series: hits.series_ids.len(), + postings_missing: false, + }, + ) + } + + /// Post-decode label-equality filter. Always-true when no + /// matchers are present (most common). Used as a safety net so + /// chunks with `label_hash = 0` (multi-series, can't be pruned + /// at the postings level) still respect the predicate. + fn sample_matches(&self, plan: &QueryPlan, s: &RawSample) -> bool { + if plan.label_matchers.is_empty() { + return true; + } + for LabelMatcher { name, value } in &plan.label_matchers { + match s.labels.get(name) { + Some(v) if v == value => {} + _ => return false, + } + } + true + } + /// Buffered quantile path. Materialises every in-range sample /// up to [`GorillaEngineConfig::max_buffered_samples`], sorts /// the value column, and picks the φ-rank using a @@ -134,16 +264,19 @@ impl ExactExecutor { plan: &QueryPlan, phi: f64, ) -> Result { - let (samples, chunks_fetched) = self.collect_buffered_samples(plan).await?; - if samples.is_empty() { + let buffered = self.collect_buffered_samples(plan).await?; + if buffered.samples.is_empty() { return Ok(ExecutionOutcome { value: f64::NAN, samples_scanned: 0, - chunks_fetched, + chunks_fetched: buffered.chunks_fetched, + chunks_skipped_via_postings: buffered.chunks_skipped_via_postings, + postings_filtered_series_count: buffered.postings_filtered_series_count, + postings_missing: buffered.postings_missing, }); } - let mut values: Vec = samples.iter().map(|s| s.value).collect(); + let mut values: Vec = buffered.samples.iter().map(|s| s.value).collect(); values.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal)); let n = values.len() as f64; @@ -153,7 +286,10 @@ impl ExactExecutor { Ok(ExecutionOutcome { value: values[idx], samples_scanned: values.len(), - chunks_fetched, + chunks_fetched: buffered.chunks_fetched, + chunks_skipped_via_postings: buffered.chunks_skipped_via_postings, + postings_filtered_series_count: buffered.postings_filtered_series_count, + postings_missing: buffered.postings_missing, }) } @@ -168,45 +304,60 @@ impl ExactExecutor { plan: &QueryPlan, k: usize, ) -> Result { - let (samples, chunks_fetched) = self.collect_buffered_samples(plan).await?; - if samples.is_empty() { + let buffered = self.collect_buffered_samples(plan).await?; + if buffered.samples.is_empty() { return Ok(ExecutionOutcome { value: f64::NAN, samples_scanned: 0, - chunks_fetched, + chunks_fetched: buffered.chunks_fetched, + chunks_skipped_via_postings: buffered.chunks_skipped_via_postings, + postings_filtered_series_count: buffered.postings_filtered_series_count, + postings_missing: buffered.postings_missing, }); } - let mut values: Vec = samples.iter().map(|s| s.value).collect(); + let mut values: Vec = buffered.samples.iter().map(|s| s.value).collect(); values.sort_by(|a, b| b.partial_cmp(a).unwrap_or(std::cmp::Ordering::Equal)); let take = k.min(values.len()); let topk_sum: f64 = values.iter().take(take).sum(); Ok(ExecutionOutcome { value: topk_sum, samples_scanned: values.len(), - chunks_fetched, + chunks_fetched: buffered.chunks_fetched, + chunks_skipped_via_postings: buffered.chunks_skipped_via_postings, + postings_filtered_series_count: buffered.postings_filtered_series_count, + postings_missing: buffered.postings_missing, }) } /// Shared helper for the buffered paths: walk every chunk, /// keep in-range samples, enforce the /// [`GorillaEngineConfig::max_buffered_samples`] ceiling. + /// **mvp/v5**: also applies the postings-aware filter so the + /// quantile / topk paths share the same pruning as + /// streaming-additive. async fn collect_buffered_samples( &self, plan: &QueryPlan, - ) -> Result<(Vec, usize), EngineError> { + ) -> Result { let (start_ms, end_ms) = plan.time_range_ms; let chunks = self .cold_store .list_chunks(&plan.metric, start_ms, end_ms) .await?; - let chunks_fetched = chunks.len(); + let total_chunks = chunks.len(); + let (filtered_chunks, postings_outcome) = + self.apply_postings_filter(plan, &chunks).await; + let chunks_fetched = filtered_chunks.len(); let mut buffer: Vec = Vec::new(); let limit = self.config.max_buffered_samples; - for chunk in chunks { + for chunk in filtered_chunks { let samples = self.cold_store.read_chunk(&chunk).await?; for s in samples { if s.ts_ms >= start_ms && s.ts_ms < end_ms { + if !self.sample_matches(plan, &s) { + continue; + } if buffer.len() >= limit { // Surface the over-budget sample count // (limit + 1) so callers can pin @@ -222,10 +373,32 @@ impl ExactExecutor { } } } - Ok((buffer, chunks_fetched)) + Ok(BufferedScan { + samples: buffer, + chunks_fetched, + chunks_skipped_via_postings: total_chunks - chunks_fetched, + postings_filtered_series_count: postings_outcome.matched_series, + postings_missing: postings_outcome.postings_missing, + }) } } +/// Output of [`ExactExecutor::collect_buffered_samples`]. +struct BufferedScan { + samples: Vec, + chunks_fetched: usize, + chunks_skipped_via_postings: usize, + postings_filtered_series_count: usize, + postings_missing: bool, +} + +/// Output of [`ExactExecutor::apply_postings_filter`]. +#[derive(Debug, Clone, Copy)] +struct PostingsOutcome { + matched_series: usize, + postings_missing: bool, +} + /// Per-sample fold for the streaming-additive path. Fields are /// kept in raw f64 (sum / min / max) + i64 (count) so the /// finaliser can pick the right arithmetic per op. diff --git a/asap-query-engine/src/engines/gorilla_engine/mod.rs b/asap-query-engine/src/engines/gorilla_engine/mod.rs index 8c5ee3b2..33eda5b2 100644 --- a/asap-query-engine/src/engines/gorilla_engine/mod.rs +++ b/asap-query-engine/src/engines/gorilla_engine/mod.rs @@ -156,6 +156,14 @@ impl GorillaQueryEngine { &self.config } + /// Test-only accessor for the underlying cold store. mvp/v5 + /// tests use this to construct a `ExactExecutor` that shares + /// the same mock without re-wrapping in a fresh `Arc`. + #[cfg(test)] + pub(super) fn cold_store_for_tests(&self) -> Arc { + self.cold_store.clone() + } + /// Execute a parsed PromQL query against the cold tier. /// /// The query string is parsed via [`query_planner::plan_query`], @@ -241,6 +249,22 @@ pub struct ExecutionOutcome { pub samples_scanned: usize, /// Number of chunks the executor fetched from the cold store. pub chunks_fetched: usize, + /// **mvp/v5**: number of chunks the postings filter pruned — + /// the executor was able to skip these without a chunk-body + /// fetch. `0` when the postings-aware path didn't run (no label + /// matchers / postings missing). + pub chunks_skipped_via_postings: usize, + /// **mvp/v5**: number of series the postings file said matched + /// the label predicates. The executor uses this to decide + /// whether a chunk's `label_hash` is interesting before paying + /// for the chunk body. Surfaces in `infos` as + /// `postings_filtered_series_count`. + pub postings_filtered_series_count: usize, + /// **mvp/v5**: `true` when the engine hit a missing postings + /// sidecar in the request window and fell back to the scan-all + /// path. Drives the `data_source_quirk: postings_missing` + /// `infos` annotation. + pub postings_missing: bool, } impl ExecutionOutcome { @@ -250,18 +274,34 @@ impl ExecutionOutcome { value: f64::NAN, samples_scanned: 0, chunks_fetched: 0, + chunks_skipped_via_postings: 0, + postings_filtered_series_count: 0, + postings_missing: false, } } /// Build the `infos` array surfaced on the wire response. /// Pulled out so tests can pin the exact strings. pub fn info_lines(&self) -> Vec { - vec![ + let mut out = vec![ AccuracyProfile::exact().summary(), DATA_SOURCE_GORILLA_ARCHIVE.to_string(), format!("samples_scanned: {}", self.samples_scanned), format!("chunks_fetched: {}", self.chunks_fetched), - ] + ]; + // mvp/v5: surface postings-aware execution counters. + out.push(format!( + "chunks_skipped_via_postings: {}", + self.chunks_skipped_via_postings + )); + out.push(format!( + "postings_filtered_series_count: {}", + self.postings_filtered_series_count + )); + if self.postings_missing { + out.push("data_source_quirk: postings_missing".to_string()); + } + out } } diff --git a/asap-query-engine/src/engines/gorilla_engine/query_planner.rs b/asap-query-engine/src/engines/gorilla_engine/query_planner.rs index 6334f6fd..80a9d980 100644 --- a/asap-query-engine/src/engines/gorilla_engine/query_planner.rs +++ b/asap-query-engine/src/engines/gorilla_engine/query_planner.rs @@ -64,6 +64,23 @@ impl QueryStatistic { } } +/// One label-equality matcher extracted from the PromQL AST. mvp/v5 +/// uses these to drive the postings-aware chunk-pruning path. +/// +/// The MVP only supports exact equality (`label = "value"`). Regex +/// (`=~`) and inequality (`!=`, `!~`) matchers fall through to a +/// post-decode filter — the postings file holds *exact* values per +/// label, not patterns. The fall-through is correct (just slower) +/// and is signalled to callers via +/// [`QueryPlan::has_unsupported_matchers`]. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct LabelMatcher { + /// Label name, e.g. `"service"`. + pub name: String, + /// Label value, e.g. `"api"`. + pub value: String, +} + /// Output of [`plan_query`]. #[derive(Debug, Clone, PartialEq)] pub struct QueryPlan { @@ -73,6 +90,19 @@ pub struct QueryPlan { /// `[range]` duration. pub time_range_ms: (i64, i64), pub statistic: QueryStatistic, + /// **mvp/v5**: exact-equality label matchers extracted from the + /// vector selector. Empty for `metric[range]` (no predicate). + /// Non-empty for `metric{label="value"}[range]`. Used by the + /// postings-aware chunk filter; `=~` / `!=` / `!~` matchers are + /// dropped from this list and signalled via + /// [`Self::has_unsupported_matchers`]. + pub label_matchers: Vec, + /// **mvp/v5**: `true` iff the original PromQL had at least one + /// matcher we couldn't translate into a postings lookup (regex, + /// inequality). The caller must still apply those matchers + /// post-decode; we surface the flag so `data_source_quirk` + /// annotations make it back to the client. + pub has_unsupported_matchers: bool, } /// Parse `query` and produce a [`QueryPlan`]. `now` defaults to @@ -128,10 +158,13 @@ fn plan_from_call(call: &Call, now_ms: i64) -> Result { "increase" => QueryStatistic::Increase, _ => unreachable!(), }; + let (label_matchers, has_unsupported) = extract_label_matchers(&ms.vs); Ok(QueryPlan { metric, time_range_ms: (now_ms - range_ms, now_ms), statistic: stat, + label_matchers, + has_unsupported_matchers: has_unsupported, }) } "quantile_over_time" => { @@ -145,10 +178,13 @@ fn plan_from_call(call: &Call, now_ms: i64) -> Result { let phi = expect_number(&call.args.args[0], "quantile_over_time φ")?; let ms = expect_matrix_selector(&call.args.args[1], "quantile_over_time")?; let (metric, range_ms) = matrix_metric_and_range_ms(ms); + let (label_matchers, has_unsupported) = extract_label_matchers(&ms.vs); Ok(QueryPlan { metric, time_range_ms: (now_ms - range_ms, now_ms), statistic: QueryStatistic::QuantileOverTime { phi }, + label_matchers, + has_unsupported_matchers: has_unsupported, }) } other => Err(format!( @@ -187,10 +223,13 @@ fn plan_from_aggregate(agg: &AggregateExpr, now_ms: i64) -> Result { let (metric, range_ms) = matrix_metric_and_range_ms(ms); + let (label_matchers, has_unsupported) = extract_label_matchers(&ms.vs); QueryPlan { metric, time_range_ms: (now_ms - range_ms, now_ms), statistic: QueryStatistic::SumOverTime, // overlay below + label_matchers, + has_unsupported_matchers: has_unsupported, } } _ => plan_from_ast(&agg.expr, now_ms)?, @@ -199,6 +238,8 @@ fn plan_from_aggregate(agg: &AggregateExpr, now_ms: i64) -> Result String { String::new() } +/// **mvp/v5**: extract exact-equality label matchers from a vector +/// selector for postings-aware chunk pruning. +/// +/// Returns `(supported_matchers, has_unsupported_matchers)`. Supported +/// matchers are the `label = "value"` tuples the postings file can +/// answer directly. Anything else (regex, inequality, the implicit +/// `__name__` matcher) is excluded from `supported_matchers` and +/// flips the second return value to `true` — the executor still +/// applies them post-decode for correctness. +pub(crate) fn extract_label_matchers(vs: &VectorSelector) -> (Vec, bool) { + use promql_parser::label::MatchOp; + + let mut supported = Vec::new(); + let mut has_unsupported = false; + for m in vs.matchers.matchers.iter() { + // The implicit `__name__` matcher is the metric name itself + // — we already pulled that out of the selector elsewhere. + if m.name == "__name__" { + continue; + } + match &m.op { + MatchOp::Equal => { + supported.push(LabelMatcher { + name: m.name.clone(), + value: m.value.clone(), + }); + } + // Regex / inequality matchers are correctness-relevant + // but cannot be answered by an exact postings lookup. + // Surface the flag so the caller emits a quirk + // annotation; the actual filter is applied post-decode. + MatchOp::NotEqual | MatchOp::Re(_) | MatchOp::NotRe(_) => { + has_unsupported = true; + } + } + } + (supported, has_unsupported) +} + #[cfg(test)] mod tests { use super::*; diff --git a/asap-query-engine/src/engines/gorilla_engine/tests.rs b/asap-query-engine/src/engines/gorilla_engine/tests.rs index 70b68d15..11e29d05 100644 --- a/asap-query-engine/src/engines/gorilla_engine/tests.rs +++ b/asap-query-engine/src/engines/gorilla_engine/tests.rs @@ -15,7 +15,7 @@ use async_trait::async_trait; use tokio::time::sleep; use crate::drivers::query::fallback::cold_store::{ - ChunkRef, ColdStore, ColdStoreError, RawSample, + ChunkRef, ColdStore, ColdStoreError, PostingsHits, RawSample, }; use crate::engines::query_result::QueryResult; use crate::stores::sketch_db::accuracy::{AccuracyKind, AccuracyProfile}; @@ -39,6 +39,11 @@ struct MockColdStore { /// If set, every `read_chunk` call sleeps for this duration — /// used by the timeout test. read_delay: Option, + /// **mvp/v5**: optional postings table keyed by `(label_name, + /// label_value)`. When `Some`, [`ColdStore::list_postings_for`] + /// answers from this table; when `None`, returns a "missing + /// postings" outcome (driving the fall-back path test). + postings: Option>>, } impl MockColdStore { @@ -46,6 +51,7 @@ impl MockColdStore { Self { chunks, read_delay: None, + postings: None, } } @@ -53,6 +59,16 @@ impl MockColdStore { self.read_delay = Some(d); self } + + /// mvp/v5: install a postings table for the + /// `list_postings_for` path. + fn with_postings( + mut self, + postings: BTreeMap<(String, String), Vec>, + ) -> Self { + self.postings = Some(postings); + self + } } #[async_trait] @@ -107,6 +123,49 @@ impl ColdStore for MockColdStore { chunk.key ))) } + + async fn list_postings_for( + &self, + _metric: &str, + _start_ms: i64, + _end_ms: i64, + matchers: &[(String, String)], + ) -> Result { + let Some(table) = &self.postings else { + // Mirror "real" missing-postings behaviour: the trait + // says return Unsupported when the backend doesn't + // know how to compute this. The executor treats that + // as fall-through. + return Err(ColdStoreError::Unsupported("list_postings_for")); + }; + let mut hits = PostingsHits { + series_ids: Vec::new(), + buckets_in_range: 1, + buckets_with_postings: 1, + }; + if matchers.is_empty() { + // Union of every series id in the table. + let mut set: std::collections::BTreeSet = + std::collections::BTreeSet::new(); + for ids in table.values() { + set.extend(ids.iter().copied()); + } + hits.series_ids = set.into_iter().collect(); + return Ok(hits); + } + let first = table + .get(&matchers[0]) + .cloned() + .unwrap_or_default(); + let mut acc: std::collections::BTreeSet = first.into_iter().collect(); + for m in &matchers[1..] { + let next = table.get(m).cloned().unwrap_or_default(); + let next_set: std::collections::BTreeSet = next.into_iter().collect(); + acc = acc.intersection(&next_set).copied().collect(); + } + hits.series_ids = acc.into_iter().collect(); + Ok(hits) + } } // ───────────────────────────────────────────────────────────────────── @@ -553,6 +612,9 @@ async fn result_includes_data_source_gorilla_archive() { value: 42.0, samples_scanned: 7, chunks_fetched: 2, + chunks_skipped_via_postings: 0, + postings_filtered_series_count: 0, + postings_missing: false, }; let infos = outcome.info_lines(); assert!( @@ -613,3 +675,114 @@ async fn engine_respects_config_timeout() { other => panic!("expected Timeout, got {other:?}"), } } + + +// ───────────────────────────────────────────────────────────────────── +// mvp/v5 — postings-aware path tests +// ───────────────────────────────────────────────────────────────────── + +/// Build a chunk with explicit `label_hash` so the postings-aware +/// path can prune via `series_id == label_hash`. +fn labeled_chunk( + key: &str, + label_hash: u64, + label_value: &str, + start_ms: i64, + samples: &[(i64, f64)], +) -> (ChunkRef, Vec) { + let last_ts = samples.last().map(|(t, _)| *t).unwrap_or(start_ms); + let chunk = ChunkRef { + key: key.to_string(), + metric: METRIC.to_string(), + time_range_ms: (start_ms, last_ts + 1), + label_hash, + sample_count: samples.len() as u32, + size_bytes: 0, + }; + let mut labels = BTreeMap::new(); + labels.insert("zone".to_string(), label_value.to_string()); + let raw_samples: Vec = samples + .iter() + .map(|(t, v)| RawSample { + ts_ms: *t, + labels: labels.clone(), + value: *v, + }) + .collect(); + (chunk, raw_samples) +} + +#[tokio::test] +async fn postings_aware_path_prunes_chunks() { + // Two chunks: one for zone=a (label_hash=11), one for zone=b + // (label_hash=22). Postings says zone=a → [11]. The engine + // must read only the zone=a chunk. + let (chunk_a, samples_a) = + labeled_chunk("k-a", 11, "a", NOW_MS - 30_000, &[(NOW_MS - 1_000, 5.0), (NOW_MS - 500, 5.0)]); + let (chunk_b, samples_b) = + labeled_chunk("k-b", 22, "b", NOW_MS - 30_000, &[(NOW_MS - 1_000, 99.0), (NOW_MS - 500, 99.0)]); + let mut postings: BTreeMap<(String, String), Vec> = BTreeMap::new(); + postings.insert(("zone".to_string(), "a".to_string()), vec![11]); + postings.insert(("zone".to_string(), "b".to_string()), vec![22]); + let mock = MockColdStore::new(vec![(chunk_a, samples_a), (chunk_b, samples_b)]) + .with_postings(postings); + let engine = GorillaQueryEngine::new(Arc::new(mock), cfg()); + let plan = + plan_query_at(&format!(r#"sum_over_time({METRIC}{{zone="a"}}[5m])"#), NOW_MS).unwrap(); + assert_eq!(plan.label_matchers.len(), 1); + let exec = ExactExecutor::new(engine.cold_store_for_tests(), cfg()); + let outcome = exec.execute_plan(&plan).await.unwrap(); + // Only zone=a chunk contributed: 5.0 + 5.0 = 10.0 (NOT 5+5+99+99=208). + assert_eq!(outcome.value, 10.0); + assert_eq!(outcome.chunks_fetched, 1); + assert_eq!(outcome.chunks_skipped_via_postings, 1); + assert_eq!(outcome.postings_filtered_series_count, 1); + assert!(!outcome.postings_missing); +} + +#[tokio::test] +async fn postings_missing_falls_back_to_scan_all() { + // Same chunks, NO postings table → the executor falls through + // to the scan-all path and uses the post-decode label filter + // for correctness. The `postings_missing` flag must be set. + let (chunk_a, samples_a) = + labeled_chunk("k-a", 11, "a", NOW_MS - 30_000, &[(NOW_MS - 1_000, 5.0)]); + let (chunk_b, samples_b) = + labeled_chunk("k-b", 22, "b", NOW_MS - 30_000, &[(NOW_MS - 1_000, 99.0)]); + let mock = MockColdStore::new(vec![(chunk_a, samples_a), (chunk_b, samples_b)]); + let engine = GorillaQueryEngine::new(Arc::new(mock), cfg()); + let plan = + plan_query_at(&format!(r#"sum_over_time({METRIC}{{zone="a"}}[5m])"#), NOW_MS).unwrap(); + let exec = ExactExecutor::new(engine.cold_store_for_tests(), cfg()); + let outcome = exec.execute_plan(&plan).await.unwrap(); + // Correctness: only zone=a sample (5.0) folded in. The + // post-decode filter does the work. + assert_eq!(outcome.value, 5.0); + // Both chunks were fetched — postings filter no-oped. + assert_eq!(outcome.chunks_fetched, 2); + assert_eq!(outcome.chunks_skipped_via_postings, 0); + assert!(outcome.postings_missing, "missing-postings flag must be set"); + let infos = outcome.info_lines(); + assert!(infos.iter().any(|i| i == "data_source_quirk: postings_missing")); +} + +#[tokio::test] +async fn postings_path_no_label_predicate_skips_postings_lookup() { + // No label predicate → postings filter is a no-op; the + // postings table is never consulted. Total = 5+99 = 104. + let (chunk_a, samples_a) = + labeled_chunk("k-a", 11, "a", NOW_MS - 30_000, &[(NOW_MS - 1_000, 5.0)]); + let (chunk_b, samples_b) = + labeled_chunk("k-b", 22, "b", NOW_MS - 30_000, &[(NOW_MS - 1_000, 99.0)]); + let mock = MockColdStore::new(vec![(chunk_a, samples_a), (chunk_b, samples_b)]); + let engine = GorillaQueryEngine::new(Arc::new(mock), cfg()); + let plan = plan_query_at(&format!("sum_over_time({METRIC}[5m])"), NOW_MS).unwrap(); + assert!(plan.label_matchers.is_empty()); + let exec = ExactExecutor::new(engine.cold_store_for_tests(), cfg()); + let outcome = exec.execute_plan(&plan).await.unwrap(); + assert_eq!(outcome.value, 104.0); + assert_eq!(outcome.chunks_fetched, 2); + assert!(!outcome.postings_missing); + assert_eq!(outcome.chunks_skipped_via_postings, 0); +} +