diff --git a/src/promql/AGENTS.md b/src/promql/AGENTS.md index dc90c5a0eec..3962add5e1b 100644 --- a/src/promql/AGENTS.md +++ b/src/promql/AGENTS.md @@ -16,38 +16,41 @@ operators. Statement-to-plan translation lives in `src/query/src/promql/`. | Kernels | `src/promql/src/functions.rs`, `src/promql/src/functions/` | PromQL functions | | Matching | `src/promql/src/vector_match.rs` | Binary vector matching | | Extension operators | `src/promql/src/extension_plan.rs`, `src/promql/src/extension_plan/` | Logical nodes and execution plans | -| Native model | `src/promql/src/native/` | Compiled tree, chunks, display, metrics | +| Native model | `src/promql/src/native/` | Compiled tree, chunks, display, metrics and execution | -## Two evaluation paths, and their status +## Evaluation paths -The easiest thing to get wrong here. Both exist; only one runs. - -- **Extension operators are live.** `PromPlanner` builds `SeriesDivide`, - `SeriesNormalize`, `RangeManipulate`, `InstantManipulate`, `EmptyMetric` and - `UnionDistinctOn`; `PromExtensionPlanner` - (`src/promql/src/extension_plan/planner.rs`) maps each logical node to its - execution plan, one branch per node, `None` for anything else. -- **Nothing plans into the native model yet.** `src/query/src/promql/native_compile.rs` - compiles a statement into a `NativeTree`, but no planner calls it, so a tree - built here reaches no execution. `src/promql/src/native/` holds the tree, the - chunk protocol, display and metrics, and — once a runtime lands — the driver - that runs a tree node by node. Compiling and running are separate concerns - owned by separate files; do not assume a tree is executable because it - compiles. +- **Extension operators:** the existing PromQL path plans operators such as + `SeriesDivide`, `SeriesNormalize`, `RangeManipulate`, `InstantManipulate`, + `EmptyMetric` and `UnionDistinctOn`. `PromExtensionPlanner` + (`src/promql/src/extension_plan/planner.rs`) maps the logical nodes to their + execution plans. +- **Fused native-series route:** the query planner offers a PromQL statement to + `native_compile::compile` only when the engine's + `in_process_region_transport` and `QueryOptions::experimental_native_promql` + options are enabled and the per-request `query.experimental_native_promql` + extension is true. Accepted plans use process-local `FusedSeries`; declined + or unsupported shapes fall back to the existing planner. The supported subset + requires one selector, a direct scan with a bound metric name, and no residual + label matchers or `__field__` matcher that narrows the scan, plus an eligible + raw-series table/field set and supported expression/grid. Metric scans retain + the static `__table_id` predicate. Hybrid CTEs and CREATE VIEW TQL creation + currently fall back to the existing planner. ## Change coupling -- **An extension node reaches a datanode only if three things line up**: a payload - branch in `src/common/substrait/src/extension_serializer.rs`, a physical branch - in `PromExtensionPlanner`, and a `Commutativity::Commutative` arm in - `check_extension_plan` (`src/query/src/dist_plan/commutativity.rs`). Missing the - third is silent: the node falls through to `Unsupported`, the distributed - boundary is placed above it, and it never reaches a region — with no error. +- Distributed extension operators need matching handling in the Substrait + serializer (`src/common/substrait/src/extension_serializer.rs`), the physical + `PromExtensionPlanner`, and `check_extension_plan` + (`src/query/src/dist_plan/commutativity.rs`). Process-local `FusedSeries` is + the exception: it is not serialized to a datanode. Its distributed placement + is restricted to a direct table scan, a bound metric name and trees that read + one series at a time. The single-region placement warning in + `commutativity.rs` remains important: do not assume a series stays in one + region. - A new logical node also needs its physical planner reachable from `DfQueryPlanner::create_physical_plan`, assembled in `src/query/src/query_engine/state.rs`. -- User-visible changes need planner coverage plus a sqlness case under - `tests/cases/standalone/common/promql/`. ## Gotchas @@ -72,9 +75,9 @@ The easiest thing to get wrong here. Both exist; only one runs. the producer ordering the adapter adds. Nothing in that metadata binds the key to a decode — the frame decoder compares those format keys and the field list, and never a label identity. `unpack_raw_series` therefore requires a - caller-supplied `RawLabelSource`: decode with the source that was actually bound - when the stream was produced, because the frame schema does not replace that - binding. The metadata describes format, not label identity. + caller-supplied `RawLabelSource`: decode with the source that was actually + bound when the stream was produced, because the frame schema does not replace + that binding. The metadata describes format, not label identity. - **A wrong label binding is silently wrong, not an error.** Storage validates the binding it is given (metadata identity, tag-ness, name matches, no duplicates) but cannot tell whether that set is the one the query meant, and the metric name @@ -85,26 +88,5 @@ The easiest thing to get wrong here. Both exist; only one runs. ```bash cargo nextest run -p promql cargo nextest run -p query # planner and native-compile changes -cargo sqlness bare -t # PromQL cases live under tests/cases/standalone/common/promql/ +cargo sqlness bare -t # PromQL cases under tests/cases/standalone/common/promql/ ``` - -## Not wired up yet - -The storage side of the raw path exists; what is missing is planner wiring. Both -gaps below are structural, not bugs. - -- **The native tree has no planner entry point.** `native_compile::compile` has no - caller, and a `NativeTree` becomes a physical plan only once something builds it - while it still holds the AST and the selector binding — the point where the - matchers, `__field__` and the lookback are still available. Compiling later, from - a plan, loses them. -- **No planner reaches a raw scan.** mito2 implements - `RegionScanner::raw_series_schema` (`mito2/src/read/series_scan.rs`), so - `RegionScanExec::try_to_series_scan` can build `SeriesScanExec` from a prepared - scanner; the trait's `Ok(None)` default only refuses engines that have not - implemented it. What is missing is the planner side: nothing selects that - conversion, and `SeriesScanExec::children()` is empty, so wrapping a - `RegionScanExec` before partition preparation leaves `ParallelizeScan` unable to - assign partitions. Either give the partitioner a child it can see, or convert to - the native leaf only after partition preparation — the second is the contract - `RegionScanExec::try_to_series_scan` documents. diff --git a/src/query/src/dist_plan/commutativity.rs b/src/query/src/dist_plan/commutativity.rs index 940f921c7cf..c11e78cf819 100644 --- a/src/query/src/dist_plan/commutativity.rs +++ b/src/query/src/dist_plan/commutativity.rs @@ -385,14 +385,6 @@ impl Categorizer { } } -/// Whether any node of `tree` reads more than one series at a time. -/// -/// Everything the native compiler emits today reads one series: an instant selection, a window and -/// a point-wise map each select, fold or map the samples of the series they were handed. An -/// aggregate folds a label set's series together and a binary operation pairs them, so both consume -/// several series at once. Neither has a compiler producer yet, which is why refusing them costs -/// nothing today: a tree built some other way is exactly what this check is for, because evaluating -/// it one region at a time would fold a partial input and answer a wrong query silently. #[cfg(test)] mod tests { use std::collections::{BTreeMap, BTreeSet}; diff --git a/src/table/src/table/scan.rs b/src/table/src/table/scan.rs index 38548fa0853..6e69daa00a2 100644 --- a/src/table/src/table/scan.rs +++ b/src/table/src/table/scan.rs @@ -1360,9 +1360,11 @@ mod test { } #[test] - fn assign_partition_ranges_preserves_requested_partition_count() { - let ranges = (0..4) - .map(|identifier| PartitionRange { + fn assign_partition_ranges_balances_rows_deterministically() { + let ranges: Vec<_> = [100, 200, 150, 250] + .into_iter() + .enumerate() + .map(|(identifier, num_rows)| PartitionRange { start: common_time::Timestamp::new( identifier as i64, common_time::timestamp::TimeUnit::Second, @@ -1371,11 +1373,53 @@ mod test { identifier as i64 + 1, common_time::timestamp::TimeUnit::Second, ), - num_rows: 1, - identifier, + num_rows, + identifier: identifier + 1, }) .collect(); - assert_eq!(assign_partition_ranges(ranges, 4).len(), 4); + let assigned = assign_partition_ranges(ranges.clone(), 2); + assert_eq!( + assigned + .iter() + .map(|partition| { + partition + .iter() + .map(|range| range.identifier) + .collect::>() + }) + .collect::>(), + vec![vec![4, 1], vec![2, 3]] + ); + let mut assigned_identifiers = assigned + .iter() + .flatten() + .map(|range| range.identifier) + .collect::>(); + assigned_identifiers.sort_unstable(); + assert_eq!(assigned_identifiers, vec![1, 2, 3, 4]); + + let assigned = assign_partition_ranges(ranges.clone(), 5); + assert_eq!( + assigned + .iter() + .map(|partition| { + partition + .iter() + .map(|range| range.identifier) + .collect::>() + }) + .collect::>(), + vec![vec![4], vec![1], vec![2], vec![], vec![3]] + ); + let mut assigned_identifiers = assigned + .iter() + .flatten() + .map(|range| range.identifier) + .collect::>(); + assigned_identifiers.sort_unstable(); + assert_eq!(assigned_identifiers, vec![1, 2, 3, 4]); + + assert_eq!(assign_partition_ranges(vec![], 5), vec![vec![]; 5]); } #[test]