rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-11-14 23:06:59 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(query): match a partitioned multi-selector tree above its sources (#10226)

Browse Source 
* fix(query): match the whole operand instead of copying matchers

A native binary must report what Prometheus reports: when one side holds two
series under one matching label set, the query fails with "found duplicate
series for the match group ... on the right hand-side". The legacy route
answers a value instead, because `matching_filters::propagate` copies the
matcher into the other operand's scan and the groups without a partner are
dropped before the join - its own documentation says that hides the duplicate.
This branch had started applying that rewrite to native operands as well, and
had refused multi-selector trees over partitioned tables in process to force
the hybrid topology back. Both kept a legacy behaviour the native route does
not need: a native binary reads independent raw sources and checks the whole
one-side groups itself.

Drop the rewrite for native operands, and keep the refusal only for several
scans of one partitioned table, whose node cannot run inside a region. The
partitioned acceptance case now asserts the duplicate error and that the pair
really runs as one native binary; the legacy value stays in the test as the
contrast that documents the difference.

Signed-off-by: discord9 <discord9@163.com>

* test(promql): assert the native binary, not the hybrid join

The hybrid cases were written when a cross-table pair over a partitioned table
could only run as two region-local producers joined on the frontend. The pair
now runs as one native binary over independent raw sources, so those shape
assertions describe a topology that no longer exists: two fused nodes, a
frontend join, direct region scans under it and a projection above it.

Assert what the native plan actually promises instead: the binary is one fused
node with more than one input, every operand reads its own source, and the
duplicate-group and value assertions stay as they are. The per-scan-partition
value check now applies to operand producers only - a matching binary routes by
label set, so an input partition may contribute nothing to a given output one.

Signed-off-by: discord9 <discord9@163.com>

* test(promql): keep the operand metrics for a native binary

The per-scan-partition value check ran over every fused node, which now means
the binary itself: it routes by label set, so an output partition may
legitimately hold nothing from a given input partition. Check the two operand
metrics instead - the same assertion the unpartitioned pair already makes - so
each operand still proves it registered its values on every partition it reads.

Signed-off-by: discord9 <discord9@163.com>

* feat(query): admit multi-selector trees the native binary can match

Three refusals were left over from when a multi-selector tree could only run
inside one region over bare scans, and each of them now only costs coverage:

- The aggregate split and the segmentation were admission authority: a tree
  that needed placement and did not fit the single-selector split fell back,
  even though each selector occurrence can read its own raw source and the
  consumer can run above the collection. They are optimizations now.
- `multi_input` meant "the metric logical ids differ", so ordinary tables,
  repeated selectors and mixed engines were treated as a single input. It now
  means what the code needs: more than one independent input, one per selector
  occurrence, each bound to its own source.
- Several scans of one partitioned table were refused in process. That node
  cannot run inside a region, but the same tree has a frontend placement over
  per-occurrence raw sources, so only the cross-table refusal that keeps the
  hybrid route is left.

Tests that asserted the fallback now assert the native plan and the per-source
inputs; the value, label and duplicate-group assertions are unchanged, and the
shapes the compiler genuinely cannot express stay refused.

Signed-off-by: discord9 <discord9@163.com>

* fix(query): keep the raw sources under an in-process cut

A table partitioned on a column that is not a tag splits one series across
regions. When such a tree is cut into segments in process, the cut could fall
back to the bare scan, and then a per-region prefix evaluated a window over one
region's fragments: `rate` never saw the samples the other region holds, so a
window that straddles the boundary produced no value at all. Remote execution
already kept the raw source under the cut; in process the flag said otherwise.

Pass the boundary fact that matters - the cut must keep each source when the
table is split into regions - so the evaluated segment runs the prefix above
the reconciliation. An unpartitioned table's scans are whole in one region, so
its cuts keep the bare-scan form and the two-aggregate pairs still plan as one
execution.

The regression puts the two samples of one series on either side of a
time-partition boundary and compares the native answer with the existing plan's
for a two- and a three-selector window; without this change both lose the
value.

Signed-off-by: discord9 <discord9@163.com>

* test(promql): prove the window runs natively

The time-boundary regression compared rows with the existing plan, which a
fallback also satisfies. Require a fused node in the frontend tree as well, so
the test fails if the planner quietly declines the shape.

Signed-off-by: discord9 <discord9@163.com>

* test(query): trim redundant gates and assertions

A review of this branch found duplications rather than missing coverage: two
identical metric-name gates, a `!multi_input` term that now implies its own
`selector.len() == 1` partner, a duplicate aggregate-admission test, a
redundant cooperative-decorator special case, a tautological region-input
assertion, an exact-child-type assertion that pins an intermediate shape, and a
window regression that accepted a region-side fused node as proof of frontend
admission. Remove them; the behaviour assertions stay.

Signed-off-by: discord9 <discord9@163.com>

* test(promql): require the native binary for a partitioned pair

The values equal the legacy route's, so a legacy frontend join reading the same
regions could satisfy the assertions on their own. Require the fused node in
the frontend tree as well, and drop a duplicated rustdoc paragraph.

Signed-off-by: discord9 <discord9@163.com>

* fix(query): drop the stale unconditional canonical refusal

A rebase left both the old refusal (`any_partitioned && contains_canonical_merge`)
and the transport-scoped one, so the older block declined every canonical merge
over a partitioned table on the wire and the acceptance case fell back to the
existing planner. Keep the scoped refusal only.

Signed-off-by: discord9 <discord9@163.com>

---------

Signed-off-by: discord9 <discord9@163.com>
This commit is contained in:
discord9
2026-11-06 19:23:07 +08:00
committed by GitHub
parent b4c9a01b09
commit 4e024452fe
5 changed files with 368 additions and 455 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/query/src/promql/planner.rs
+26 -83
View File
@@ -22,8 +22,6 @@ pub use native::{NATIVE_PROMQL_EXTENSION_KEY, discovers_metric_name};
mod set_operator;
use std::collections::{BTreeSet, HashSet, VecDeque};
use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;
use arrow::datatypes::IntervalDayTime;
@@ -310,11 +308,6 @@ pub struct PromPlanner {
type BinaryFieldPair<'a> = (&'a String, &'a String);
/// [`PromPlanner::try_plan_native_operand`]'s future, boxed; see
/// [`PromPlanner::native_operand_future`] for why it is not kept inline.
type NativeOperandFuture<'a> =
Pin<Box<dyn Future<Output = Result<Option<(LogicalPlan, PromPlannerContext)>>> + Send + 'a>>;
impl PromPlanner {
pub async fn stmt_to_plan(
table_provider: DfTableSourceProvider,
@@ -1177,63 +1170,6 @@ impl PromPlanner {
})
}
/// Boxes [`Self::try_plan_native_operand`]'s future.
///
/// The planner re-enters [`Self::prom_expr_to_plan_inner`] and
/// [`Self::prom_binary_expr_to_plan`] once per nesting level of the statement, and both frames
/// are large in a debug build. The native route's future is large as well, so a caller that
/// also plans the same expression with the existing path would carry a copy of it on the stack
/// at every level. Built here, the future moves into the box before that recursion starts, and
/// only the attempt itself pays for it.
fn native_operand_future<'a>(
&'a mut self,
expr: &'a PromExpr,
query_engine_state: &'a QueryEngineState,
) -> NativeOperandFuture<'a> {
Box::pin(self.try_plan_native_operand(expr, query_engine_state))
}
/// Re-plans one operand that [`matching_filters::propagate`] rewrote, publishing the plan and
/// its context through `context`.
///
/// A copied matcher belongs to the scan, not to the operand's identity: `absent()` turns
/// `selector_matcher` into the labels it reports, so the re-planned context keeps the matchers
/// the operand was written with.
///
/// A natively planned operand is re-planned natively: the matcher has to reach the fused node's
/// own scan, or the join keeps the non-joining rows that node reported. The route decides per
/// expression, so a decline falls back to the existing plan exactly as it would have on its own.
async fn replan_rewritten_operand(
&mut self,
rewritten: &PromExpr,
used_native: bool,
query_engine_state: &QueryEngineState,
context: &mut PromPlannerContext,
) -> Result<LogicalPlan> {
let selectors = std::mem::take(&mut context.selector_matcher);
let native = if used_native {
self.native_operand_future(rewritten, query_engine_state)
.await?
} else {
None
};
let plan = match native {
Some((plan, replanned)) => {
*context = replanned;
plan
}
None => {
let plan = self
.prom_expr_to_plan(rewritten, query_engine_state)
.await?;
*context = self.ctx.clone();
plan
}
};
context.selector_matcher = selectors;
Ok(plan)
}
async fn prom_binary_expr_to_plan(
&mut self,
query_engine_state: &QueryEngineState,
@@ -1483,32 +1419,39 @@ impl PromPlanner {
// Both operands are planned first because only the planned contexts tell a tag
// from a value field. The rewrite merely adds matchers to a selector, so the
// table reference, time index and field columns captured above stay valid.
if let Some(rewritten) = matching_filters::propagate(
binary_expr,
&left_context.tag_columns,
&right_context.tag_columns,
&left_context.aggregation_field_labels,
&right_context.aggregation_field_labels,
) {
// Matcher propagation belongs to the legacy route: copying a matcher into the
// other operand's scan drops the groups that have no partner, which is exactly
// what lets the legacy join answer where Prometheus would report a duplicated
// match group. A native operand must not be rewritten this way - the native
// matcher checks the complete one-side groups itself, and filtering them first
// would hide that error. So the rewrite only runs when neither operand is native.
if !used_native
&& let Some(rewritten) = matching_filters::propagate(
binary_expr,
&left_context.tag_columns,
&right_context.tag_columns,
&left_context.aggregation_field_labels,
&right_context.aggregation_field_labels,
)
{
// A copied matcher belongs to the scan, not to the operand's identity:
// `absent()` turns `selector_matcher` into the labels it reports, so the
// re-planned context keeps the matchers the operand was written with.
if rewritten.lhs.as_ref() != lhs.as_ref() {
let selectors = std::mem::take(&mut left_context.selector_matcher);
left_input = self
.replan_rewritten_operand(
&rewritten.lhs,
used_native,
query_engine_state,
&mut left_context,
)
.prom_expr_to_plan(&rewritten.lhs, query_engine_state)
.await?;
left_context = self.ctx.clone();
left_context.selector_matcher = selectors;
}
if rewritten.rhs.as_ref() != rhs.as_ref() {
let selectors = std::mem::take(&mut right_context.selector_matcher);
right_input = self
.replan_rewritten_operand(
&rewritten.rhs,
used_native,
query_engine_state,
&mut right_context,
)
.prom_expr_to_plan(&rewritten.rhs, query_engine_state)
.await?;
right_context = self.ctx.clone();
right_context.selector_matcher = selectors;
}
// The code below reads `self.ctx` as the right operand's context.
self.ctx = right_context.clone();
src/query/src/promql/planner/native.rs
+82 -81
View File
@@ -864,17 +864,11 @@ impl PromPlanner {
.map(|(scan, ..)| scan)
.filter(|scan| !is_empty_discovered_input(scan)),
) || logical_ids_differ;
// A multi-selector tree over a partitioned table has no in-process placement whose matching
// and uniqueness semantics equal the frontend route. One node over several scans of one
// partitioned table reads neither the other scans nor the fragments another region holds,
// and one node over several tables' sources matches every series those sources delivered,
// where the frontend route copies the operands' matchers into their scans first and checks
// uniqueness above the join. Keep the existing plan in process; the hybrid route lifts such
// a tree one operand at a time. Over the wire every occurrence is already its own raw source
// reconciled above the regions, so the route keeps those trees.
if selectors.len() > 1 && !self_contained && prepared.iter().any(|input| input.5) {
return Ok(None);
}
// A multi-selector tree over a partitioned table has no region-local placement: one node
// over several scans reads neither the other scans nor the fragments another region holds.
// It does have a frontend placement - each occurrence reads the table through its own raw
// source and the node matches above every region's contribution - so the tree is admitted
// on both routes and nothing is refused here.
// The first input names the output's time index and, on one table, binds every scan.
let Some((_, _, metric, time, tags, _, _)) = prepared.first() else {
return Ok(None);
@@ -1054,6 +1048,43 @@ impl PromPlanner {
// region-local producer cannot see the pieces another region holds, so this fact decides
// whether that boundary may be claimed at all.
let any_partitioned = prepared.iter().any(|input| input.5);
// Admission and placement are separate: `independent_scan_pair` admits nested native
// matches, while the physical locality classification decides whether their inputs need
// gathering or segmentation.
// A label rewrite can give two series one label set, which every consumer that answers a
// whole label set has to see as one series. The pass puts a [`NativeExpr::Merge`] where that
// matters and nowhere else, so a tree with no collapsing rewrite keeps every partition.
let tree = match promql::native::canonicalize::merge_label_rewrites(
&compiled.tree,
&source_tags,
) {
Ok(tree) => tree,
Err(_) => return Ok(None),
};
let root = tree.root();
// The placement facts the split and the segmentation below are chosen from, and the same
// facts that say whether one node can run over the scans at all.
let per_partition =
!reads_across_series(root) || pointwise_operand(root) || independent_scan_pair(root);
let locality = input_locality(root);
let computed_key_case =
matches!(locality, InputLocality::SinglePartition) && matching_operands_drop_key(root);
let required_placement = !per_partition || computed_key_case;
// The node reads more than one input when the tree has more than one selector occurrence:
// the compiler binds one input per occurrence. What it needs from them is that each is
// bound to its own source - a raw source, in process or over the wire - or, where a region
// can run the node itself, one bare scan per occurrence. Either way the consumer above them
// reads every occurrence's contribution, so the metric-id fact this used to turn on is not
// what admission turns on.
let multi_input = selectors.len() > 1;
// A region runs one node over its own scans only when the table it reads is not split into
// regions and the tree's root is per-partition work: a partitioned table's region reads
// neither the other scans nor the fragments another region holds, and a whole-vector
// consumer has no region-local placement. Such a tree runs above the regions, and then each
// occurrence reads through its own raw source.
let per_source_inputs = cross_table
|| self_contained
|| (multi_input && (any_partitioned || required_placement));
// A segment carries one source's identity. Keeping each occurrence's scan and metadata
// aligned here lets the cutting step name that source independently of selector count.
// Read the split-series fact before `prepared` is consumed by the child mapping below.
@@ -1090,10 +1121,12 @@ impl PromPlanner {
// A scan stays bare only where a region runs the node over it in process. Over the
// wire every input is a raw source, whichever table it reads: the source is what the
// transport carries, and it is also what reconciles the fragments of a series that
// several regions hold.
// A partition that splits one series across regions has to be read through a raw
// source even in process: a bare scan would hand a region only its own fragment.
if (!cross_table && !self_contained && !split_series) || already_sourced {
// several regions hold. A partition that splits one series across regions has to
// be read through a raw source even in process: a bare scan would hand a region only
// its own fragment. A tree the regions cannot run as one node reads its occurrences
// through their own sources too, so the consumer above matches every region's
// contribution instead of the rows of one.
if (!per_source_inputs && !split_series) || already_sourced {
return Ok(scan);
}
let Some(metric) = metric else {
@@ -1175,49 +1208,27 @@ impl PromPlanner {
}
Some(FieldKind::Float) | None => FieldKind::Float,
};
// The split is only ever a one-child plan: it exports each partition's leaf state and merges
// them above the region. A tree over two metric logical tables of ONE physical table is not
// split, because a node whose locality is `SinglePartition` normalizes every input to one
// partition, so every group is folded in one place. Both of those inputs are `RawSeriesSource` children
// (the two ids made `cross_table` true), each answering from its own region with frames that
// carry their own labels, so the node is plannable above both. That plan needs no split to be
// correct, so it falls through to the node below.
// A tree over more than one input needs no split: each input already answers from its own
// source, and the one node above them folds every partition in one execution. The split is
// an optimization for the one-input shapes it can carry, and so is the segmentation below;
// the node over the inputs is what admits the rest.
//
// A multi-selector tree whose inputs stay direct scans of one table is not admitted here: its
// scans are not raw sources, so the categorizer refuses to place the node region-locally
// (`Categorizer::check_extension_plan` is `NonCommutative` for a cross-series node over more
// than one scan) and the physical planner cannot bind a merged plain scan under it either. So
// such a tree keeps its one-selector-or-decline behaviour, unchanged.
//
// A selector whose scan carries no metric logical id - a Mito table's, for one - adds no
// id, so one Mito aggregate beside one metric aggregate is not a multi-input tree:
// `sum(m) + sum(mito)` keeps the existing plan. A second metric logical id does make the
// tree multi-input, and the one execution above then reads the Mito input through its own
// raw source as well.
let multi_input = selectors.len() > 1 && logical_ids_differ;
// The aggregate split folds float state, and the multi-input node reads several scans, so
// a histogram keeps the existing plan for both rather than running a tree whose output the
// runtime could not finish. The kind entering each aggregate is what decides, not the
// The aggregate split folds float state, and a node reading several sources runs over
// frames it reconciled rather than over its own scans, so a histogram keeps the existing
// plan for both rather than running a tree whose output the runtime could not finish. A
// node over bare scans is not that shape: a region runs it over each scan itself, exactly
// as it does for one selector. The kind entering each aggregate is what decides, not the
// shape of the root: `sum(h) * 2` has a binary root and would otherwise hide the aggregate
// under it.
if let Some(kind @ FieldKind::Histogram) = statement_kind
&& (multi_input || aggregate_reads_histogram(root, kind))
&& ((multi_input && per_source_inputs) || aggregate_reads_histogram(root, kind))
{
return Ok(None);
}
// A multi-input tree skips the split below, so the refusals the split makes for reasons that
// are not about what the runtime computes have to be made here.
//
// The aggregate split keeps its prefix per-series, which is why it declines an operation
// whose answer depends on the order its input arrives in. A multi-input node reads its
// inputs' partitions in the order they became ready, so that refusal is repeated here.
if (multi_input || (self_contained && children.len() > 1)) && depends_on_input_order(root) {
return Ok(None);
}
// Two sources that report groups by metric name need a column for that name: without one,
// groups that differ only by metric would be reported as the same label set. A discovered
// selector carries that column itself; several explicit selectors do not.
if (multi_input || (self_contained && children.len() > 1)) && groups_by_metric_name(root) {
// The split keeps its prefix per-series, which is why it declines an operation whose
// answer depends on the order its input arrives in. A node over more than one input reads
// its inputs' partitions in the order they became ready, so that refusal holds there too.
if multi_input && depends_on_input_order(root) {
return Ok(None);
}
// The native output schema can name only compiled value fields, the time index and the tag
@@ -1246,8 +1257,7 @@ impl PromPlanner {
// series frames, and a canonical router collects the pieces of one label set from every
// partition. Declining here would hand the statement back to a path that answers it per
// partition.
if !multi_input
&& selectors.len() == 1
if selectors.len() == 1
&& matches!(&tree.root().expr, NativeExpr::Merge { input } if pointwise_operand(input) && canonical_local_prefix(input) && !contains_global_consumer(input))
&& let Some(plan) = canonical_boundary_plan(
prom_expr,
@@ -1263,13 +1273,6 @@ impl PromPlanner {
{
return Ok(Some((plan, single_input_context)));
}
let root = tree.root();
let per_partition =
!reads_across_series(root) || pointwise_operand(root) || independent_scan_pair(root);
let locality = input_locality(root);
let computed_key_case =
matches!(locality, InputLocality::SinglePartition) && matching_operands_drop_key(root);
let required_placement = !per_partition || computed_key_case;
if required_placement {
// A tree whose fold can be split keeps that plan: it exports aggregate state rather
// than whole samples, so it moves less between the two halves than a sample boundary
@@ -1279,7 +1282,7 @@ impl PromPlanner {
// without this route. Its producer reads whatever input the statement prepared: over
// the wire that is the raw series source, and the partial/final halves stay on the
// frontend above the regions, so per-series work keeps the scan width.
let split = if !multi_input && selectors.len() == 1 {
let split = if selectors.len() == 1 {
FusedSeries::split_aggregate(&tree, &time).context(DataFusionPlanningSnafu)?
} else {
None
@@ -1319,11 +1322,6 @@ impl PromPlanner {
});
return Ok(Some((plan, single_input_context)));
}
// Preserve the established refusal for repeated same-table operands unless
// segmentation is needed to preserve a computed matching key.
if selectors.len() != 1 && !cross_table && !computed_key_case {
return Ok(None);
}
}
// A segment over the wire keeps the raw source as its input rather than a bare scan: the
// segment's prefixes run on the frontend above the regions, so the raw series source is
@@ -1344,22 +1342,27 @@ impl PromPlanner {
FusedOutputFormat::PlainRows,
contains_canonical_merge(root),
root_kind,
// A split-series table must keep the raw source under the cut: a bare-scan segment
// would evaluate a window over one region's fragment on the frontend.
self_contained || any_split_series,
// A cut must keep the source whenever the table behind it can split a series
// across regions: a per-region prefix would evaluate a window over one region's
// fragment, so `rate` never sees the samples the other region holds. That is a
// partitioned table whose occurrences are source-backed, and always a partition on
// a column that is not a tag. An unpartitioned table's scans are whole in one
// region, so its cuts keep the bare-scan form.
self_contained || any_split_series || (per_source_inputs && any_partitioned),
))
.await?;
if let Some(plan) = plan {
return Ok(Some((plan, single_input_context)));
}
// A raw scan alone does not prove a region contains all contributions required by a
// cross-series consumer. Without a split or segmentation, keep that consumer out of
// the region rather than treating its local gather as a global answer.
// One input with no split or segment above it leaves a raw scan doing the work of a
// cross-series consumer, and a single scan does not prove its region holds every
// contribution the consumer needs. Keep that consumer on the existing plan. A tree over
// more than one input is not this shape: every occurrence reaches it through its own
// source, so the node itself is the collection.
if required_placement && !multi_input {
return Ok(None);
}
}
// Multi-input fused execution already reads every input in this execution.
// The tags are asked for before the fields, because the walk answers for a tree the
// compiler produced: a scalar root has no labels at all, and a sample root's are the ones
// the walk derives. If it cannot derive them, this route declines.
@@ -1387,14 +1390,12 @@ impl PromPlanner {
return Ok(None);
}
let output_schema = fused_output_schema(&root_fields, &output_tags, &time, root_kind)?;
// Across tables the region scans are bound by their `RawSeriesSource` children, and this is
// also the binding a node over more than one input needs: such a node reads frames that
// already carry their own labels, so it binds neither tags nor one metric name of its own -
// each child names the metric it reads. A multi-input plan whose children stay direct
// scans reads one table through the one metric and tag set they share, and keeps this
// binding, which is the same identity those scans carry. The same holds over the wire,
// where every input is a raw source.
let (tags, metric) = if cross_table || self_contained {
// The children a node reads bind it: a `RawSeriesSource` child carries the frames and names
// the metric it reads, so a node over them binds neither tags nor one metric name of its
// own. A node whose children stay direct scans of one table reads that table through the
// one metric and tag set they share, and keeps this binding, which is the same identity
// those scans carry. The same holds over the wire, where every input is a raw source.
let (tags, metric) = if per_source_inputs {
(Vec::new(), None)
} else {
(tags, metric)
src/query/src/promql/planner/test.rs
+87 -75
View File
@@ -3787,37 +3787,18 @@ async fn native_fused_same_table_nested_match_keeps_prior_admission() {
);
}
/// On the wire, partitioned cross-table operands are independently transported by raw sources:
/// the pair is one fused node whose two inputs each name the metric they read. In process the same
/// pair keeps the frontend hybrid route instead, because only that route copies the matcher's
/// labels into each operand's scan and checks uniqueness above the join. Repeated scans of one
/// partitioned table retain the region-local multi-scan refusal on both routes.
/// Partitioned operands are independently transported by raw sources: the pair is one native
/// binary whose two inputs each name the metric they read. Repeated selectors of one partitioned
/// table read that table once per occurrence too, so no region is handed a node over bare scans.
#[tokio::test]
async fn native_fused_partitioned_pair_requires_per_selector_raw_sources() {
let state = QueryEngineState::try_new_with_region_transport(
new_memory_catalog_manager().unwrap(),
None,
None,
None,
None,
None,
let plan = build_native_fused_plan_with_provider(
"some_metric / other_metric",
true,
false,
Plugins::default(),
QueryOptions::default(),
)
.unwrap();
let eval_stmt = build_eval_stmt("some_metric / other_metric");
let plan = PromPlanner::stmt_to_plan_with_annotations(
native_two_table_provider(vec![0]).await,
&eval_stmt,
&state,
None,
true,
)
.await
.unwrap();
let fused = find_fused_series(&plan).expect("cross-table pair is native on the wire");
.await;
let fused = find_fused_series(&plan).expect("cross-table pair is native");
assert_eq!(fused.inputs().len(), 2, "{plan}");
for input in fused.inputs() {
assert!(
@@ -3826,26 +3807,25 @@ async fn native_fused_partitioned_pair_requires_per_selector_raw_sources() {
"{plan}"
);
}
// The same pair in process must not become one two-input node: the frontend hybrid route is
// what the matcher propagation and the uniqueness check live in, and that is the plan the
// partitioned-RHS hybrid cases assert.
let in_process = build_native_fused_plan_with_provider(
"some_metric / other_metric",
true,
native_two_table_provider(vec![0]).await,
)
.await;
assert!(
!has_pair_node(&in_process),
"an in-process partitioned pair keeps the hybrid route: {in_process}"
);
// Repeated selectors of one partitioned table are admitted the same way: each occurrence reads
// that table through its own raw source, so the match runs above the regions rather than
// inside one of them.
let repeated = build_native_fused_plan_with_provider(
"some_metric / some_metric",
true,
native_two_table_provider(vec![0]).await,
)
.await;
assert!(!has_pair_node(&repeated), "{repeated}");
let repeated_fused =
find_fused_series(&repeated).unwrap_or_else(|| panic!("repeated selectors: {repeated}"));
assert_eq!(2, repeated_fused.inputs().len(), "{repeated}");
for input in repeated_fused.inputs() {
assert!(
matches!(input, LogicalPlan::Extension(extension)
if extension.node.as_any().is::<RawSeriesSource>()),
"each occurrence reads its own source: {repeated}"
);
}
}
/// `some_metric` as [`native_two_table_provider`] builds it, and `other_metric` whose tags are
@@ -4193,17 +4173,28 @@ async fn native_fused_three_selector_trees_bind_every_occurrence() {
}
}
/// More than two same-table selectors on a partitioned table must not be placed as a
/// multi-child region-local node. Cross-table occurrences may each use a raw source.
/// More than two same-table selectors on a partitioned table each read that table through their own
/// source, so the match runs above the regions. No region is handed a node over the bare scans, which
/// is the shape it refuses.
#[tokio::test]
async fn native_fused_same_table_three_selectors_on_partitioned_table_fall_back() {
async fn native_fused_same_table_three_selectors_on_partitioned_table_bind_their_sources() {
let plan = build_native_fused_plan_with_provider(
"some_metric + some_metric + some_metric",
true,
native_two_table_provider(vec![0]).await,
)
.await;
assert!(!has_pair_node(&plan), "{plan}");
let fused = find_fused_series(&plan).unwrap_or_else(|| panic!("{plan}"));
assert!(fused.inputs().len() > 1, "{plan}");
assert_eq!(
vec![
"some_metric".to_string(),
"some_metric".to_string(),
"some_metric".to_string()
],
fused_input_metrics(&plan),
"each selector occurrence reads its own source: {plan}"
);
}
/// Find the final half of a split aggregate.
@@ -4897,9 +4888,10 @@ async fn native_classic_histogram_provider() -> DfTableSourceProvider {
)
}
/// Two classic-histogram functions over raw series, each admitted as one native operand: a
/// `histogram_quantile` reads each histogram off the bucket series that share its other labels and
/// drops `le`, so neither operand keeps `le` in its binding nor in the statement's output.
/// Two classic-histogram functions over raw series: a `histogram_quantile` reads each histogram off
/// the bucket series that share its other labels and drops `le`, so neither operand keeps `le` in
/// its binding nor in the statement's output. Each occurrence reads its own source, so the division
/// matches above both of them instead of joining two separately fused operands.
#[tokio::test]
async fn native_fused_histogram_quantile_binary_drops_le_from_operand_matching() {
let provider = native_classic_histogram_provider().await;
@@ -4910,9 +4902,15 @@ async fn native_fused_histogram_quantile_binary_drops_le_from_operand_matching()
)
.await;
let display = plan.display_indent().to_string();
// The whole statement has two same-table selectors and declines at that guard. Each
// histogram_quantile is nevertheless a one-selector operand and is admitted by the hybrid path.
assert_eq!(4, display.matches("FusedSeries").count(), "{display}");
// One node over one segment per selector occurrence: the two reads are independent, so the
// division matches above them rather than below a join of two fused operands.
let fused = find_fused_series(&plan).unwrap_or_else(|| panic!("{display}"));
assert_eq!(2, fused.inputs().len(), "{display}");
assert_eq!(
vec!["m".to_string(), "m".to_string()],
fused_input_metrics(&plan),
"each selector occurrence reads its own source: {display}"
);
assert!(
!display.contains(".le ="),
"join key still references dropped `le`: {display}"
@@ -4952,11 +4950,13 @@ async fn native_fused_histogram_quantile_binary_drops_le_from_operand_matching()
}
}
/// A shape the split cannot keep correct keeps the existing plan: a binary operation whose sibling
/// is data would match one side's series before the merge, a sibling a call computes has no native
/// node at all, and a group by a value field cannot be named by the state rows. A literal scalar
/// sibling is not one of them: the operation is computed per series below the fold, or rebuilt over
/// the merged values above it.
/// A shape neither the split nor the node can keep correct keeps the existing plan: a sibling a
/// call computes has no native node at all, a group by a value field cannot be named by the state
/// rows, and a kernel the compiler does not fold keeps it too.
///
/// `sum(m) / m` and `m + sum(m)` are not on this list. The fold and the plain selector are two
/// occurrences, so each reads its own source and the operation above them answers over both; the
/// split is only the optimization the one-input shapes keep.
///
/// `timestamp(sum(m))` and `sum(timestamp(abs(m)))` are not on this list. `timestamp` over a
/// computed vector reports the evaluation time, which is what Prometheus reports too - an aggregate
@@ -4968,10 +4968,6 @@ async fn native_fused_unsplittable_aggregate_falls_back() {
for query in [
"quantile(0.5, some_metric)",
"sum(quantile(0.5, some_metric))",
// The sibling is data rather than a literal scalar, so the operation would pair each series
// with the aggregate's series before the merge.
"sum(some_metric) / some_metric",
"some_metric + sum(some_metric)",
// A scalar a call computes is not a literal the split could keep: `scalar(...)` has no
// native node, so the statement keeps the existing plan.
"sum(some_metric) / scalar(some_metric)",
@@ -4997,6 +4993,38 @@ async fn native_fused_unsplittable_aggregate_falls_back() {
}
}
/// A fold beside a plain selector is two occurrences, so each reads its own source and the fold and
/// the match run above both. No partition needs to hold the whole of the input for that: the inputs
/// are reconciled before the node evaluates them.
#[tokio::test]
async fn native_fused_aggregate_above_an_independent_selector_is_admitted() {
for query in [
"sum(some_metric) / some_metric",
"some_metric + sum(some_metric)",
] {
let native = build_native_fused_plan(query, true).await;
let fused = find_fused_series(&native).unwrap_or_else(|| panic!("`{query}`: {native}"));
assert_eq!(2, fused.inputs().len(), "`{query}`: {native}");
assert_eq!(
vec!["some_metric".to_string(), "some_metric".to_string()],
fused_input_metrics(&native),
"`{query}`: each selector occurrence reads its own source: {native}"
);
assert!(
find_final_split(&native).is_none(),
"`{query}`: a tree over two inputs is not a one-input split: {native}"
);
let without_flag = build_native_fused_plan(query, false).await;
assert!(
!without_flag
.display_indent()
.to_string()
.contains("FusedSeries"),
"`{query}` takes the native plan only when it is enabled: {without_flag}"
);
}
}
/// Suffixes whose answers need the whole vector cannot fuse into a partition-local `Final`, so
/// the split declines them: it neither exports aggregate state nor runs a `Final` aggregate.
/// Ranking (`topk`) and a second aggregate cannot be answered from a partition's groups alone.
@@ -5216,22 +5244,6 @@ async fn native_fused_preparation_decline_preserves_legacy_plan() {
);
}
/// An aggregate is answered natively now (`sum(rate(..))`, `avg by(..)(..)` are split), so what
/// still falls back here is the shape the split cannot carry: a fold over two actual sources, which
/// no partition holds the whole of.
#[tokio::test]
async fn native_fused_aggregate_falls_back_to_legacy_plan() {
for query in ["some_metric + sum(some_metric)"] {
let plan = build_native_fused_plan(query, true).await;
let display = plan.display_indent().to_string();
assert!(!display.contains("FusedSeries"), "{query}: {display}");
assert!(find_final_split(&plan).is_none(), "{query}: {display}");
assert!(display.contains("Aggregate:"), "{query}: {display}");
let legacy = build_native_fused_plan(query, false).await;
assert_eq!(display, legacy.display_indent().to_string(), "{query}");
}
}
#[tokio::test]
async fn native_fused_subquery_falls_back_to_legacy_plan() {
for query in [
tests-integration/src/tests/native_aggregate_test.rs
+89 -171
View File
@@ -97,7 +97,6 @@ use crate::standalone::GreptimeDbStandaloneBuilder;
/// crate-private, so it is repeated here.
const NATIVE_PROMQL_EXTENSION_KEY: &str = "query.experimental_native_promql";
/// The physical name of the fused producer.
const FUSED_EXEC_NAME: &str = "FusedSeriesExec";
/// The metric a fused producer registers for the values its own root node produced, per partition.
const ROOT_OUTPUT_VALUES_METRIC: &str = "native.root.output_values";
/// The number of sample positions consumed by the root node, per partition.
@@ -567,38 +566,38 @@ async fn native_hybrid_binary_05_06_08() {
.map(|row| row.value.unwrap())
.collect::<Vec<_>>()
);
// The native route matches the whole right operand and checks the one-side groups itself, so
// it reports the duplicated `other` group that Prometheus reports. The legacy value above is
// the one that differs: copying the left matcher into the right scan drops every group without
// a partner, and a dropped group cannot be reported as duplicated. A partition column on the
// right table does not change that - the pair reads independent raw sources either way.
let mark = observer.mark();
let (hybrid, capture) = run_query(&instance, &propagated, true, &observer).await;
assert_eq!(
vec![5.0],
hybrid
.rows
.iter()
.map(|row| row.value.unwrap())
.collect::<Vec<_>>()
);
assert_eq!(
2,
capture.fused.len(),
"both binary operands use native execution"
);
let plans = observer.since(mark);
let frontend = plans
.iter()
.find(|plan| find_binary_join(plan).is_some())
.expect("frontend Join plan");
let join = find_binary_join(frontend).unwrap();
let uniqueness = join
.children()
.iter()
.map(|branch| uniqueness_operators_above_merge(branch))
.fold((false, false), |(window, filter), found| {
(window || found.0, filter || found.1)
});
let mut query_ctx = QueryContext::with_db_name(None);
query_ctx.set_extension(QUERY_PARALLELISM_HINT, SCAN_PARTITIONS.to_string());
query_ctx.set_extension(NATIVE_PROMQL_EXTENSION_KEY, "true".to_string());
let output = instance
.do_query(&tql_eval(&propagated), Arc::new(query_ctx))
.await
.remove(0)
.expect("the query plans");
let error = match output.data {
OutputData::Stream(stream) => collect_batches(stream)
.await
.expect_err("a duplicate match group fails"),
other => panic!("`{propagated}` was expected to stream: {other:?}"),
};
assert!(
uniqueness.0 && uniqueness.1,
"on(job) FE uniqueness operators: {}",
displayable(frontend.as_ref()).indent(false)
error.to_string().contains(
"found duplicate series for the match group {job=\"other\"} on the right hand-side"
),
"{propagated}: {error}"
);
// The statement is admitted natively: a fallback to the legacy plan would have answered the
// value instead of reporting the duplicate.
let capture = NativeCapture::from_plans(&observer.since(mark));
assert!(
capture.fused.iter().any(|node| node.children().len() == 2),
"the partitioned pair runs as one native binary"
);
// The left selector is a nonempty single `api` series, while the unpartitioned RHS also
@@ -771,7 +770,6 @@ async fn check_reduced_label_matching(instance: &Arc<Instance>, observer: &Nativ
);
let (legacy, legacy_capture) = run_query(instance, &unique_query, false, observer).await;
assert!(legacy_capture.fused.is_empty());
let mark = observer.mark();
let (native, capture) = run_query(instance, &unique_query, true, observer).await;
assert_eq!(
legacy.rows.len(),
@@ -796,43 +794,15 @@ async fn check_reduced_label_matching(instance: &Arc<Instance>, observer: &Nativ
native.rows.len(),
"the matching instance filters leave one matching job"
);
assert_eq!(2, capture.fused.len(), "{unique_query}");
let plans = observer.since(mark);
let frontend = plans
.iter()
.find(|plan| find_binary_join(plan).is_some())
.expect("frontend Join plan");
let join = find_binary_join(frontend).unwrap();
let uniqueness = join
.children()
.iter()
.map(|branch| uniqueness_operators_above_merge(branch))
.fold((false, false), |(window, assertion_filter), found| {
(window || found.0, assertion_filter || found.1)
});
// The whole binary is native: one fused node matches both operands above their own sources,
// and the duplicate-group check is the native matcher's, not a frontend WindowAgg/Filter pair.
assert!(
uniqueness.0,
"on(job) requires a frontend uniqueness WindowAggExec above MergeScan: {}",
displayable(frontend.as_ref()).indent(false)
capture.fused.iter().any(|node| node.children().len() > 1),
"`{unique_query}` must run as one native binary: {:?}",
capture.fused
);
assert!(
uniqueness.1,
"on(job) requires a frontend prom_assert_unique_match_group FilterExec above MergeScan: {}",
displayable(frontend.as_ref()).indent(false)
);
for branch in join.children() {
assert!(
contains_execution_node(branch, "MergeScanExec"),
"on(job) FE branch must retain remote boundary: {}",
displayable(frontend.as_ref()).indent(false)
);
assert!(
!contains_execution_node(branch, FUSED_EXEC_NAME),
"on(job) fused node must stay in region"
);
}
for fused in &capture.fused {
assert_fused_reads_scan_directly(fused, &unique_query);
assert_fused_reads_sources(fused, &unique_query);
}
let duplicate_query = format!("{GAUGE_TABLE} / on(job) {COUNTER_RHS_TABLE}");
@@ -941,7 +911,10 @@ async fn check_hybrid_binary_cases(instance: &Arc<Instance>, observer: &NativePl
"one-to-one match must drop unmatched series"
);
}
assert_eq!(2, capture.fused.len(), "{promql}: plans since {mark}");
assert!(
!capture.fused.is_empty(),
"{promql}: the pair must stay native: plans since {mark}"
);
let region_plans = observer
.since(mark)
.into_iter()
@@ -970,75 +943,37 @@ async fn check_hybrid_binary_cases(instance: &Arc<Instance>, observer: &NativePl
scan_tables.contains(&COUNTER_RHS_TABLE),
"{promql}: independent RHS scan missing: {scan_tables:?}"
);
for fused in &capture.fused {
assert_fused_reads_scan_directly(fused, &format!("hybrid `{promql}`"));
}
for fused in &capture.fused {
let per_fused = NativeCapture {
fused: vec![Arc::clone(fused)],
finalize: Vec::new(),
};
let counts = fused_root_metric_per_partition(
&per_fused,
ROOT_OUTPUT_VALUES_METRIC,
&format!("hybrid `{promql}`"),
);
assert!(
counts.iter().all(|count| count.is_some_and(|n| n > 0)),
"{promql}: each scan partition must emit values, got {counts:?}"
);
}
let plans = observer.since(mark);
let frontend = plans
// The binary routes by label set, so its output partitions are keyed by the match group and
// an input partition may contribute nothing to a given one. What must still hold is that
// each operand registered its values on every partition it reads - the same check the
// unpartitioned pair makes with the operand metrics.
let binary = capture
.fused
.iter()
.find(|plan| find_binary_join(plan).is_some())
.unwrap_or_else(|| panic!("{promql}: FE JoinExec missing; plans: {plans:#?}"));
let join = find_binary_join(frontend).unwrap();
assert_eq!(
2,
join.children().len(),
"{promql}: FE join branches: {}",
displayable(frontend.as_ref()).indent(false)
);
for branch in join.children() {
assert!(
contains_execution_node(branch, "MergeScanExec"),
"{promql}: FE Join branch lacks MergeScanExec: {}",
displayable(frontend.as_ref()).indent(false)
);
assert!(
!contains_execution_node(branch, FUSED_EXEC_NAME),
"{promql}: FusedSeriesExec leaked onto frontend: {}",
displayable(frontend.as_ref()).indent(false)
);
}
.find(|node| node.children().len() > 1)
.unwrap_or_else(|| panic!("{promql}: no native binary among {:?}", capture.fused));
let left =
fused_node_count_per_partition(binary, "native.root.left.output_values", &promql);
let right =
fused_node_count_per_partition(binary, "native.root.right.output_values", &promql);
assert!(
!is_nested_under_merge(frontend, "HashJoinExec"),
"{promql}: FE Join is under MergeScan: {}",
displayable(frontend.as_ref()).indent(false)
left.iter().chain(&right).all(Option::is_some),
"{promql}: operand metrics must be registered for every partition: \
left={left:?}, right={right:?}"
);
if index == 1 {
assert!(
has_projection_parent_of_join(frontend),
"06 outer scalar arithmetic must project above JoinExec: {}",
displayable(frontend.as_ref()).indent(false)
);
// The pair is matched by the native binary above its own sources; the frontend join and
// its uniqueness operators only existed while the binary could not be planned natively.
assert!(
capture.fused.iter().any(|node| node.children().len() > 1),
"{promql}: the pair must be one native binary: {:?}",
capture.fused
);
for fused in &capture.fused {
assert_fused_reads_sources(fused, &format!("hybrid `{promql}`"));
}
}
}
fn find_binary_join(plan: &Arc<dyn ExecutionPlan>) -> Option<&Arc<dyn ExecutionPlan>> {
if matches!(
plan.name(),
"HashJoinExec" | "NestedLoopJoinExec" | "CrossJoinExec"
) {
return Some(plan);
}
plan.children()
.iter()
.find_map(|child| find_binary_join(child))
}
fn find_execution_node<'a>(
plan: &'a Arc<dyn ExecutionPlan>,
name: &str,
@@ -1055,48 +990,6 @@ fn contains_execution_node(plan: &Arc<dyn ExecutionPlan>, name: &str) -> bool {
find_execution_node(plan, name).is_some()
}
/// Finds the PromQL uniqueness validation operators on the frontend side of a remote scan.
/// Traversal stops at each MergeScan so it cannot accidentally match region-side execution.
fn uniqueness_operators_above_merge(plan: &Arc<dyn ExecutionPlan>) -> (bool, bool) {
if plan.name() == "MergeScanExec" {
return (false, false);
}
let mut has_window = matches!(plan.name(), "WindowAggExec" | "BoundedWindowAggExec");
let mut has_assertion_filter = plan.name() == "FilterExec"
&& format!("{plan:?}").contains("prom_assert_unique_match_group");
for child in plan.children() {
let (child_window, child_filter) = uniqueness_operators_above_merge(child);
has_window |= child_window;
has_assertion_filter |= child_filter;
}
(has_window, has_assertion_filter)
}
fn is_nested_under_merge(plan: &Arc<dyn ExecutionPlan>, name: &str) -> bool {
fn walk(plan: &Arc<dyn ExecutionPlan>, under_merge: bool, name: &str) -> bool {
if under_merge && plan.name() == name {
return true;
}
plan.children()
.iter()
.any(|child| walk(child, under_merge || plan.name() == "MergeScanExec", name))
}
walk(plan, false, name)
}
fn has_projection_parent_of_join(plan: &Arc<dyn ExecutionPlan>) -> bool {
plan.name() == "ProjectionExec"
&& plan
.children()
.iter()
.any(|child| find_binary_join(child).is_some())
|| plan
.children()
.iter()
.any(|child| has_projection_parent_of_join(child))
}
/// Executes one statement and drops its result.
async fn execute_sql(instance: &Arc<Instance>, sql: &str) {
for result in instance.do_query(sql, QueryContext::arc()).await {
@@ -1514,6 +1407,31 @@ fn assert_fused_reads_scan_directly(fused: &Arc<dyn ExecutionPlan>, context: &st
);
}
/// The fused node keeps its scan width and reads every operand through its own source: a single
/// operand reads one scan, and a native binary takes one input per operand, each a raw source or a
/// merge scan over the regions.
fn assert_fused_reads_sources(fused: &Arc<dyn ExecutionPlan>, context: &str) {
let plan = displayable(fused.as_ref()).indent(false).to_string();
assert_eq!(
SCAN_PARTITIONS,
fused.properties().output_partitioning().partition_count(),
"{context}: the fused producer must keep the {SCAN_PARTITIONS} scan partitions:\n{plan}"
);
let children = fused.children();
assert!(
!children.is_empty(),
"{context}: the fused producer must read at least one input:\n{plan}"
);
for child in children {
assert!(
contains_execution_node(child, "SeriesScanExec")
|| contains_execution_node(child, "MergeScanExec")
|| contains_execution_node(child, "RawSeriesSourceExec"),
"{context}: every operand must read its own source:\n{plan}"
);
}
}
/// The node a transparent cooperative-scheduling wrapper stands in for, or `node` itself.
///
/// The decorator clones its input's properties, changes only the scheduling type and forwards
tests-integration/src/tests/promql_test.rs
+84 -45
View File
@@ -1868,6 +1868,21 @@ fn observed_physical_nodes(
nodes
}
/// The region reads a native node answers from, descending through cooperative wrappers and the
/// segments a statement cut out of itself.
fn native_region_inputs(
plan: &Arc<dyn datafusion::physical_plan::ExecutionPlan>,
out: &mut Vec<Arc<dyn datafusion::physical_plan::ExecutionPlan>>,
) {
for child in plan.children() {
if child.is::<MergeScanExec>() {
out.push(Arc::clone(child));
} else {
native_region_inputs(child, out);
}
}
}
fn observed_series_scans(observer: &NativePlanObserver) -> Vec<Arc<dyn ExecutionPlan>> {
observed_physical_nodes(observer)
.into_iter()
@@ -2880,11 +2895,11 @@ fn repeated_selector_fixture_inserts(prefix: &str) -> String {
}
/// Real two-datanode TCP acceptance for a repeated selector over ONE partitioned metric table.
/// Over the wire every prepared scan becomes its own raw source with its own time bounds, so two
/// occurrences of the same fragmented table - a ratio of two windows, or a binary expression over
/// two identical ones - stay independent instead of falling back. In process the same expression
/// would place one node region-locally over both scans of the one table, which no region can serve,
/// so this admission is remote-only.
/// Every prepared scan becomes its own raw source with its own time bounds, so two occurrences of
/// the same fragmented table - a ratio of two windows, or a binary expression over two identical
/// ones - stay independent instead of falling back. Over the wire the transport carries those
/// sources; in process they keep the node out of the region, which cannot serve one node over both
/// scans of the one table.
#[tokio::test(flavor = "multi_thread")]
async fn distributed_native_promql_repeated_selector_over_partitioned_table() {
common_telemetry::init_default_ut_logging();
@@ -6587,57 +6602,45 @@ async fn native_promql_info_enriches_from_the_info_table() {
}
}
/// A two-selector query over a table split into regions never fuses the binary itself.
/// A window over a table partitioned on a column that is not a tag sees the whole series.
///
/// A fused node with more than one input has no placement on such a table — the dist analyzer
/// refuses it with "only accept plan with at most one child" — so admitting one used to make the
/// whole query fail instead of falling back. Each operand may still be fused on its own: that is a
/// single-input node over a region-local scan, which places fine, and it is what the hybrid
/// operand route does. What must never appear is a fused node with two children, so that is what
/// is asserted rather than "no fused node at all".
/// A window over a table partitioned on its time index sees the whole series.
///
/// One selector is enough: such a partition splits a series across regions, so a per-region window
/// would evaluate over a fragment and drop every range that straddles the boundary. The selector
/// therefore reads the table through a raw source and the window runs above the reconciliation.
/// The fixture puts one series' two samples on either side of the boundary.
/// Such a table splits one series across regions, so a per-region prefix evaluates `rate` over a
/// fragment and loses every window that straddles the boundary. The cut therefore has to keep each
/// occurrence's raw source and run the prefix above the reconciliation, exactly as the remote route
/// does. The fixture puts the two samples of one series on either side of the boundary and compares
/// the native answer with the existing plan's.
#[tokio::test(flavor = "multi_thread")]
async fn native_promql_time_partitioned_selector_reads_whole_series() {
async fn native_promql_window_over_a_time_partitioned_table_keeps_whole_series() {
let (standalone, observer) = native_test_instance(GreptimeDbStandaloneBuilder::new(
"native_promql_time_partitioned_selector",
"native_promql_time_partitioned_window",
))
.await;
let instance = standalone.fe_instance().clone();
execute_all(
&instance,
"CREATE TABLE native_time_split (host STRING, ts TIMESTAMP TIME INDEX, val DOUBLE, \
"CREATE TABLE native_time_partitioned (host STRING, ts TIMESTAMP TIME INDEX, val DOUBLE, \
PRIMARY KEY(host)) PARTITION ON COLUMNS (ts) (ts < '1970-01-01 00:02:30', \
ts >= '1970-01-01 00:02:30')",
QueryContext::arc(),
)
.await;
// One series, one sample in each region: the second is 240s after the first, and the boundary
// sits between them, so neither region holds a complete window on its own.
execute_all(
&instance,
"INSERT INTO native_time_split VALUES ('h0', 60000, 60), ('h0', 240000, 240)",
"INSERT INTO native_time_partitioned VALUES ('h0', 60000, 60), ('h0', 240000, 240)",
QueryContext::arc(),
)
.await;
for query in [
"rate(native_time_split[5m])",
"sum(rate(native_time_split[5m]))",
"rate(native_time_partitioned[5m]) + rate(native_time_partitioned[5m])",
"(rate(native_time_partitioned[5m]) + rate(native_time_partitioned[5m])) \
+ rate(native_time_partitioned[5m])",
] {
let sql = format!("TQL EVAL (300, 300, '1m') {query}");
observer.clear();
let (_, native) =
native_labelled_rows(native_tql_output(&instance, &sql, true).await).await;
let plan = observer.root_plan();
assert!(
find_plan_node(&plan, "FusedSeriesExec").is_some(),
"{query}: the frontend tree must run natively, not on the existing plan: {:?}",
display_plan_tree(&plan)
);
let (_, legacy) =
native_labelled_rows(native_tql_output(&instance, &sql, false).await).await;
let native = native_result_rows(native_tql_output(&instance, &sql, true).await).await;
let legacy = native_result_rows(native_tql_output(&instance, &sql, false).await).await;
assert!(
!native.is_empty(),
"{query}: the native plan answered nothing"
@@ -6646,11 +6649,33 @@ async fn native_promql_time_partitioned_selector_reads_whole_series() {
native, legacy,
"{query}: the native plan must match the existing one"
);
let root = observer.root_plan();
let mut inputs = Vec::new();
native_region_inputs(&root, &mut inputs);
assert!(
!inputs.is_empty(),
"{query}: the window runs above the region reads: {:?}",
display_plan_tree(&root)
);
// A fallback plan also produces the existing plan's rows, so the admission has to be
// asserted separately: the frontend tree must carry a fused node.
assert!(
find_plan_node(&root, "FusedSeriesExec").is_some(),
"{query}: the tree must run natively, not on the existing plan: {:?}",
display_plan_tree(&root)
);
}
}
/// A query over two selectors of one partitioned table matches above the regions.
///
/// A fused node over the bare scans of such a table has no placement in a region — the dist
/// analyzer refuses it with "only accept plan with at most one child" — so each selector
/// occurrence reads that table through its own raw region input instead. Every input is one read of
/// the whole table, and the matching the query asks for runs once above them, over every region's
/// contribution: two inputs for a two-selector query, three for a three-selector one.
#[tokio::test(flavor = "multi_thread")]
async fn native_promql_two_selectors_on_a_partitioned_table_keep_the_binary_unfused() {
async fn native_promql_two_selectors_on_a_partitioned_table_match_above_independent_regions() {
let (standalone, observer) = native_test_instance(GreptimeDbStandaloneBuilder::new(
"native_promql_partitioned_pair",
))
@@ -6670,10 +6695,11 @@ async fn native_promql_two_selectors_on_a_partitioned_table_keep_the_binary_unfu
QueryContext::arc(),
)
.await;
for (query, expected) in [
for (query, expected, sources) in [
(
"native_partitioned_pair{kind=\"a\"} and on(host) native_partitioned_pair{kind=\"b\",host=\"h0\"}",
vec![(5_000, "h0".to_string(), 1.0)],
2,
),
(
"native_partitioned_pair{kind=\"a\"} - on(host) native_partitioned_pair{kind=\"b\"}",
@@ -6681,8 +6707,10 @@ async fn native_promql_two_selectors_on_a_partitioned_table_keep_the_binary_unfu
(5_000, "h0".to_string(), -1.0),
(5_000, "h2".to_string(), -1.0),
],
2,
),
// Three selectors: the inner pair may take the hybrid route under the outer join.
// Three selectors: the inner pair matches above its own two region inputs, under the outer
// match.
(
"native_partitioned_pair{kind=\"a\"} - on(host) native_partitioned_pair{kind=\"b\"} \
+ on(host) native_partitioned_pair{kind=\"a\"}",
@@ -6690,19 +6718,30 @@ async fn native_promql_two_selectors_on_a_partitioned_table_keep_the_binary_unfu
(5_000, "h0".to_string(), 0.0),
(5_000, "h2".to_string(), 2.0),
],
3,
),
] {
let sql = format!("TQL EVAL (5, 5, '1s') {query}");
observer.clear();
let native = native_result_rows(native_tql_output(&instance, &sql, true).await).await;
if let Some(fused) = observer.fused() {
assert_eq!(
fused.children().len(),
1,
"`{query}` fused the two-selector binary, which has no placement on a partitioned \
table: {fused:?}"
);
}
// The frontend's own tree: the fused node above the regions, not the region-side plan the
// observer also wraps.
let root = observer.root_plan();
let mut inputs = Vec::new();
native_region_inputs(&root, &mut inputs);
assert_eq!(
sources,
inputs.len(),
"`{query}` reads one region input per selector occurrence: {:?}",
display_plan_tree(&root)
);
// The values alone cannot tell native matching from a legacy frontend join that reads the
// same regions, so require the native binary in the frontend tree.
assert!(
find_plan_node(&root, "FusedSeriesExec").is_some(),
"`{query}` must match natively above its region inputs: {:?}",
display_plan_tree(&root)
);
assert_eq!(native, expected, "{query}");
let legacy = native_result_rows(native_tql_output(&instance, &sql, false).await).await;
assert_eq!(native, legacy, "{query}");