From b9aa627180ac696b83de2e675a8ceb0910295216 Mon Sep 17 00:00:00 2001 From: dennis zhuang Date: Tue, 20 Oct 2026 08:03:35 +0000 Subject: [PATCH] fix(query): forward scalar subquery filters to region scans (#9738) * fix(query): forward scalar subquery filters to region scans Since DataFusion 55, uncorrelated scalar subqueries run in ScalarSubqueryExec instead of being rewritten into joins, so filters such as ts = (SELECT ts ... ORDER BY ts DESC LIMIT 1) no longer reach region scans through a join dynamic filter. MergeScanExec now captures such filters as remote dynamic filters, keeps them in the parent FilterExec, and replaces the subquery with its result before querying regions. Signed-off-by: Dennis Zhuang * fix(query): check the remote dyn filter budget across pushdown phases Filters captured in the Pre and Post phases are sent in one request, but each phase validated the initial registration budget alone. When their sum exceeded the budget, serialization dropped every registration, including join and TopK filters. Validate the combined set and keep filters accepted by the other phase. Signed-off-by: Dennis Zhuang * fix(query): keep IN-bearing scalar subquery filters at the parent Oversized remote dyn filter payloads replace IN lists with true. That only widens a conjunction of bounds, but inside an arbitrary predicate such as (host IN (...)) = (SELECT flag ...) it can drop rows the parent filter would keep. Do not forward scalar subquery filters that contain IN lists. Signed-off-by: Dennis Zhuang * chore(query): note the IN fallback restriction for scalar subquery filters Signed-off-by: Dennis Zhuang * chore(query): document the scalar subquery dyn filter bridge removal condition Signed-off-by: Dennis Zhuang * test(query): keep only dyn_filters of region scans in the scalar subquery plan case #9565 added scan diagnostics to EXPLAIN ANALYZE VERBOSE and partition filter rendering changed; the case only checks the forwarded dyn filter. Signed-off-by: Dennis Zhuang --------- Signed-off-by: Dennis Zhuang --- src/query/AGENTS.md | 3 + src/query/src/dist_plan/dyn_filter_bridge.rs | 375 ++++++++++++++++-- src/query/src/dist_plan/merge_scan.rs | 59 ++- .../time_index_filter_pushdown.result | 82 ++++ .../optimizer/time_index_filter_pushdown.sql | 33 ++ .../time_index_filter_pushdown.result | 82 ++++ .../optimizer/time_index_filter_pushdown.sql | 33 ++ 7 files changed, 619 insertions(+), 48 deletions(-) diff --git a/src/query/AGENTS.md b/src/query/AGENTS.md index 8dc03a22877..fe794dcfc73 100644 --- a/src/query/AGENTS.md +++ b/src/query/AGENTS.md @@ -24,6 +24,9 @@ side effects live in `operator`. - Optimizer or `MergeScan` changes usually touch rule ordering, physical planning, serialization, and distributed-plan tests together. - User-visible SQL/PromQL/log behavior needs planner coverage and a sqlness case. +- `MergeScanExec` sends remote dyn filters captured in every filter pushdown + phase in one request; the initial registration budget applies to all of them + together (`dist_plan/dyn_filter_bridge.rs`). - `RegionQueryHandler` changes require the frontend implementation to move with it. - DataFusion dependency changes follow the pinned-fork rule in the repo-wide diff --git a/src/query/src/dist_plan/dyn_filter_bridge.rs b/src/query/src/dist_plan/dyn_filter_bridge.rs index 4761c7aea93..fb49839b304 100644 --- a/src/query/src/dist_plan/dyn_filter_bridge.rs +++ b/src/query/src/dist_plan/dyn_filter_bridge.rs @@ -19,9 +19,16 @@ use common_query::request::{ DynFilterPayload, INITIAL_REMOTE_DYN_FILTER_REGISTRATIONS_EXTENSION_KEY, InitialDynFilterReg, InitialDynFilterRegs, InitialDynFilterSnapshot, REMOTE_DYN_FILTER_PAYLOAD_MAX_BYTES, }; +use datafusion::physical_plan::filter_pushdown::PushedDown; use datafusion_common::Result; +use datafusion_common::tree_node::{Transformed, TreeNode, TreeNodeRecursion}; +use datafusion_expr::ColumnarValue; use datafusion_physical_expr::PhysicalExpr; -use datafusion_physical_expr::expressions::DynamicFilterPhysicalExpr; +use datafusion_physical_expr::expressions::{ + Column, DynamicFilterPhysicalExpr, InListExpr, Literal, lit, +}; +use datafusion_physical_expr::scalar_subquery::ScalarSubqueryExpr; +use datafusion_physical_expr::utils::collect_columns; use session::context::{QueryContext, QueryContextRef}; use store_api::storage::RegionId; @@ -36,49 +43,82 @@ pub(crate) struct CapturedDynFilter { filter_id: FilterId, initial_registration: InitialDynFilterReg, pub(crate) alive_dyn_filter: Arc, + /// A parent filter that references uncorrelated scalar subqueries. Its value is only + /// known once `ScalarSubqueryExec` has run the subqueries, so `alive_dyn_filter` + /// starts as `true` and is replaced by the resolved filter before regions are queried. + // TODO(discord9): temporary bridge. DataFusion fills the scalar subquery results in + // `ScalarSubqueryExec` but does not expose predicates using them as + // `DynamicFilterPhysicalExpr`, so we keep the predicate and bind it ourselves. Once + // upstream does and the pinned DataFusion includes it, remove this field and the + // scalar subquery capture/resolve path in favor of the regular dyn filter capture. + // Keep the per-phase bookkeeping in `MergeScanExec`. + scalar_subquery_filter: Option>, } #[derive(Debug, Clone)] pub(crate) struct RemoteDynFilterPushdown { pub(crate) captured_dyn_filters: Vec, /// Preflight result per parent filter. - pub(crate) pushed_down: Vec, + pub(crate) pushed_down: Vec, } +/// Captures `parent_filters` as remote dyn filters. `accepted` holds filters captured by +/// another pushdown phase; they are sent in the same request, so the initial registration +/// budget is checked against both and they are kept if the new filters do not fit. pub(crate) fn capture_remote_dyn_filters_for_pushdown( remote_dyn_filter_producer_id: RemoteDynFilterProducerId, + first_producer_local_ordinal: usize, + accepted: &[CapturedDynFilter], parent_filters: Vec>, ) -> RemoteDynFilterPushdown { let mut pushed_down = Vec::with_capacity(parent_filters.len()); let mut captured_dyn_filters = Vec::new(); - for (producer_local_ordinal, filter) in parent_filters.into_iter().enumerate() { - let Some(alive_dyn_filter) = downcast_dynamic_filter(filter) else { - pushed_down.push(false); + for (index, filter) in parent_filters.into_iter().enumerate() { + let producer_local_ordinal = first_producer_local_ordinal + index; + let captured = if let Some(alive_dyn_filter) = downcast_dynamic_filter(filter.clone()) { + build_captured_dyn_filter( + remote_dyn_filter_producer_id, + producer_local_ordinal, + alive_dyn_filter, + ) + .map(|captured| (captured, PushedDown::Yes)) + } else if is_scalar_subquery_filter(&filter) { + // The filter is only forwarded to regions for pruning; the parent keeps + // evaluating it, so report it as not pushed down. + build_scalar_subquery_dyn_filter( + remote_dyn_filter_producer_id, + producer_local_ordinal, + filter, + ) + .map(|captured| (captured, PushedDown::No)) + } else { + pushed_down.push(PushedDown::No); continue; }; - match build_captured_dyn_filter( - remote_dyn_filter_producer_id, - producer_local_ordinal, - alive_dyn_filter, - ) { - Ok(captured_dyn_filter) => { - pushed_down.push(true); + match captured { + Ok((captured_dyn_filter, pushdown)) => { + pushed_down.push(pushdown); captured_dyn_filters.push(captured_dyn_filter); } Err(error) => { common_telemetry::warn!(error; "Remote dyn filter is not pushed down because initial registration cannot be built"); - pushed_down.push(false); + pushed_down.push(PushedDown::No); } } } - if let Err(error) = validate_initial_registrations_for_pushdown(&captured_dyn_filters) { + let combined = accepted + .iter() + .chain(&captured_dyn_filters) + .cloned() + .collect::>(); + if let Err(error) = validate_initial_registrations_for_pushdown(&combined) { common_telemetry::warn!(error; "Remote dyn filters are not pushed down because initial registrations are invalid"); return RemoteDynFilterPushdown { captured_dyn_filters: Vec::new(), - pushed_down: vec![false; pushed_down.len()], + pushed_down: vec![PushedDown::No; pushed_down.len()], }; } @@ -88,6 +128,105 @@ pub(crate) fn capture_remote_dyn_filters_for_pushdown( } } +/// Returns true if `filter` references at least one scalar subquery and is otherwise +/// built only from columns and literals, so it becomes a plain predicate once the +/// subqueries are resolved. +/// +/// Filters containing `IN` lists are rejected: an oversized payload replaces `IN` with +/// `true`, which only widens a conjunction but can narrow an arbitrary predicate such as +/// `(host IN (...)) = false`. +// TODO(dennis): make the bounds-only fallback in `DynFilterPayload` replace only `IN` lists +// in conjunctive positions, then drop this restriction. +fn is_scalar_subquery_filter(filter: &Arc) -> bool { + let mut has_scalar_subquery = false; + let mut supported = true; + let _ = filter.apply(|expr| { + if expr.is::() { + has_scalar_subquery = true; + } else if expr.is::() + || (expr.children().is_empty() && !expr.is::() && !expr.is::()) + { + supported = false; + return Ok(TreeNodeRecursion::Stop); + } + Ok(TreeNodeRecursion::Continue) + }); + has_scalar_subquery && supported +} + +fn build_scalar_subquery_dyn_filter( + remote_dyn_filter_producer_id: RemoteDynFilterProducerId, + producer_local_ordinal: usize, + filter: Arc, +) -> Result { + let mut columns = collect_columns(&filter).into_iter().collect::>(); + columns.sort_by_key(|column| column.index()); + let children = columns + .into_iter() + .map(|column| Arc::new(column) as Arc) + .collect(); + let alive_dyn_filter = Arc::new(DynamicFilterPhysicalExpr::new(children, lit(true))); + let mut captured = build_captured_dyn_filter( + remote_dyn_filter_producer_id, + producer_local_ordinal, + alive_dyn_filter, + )?; + captured.scalar_subquery_filter = Some(filter); + Ok(captured) +} + +/// Replaces the scalar subqueries in captured filters with their results. +/// +/// `ScalarSubqueryExec` runs its subqueries before executing its input, so the results +/// are available when `MergeScanExec` is executed. This runs on every execution because +/// `ScalarSubqueryExec::reset_state` clears the results for re-execution. A filter whose +/// subqueries are unresolved falls back to `true`, which only disables pruning. +pub(crate) fn resolve_scalar_subquery_dyn_filters(captured_dyn_filters: &[CapturedDynFilter]) { + for captured in captured_dyn_filters { + let Some(filter) = &captured.scalar_subquery_filter else { + continue; + }; + let resolved = replace_scalar_subqueries(filter).unwrap_or_else(|error| { + common_telemetry::debug!( + "Scalar subquery filter {} is not forwarded to regions: {}", + filter, + error + ); + lit(true) + }); + if captured + .alive_dyn_filter + .current() + .is_ok_and(|current| current.as_ref() == resolved.as_ref()) + { + continue; + } + if let Err(error) = captured.alive_dyn_filter.update(resolved) { + common_telemetry::warn!(error; "Failed to update scalar subquery remote dyn filter"); + } + } +} + +fn replace_scalar_subqueries(filter: &Arc) -> Result> { + let empty_batch = + arrow::record_batch::RecordBatch::new_empty(Arc::new(arrow_schema::Schema::empty())); + filter + .clone() + .transform_up(|expr| { + if !expr.is::() { + return Ok(Transformed::no(expr)); + } + // Fails if the subquery has not been executed yet. + match expr.evaluate(&empty_batch)? { + ColumnarValue::Scalar(value) => Ok(Transformed::yes(lit(value))), + ColumnarValue::Array(_) => Err(datafusion_common::DataFusionError::Internal( + "scalar subquery evaluated to an array".to_string(), + )), + } + }) + .map(|transformed| transformed.data) +} + fn downcast_dynamic_filter( expr: Arc, ) -> Option> { @@ -159,6 +298,7 @@ fn build_captured_dyn_filter( filter_id, initial_registration: attach_initial_snapshot(initial_registration, &alive_dyn_filter), alive_dyn_filter, + scalar_subquery_filter: None, }) } @@ -315,7 +455,9 @@ mod tests { use datafusion::execution::TaskContext; use datafusion_common::ScalarValue; use datafusion_expr::ColumnarValue; - use datafusion_physical_expr::expressions::{Column, lit}; + use datafusion_expr::Operator; + use datafusion_expr::physical_planning_context::{ScalarSubqueryResults, SubqueryIndex}; + use datafusion_physical_expr::expressions::BinaryExpr; use session::query_id::QueryId; use uuid::Uuid; @@ -461,6 +603,133 @@ mod tests { .to_string() } + fn pushed_down(pushdown: &RemoteDynFilterPushdown) -> Vec { + pushdown + .pushed_down + .iter() + .map(|pushed_down| matches!(pushed_down, PushedDown::Yes)) + .collect() + } + + fn host_eq_scalar_subquery(results: &ScalarSubqueryResults) -> Arc { + Arc::new(BinaryExpr::new( + Arc::new(Column::new("host", 0)), + Operator::Eq, + Arc::new(ScalarSubqueryExpr::new( + arrow_schema::DataType::Utf8, + true, + SubqueryIndex::new(0), + results.clone(), + )), + )) + } + + #[test] + fn scalar_subquery_filter_is_captured_but_left_to_parent() { + let results = ScalarSubqueryResults::new(1); + let parent_filters = vec![ + host_eq_scalar_subquery(&results), + Arc::new(BinaryExpr::new( + Arc::new(Column::new("host", 0)), + Operator::Eq, + lit("a"), + )) as Arc, + ]; + + let pushdown = capture_remote_dyn_filters_for_pushdown( + test_remote_dyn_filter_producer_id(42), + 0, + &[], + parent_filters, + ); + + assert_eq!(pushed_down(&pushdown), vec![false, false]); + assert_eq!(pushdown.captured_dyn_filters.len(), 1); + assert_eq!( + pushdown.captured_dyn_filters[0] + .filter_id + .producer_ordinal(), + 0 + ); + } + + #[test] + fn scalar_subquery_filter_with_in_list_is_left_to_parent() { + let results = ScalarSubqueryResults::new(1); + let schema = arrow_schema::Schema::new(vec![arrow_schema::Field::new( + "host", + arrow_schema::DataType::Utf8, + true, + )]); + let host_in_list = datafusion_physical_expr::expressions::in_list( + Arc::new(Column::new("host", 0)), + vec![lit("a"), lit("b")], + &false, + &schema, + ) + .unwrap(); + // `(host IN ('a', 'b')) = (SELECT flag ...)` + let filter = Arc::new(BinaryExpr::new( + host_in_list, + Operator::Eq, + Arc::new(ScalarSubqueryExpr::new( + arrow_schema::DataType::Boolean, + true, + SubqueryIndex::new(0), + results, + )), + )) as Arc; + + let pushdown = capture_remote_dyn_filters_for_pushdown( + test_remote_dyn_filter_producer_id(42), + 0, + &[], + vec![filter], + ); + + assert_eq!(pushed_down(&pushdown), vec![false]); + assert!(pushdown.captured_dyn_filters.is_empty()); + } + + #[test] + fn scalar_subquery_dyn_filter_follows_subquery_results() { + let results = ScalarSubqueryResults::new(1); + let captured = capture_remote_dyn_filters_for_pushdown( + test_remote_dyn_filter_producer_id(42), + 0, + &[], + vec![host_eq_scalar_subquery(&results)], + ) + .captured_dyn_filters; + let snapshot = |captured: &[CapturedDynFilter]| { + decode_snapshot( + refreshed_regs(captured).regs[0] + .initial_snapshot + .as_ref() + .unwrap(), + "host", + ) + }; + + // Not executed yet: regions must not be filtered. + resolve_scalar_subquery_dyn_filters(&captured); + assert_eq!(snapshot(&captured), "true"); + + results + .set( + SubqueryIndex::new(0), + ScalarValue::Utf8(Some("a".to_string())), + ) + .unwrap(); + resolve_scalar_subquery_dyn_filters(&captured); + assert_eq!(snapshot(&captured), "host@0 = a"); + + // `ScalarSubqueryExec::reset_state` clears the results before re-execution. + results.clear(); + resolve_scalar_subquery_dyn_filters(&captured); + assert_eq!(snapshot(&captured), "true"); + } + #[test] fn capture_remote_dyn_filters_for_pushdown_preserves_parent_filter_ordinals() { let parent_filters = vec![ @@ -477,9 +746,13 @@ mod tests { ]; let remote_dyn_filter_producer_id = test_remote_dyn_filter_producer_id(42); - let captured = - capture_remote_dyn_filters_for_pushdown(remote_dyn_filter_producer_id, parent_filters) - .captured_dyn_filters; + let captured = capture_remote_dyn_filters_for_pushdown( + remote_dyn_filter_producer_id, + 0, + &[], + parent_filters, + ) + .captured_dyn_filters; assert_eq!(captured.len(), 2); assert_eq!( @@ -506,10 +779,14 @@ mod tests { ]; let remote_dyn_filter_producer_id = test_remote_dyn_filter_producer_id(42); - let pushdown = - capture_remote_dyn_filters_for_pushdown(remote_dyn_filter_producer_id, parent_filters); + let pushdown = capture_remote_dyn_filters_for_pushdown( + remote_dyn_filter_producer_id, + 0, + &[], + parent_filters, + ); - assert_eq!(pushdown.pushed_down, vec![false, true, false]); + assert_eq!(pushed_down(&pushdown), vec![false, true, false]); assert_eq!(pushdown.captured_dyn_filters.len(), 1); assert_eq!( pushdown.captured_dyn_filters[0] @@ -541,10 +818,12 @@ mod tests { let pushdown = capture_remote_dyn_filters_for_pushdown( test_remote_dyn_filter_producer_id(42), + 0, + &[], parent_filters, ); - assert_eq!(pushdown.pushed_down, vec![false]); + assert_eq!(pushed_down(&pushdown), vec![false]); assert!(pushdown.captured_dyn_filters.is_empty()); } @@ -558,10 +837,12 @@ mod tests { let pushdown = capture_remote_dyn_filters_for_pushdown( test_remote_dyn_filter_producer_id(42), + 0, + &[], parent_filters, ); - assert_eq!(pushdown.pushed_down, vec![true]); + assert_eq!(pushed_down(&pushdown), vec![true]); assert!( pushdown.captured_dyn_filters[0] .initial_registration @@ -581,10 +862,12 @@ mod tests { let pushdown = capture_remote_dyn_filters_for_pushdown( test_remote_dyn_filter_producer_id(42), + 0, + &[], parent_filters, ); - assert_eq!(pushdown.pushed_down, vec![true]); + assert_eq!(pushed_down(&pushdown), vec![true]); let snapshot = pushdown.captured_dyn_filters[0] .initial_registration .initial_snapshot @@ -610,10 +893,42 @@ mod tests { let pushdown = capture_remote_dyn_filters_for_pushdown( test_remote_dyn_filter_producer_id(42), + 0, + &[], parent_filters, ); - assert_eq!(pushdown.pushed_down, vec![false, false]); + assert_eq!(pushed_down(&pushdown), vec![false, false]); + assert!(pushdown.captured_dyn_filters.is_empty()); + } + + #[test] + fn capture_remote_dyn_filters_for_pushdown_keeps_accepted_filters_within_budget() { + const MAX_INITIAL_REGS: usize = 64; + + let accepted = capture_remote_dyn_filters_for_pushdown( + test_remote_dyn_filter_producer_id(42), + 0, + &[], + (0..MAX_INITIAL_REGS) + .map(|ordinal| { + test_dyn_filter_with_snapshot_payload(&format!("host_{ordinal}"), ordinal, 1) + as Arc + }) + .collect(), + ) + .captured_dyn_filters; + assert_eq!(accepted.len(), MAX_INITIAL_REGS); + + let results = ScalarSubqueryResults::new(1); + let pushdown = capture_remote_dyn_filters_for_pushdown( + test_remote_dyn_filter_producer_id(42), + 1 << 31, + &accepted, + vec![host_eq_scalar_subquery(&results)], + ); + + assert_eq!(pushed_down(&pushdown), vec![false]); assert!(pushdown.captured_dyn_filters.is_empty()); } @@ -630,11 +945,13 @@ mod tests { let pushdown = capture_remote_dyn_filters_for_pushdown( test_remote_dyn_filter_producer_id(42), + 0, + &[], parent_filters, ); assert!(pushdown.captured_dyn_filters.is_empty()); - assert_eq!(pushdown.pushed_down, vec![false; TOO_MANY_INITIAL_REGS]); + assert_eq!(pushed_down(&pushdown), vec![false; TOO_MANY_INITIAL_REGS]); } #[test] @@ -652,11 +969,13 @@ mod tests { let pushdown = capture_remote_dyn_filters_for_pushdown( test_remote_dyn_filter_producer_id(42), + 0, + &[], parent_filters, ); assert!(pushdown.captured_dyn_filters.is_empty()); - assert_eq!(pushdown.pushed_down, vec![false; TOO_MANY_INITIAL_REGS]); + assert_eq!(pushed_down(&pushdown), vec![false; TOO_MANY_INITIAL_REGS]); } #[test] diff --git a/src/query/src/dist_plan/merge_scan.rs b/src/query/src/dist_plan/merge_scan.rs index 9717c436e0b..90a57bf1a5c 100644 --- a/src/query/src/dist_plan/merge_scan.rs +++ b/src/query/src/dist_plan/merge_scan.rs @@ -68,6 +68,7 @@ use crate::dist_plan::dyn_filter_bridge::{ CapturedDynFilter, capture_remote_dyn_filters_for_pushdown, query_context_with_refreshed_initial_dyn_filter_regs, register_dyn_filter_subscribers_for_region, register_remote_dyn_filters, + resolve_scalar_subquery_dyn_filters, }; use crate::dist_plan::{ FilterId, RemoteDynFilterProducerId, RemoteDynFilterRegistryLease, Subscriber, @@ -126,6 +127,22 @@ fn remote_plan_row_bound(plan: &LogicalPlan) -> Option { } } +fn pushdown_phase_slot(phase: FilterPushdownPhase) -> usize { + match phase { + FilterPushdownPhase::Pre => 0, + FilterPushdownPhase::Post => 1, + } +} + +/// Producer-local ordinals must be unique across phases because they are part of the +/// remote filter id; pre-phase filters use the upper half of the `u32` ordinal space. +fn first_producer_local_ordinal(phase: FilterPushdownPhase) -> usize { + match phase { + FilterPushdownPhase::Pre => 1 << 31, + FilterPushdownPhase::Post => 0, + } +} + fn remote_dyn_filter_enabled(query_ctx: &QueryContextRef) -> Result { remote_dyn_filter_pushdown_enabled_from_extensions(&query_ctx.extensions()) .map_err(|err| DataFusionError::External(Box::new(err))) @@ -483,7 +500,10 @@ pub struct MergeScanExec { query_ctx: QueryContextRef, /// Optional because RDF must fail open: missing ids skip RDF but keep normal query execution. remote_dyn_filter_producer_id: Option, - captured_remote_dyn_filters: Arc>>, + /// Filters captured by each filter pushdown phase, indexed by [`pushdown_phase_slot`]. + /// The pre-optimization phase carries `FilterExec` predicates (e.g. scalar subquery + /// filters) and the post-optimization phase carries dynamic filters from joins and TopK. + captured_remote_dyn_filters: Arc; 2]>>, target_partition: usize, enable_per_region_metrics: bool, } @@ -616,6 +636,7 @@ impl MergeScanExec { } else { Vec::new() }; + resolve_scalar_subquery_dyn_filters(&captured_remote_dyn_filters); let dbname = context.task_id().unwrap_or_default(); let tracing_context = TracingContext::from_json(context.session_id().as_str()); let current_channel = self.query_ctx.channel(); @@ -890,7 +911,7 @@ impl MergeScanExec { } fn captured_remote_dyn_filters(&self) -> Vec { - self.captured_remote_dyn_filters.lock().unwrap().clone() + self.captured_remote_dyn_filters.lock().unwrap().concat() } pub fn sub_stage_metrics(&self) -> Vec { @@ -1092,7 +1113,7 @@ impl ExecutionPlan for MergeScanExec { fn handle_child_pushdown_result( &self, - _phase: FilterPushdownPhase, + phase: FilterPushdownPhase, child_pushdown_result: ChildPushdownResult, _config: &datafusion::config::ConfigOptions, ) -> Result>> { @@ -1106,7 +1127,7 @@ impl ExecutionPlan for MergeScanExec { // Reject remote pushdown instead of pretending success: this keeps // DataFusion/local dynamic filter semantics intact while disabling // only FE -> DN remote dynamic filter propagation. - self.captured_remote_dyn_filters.lock().unwrap().clear(); + self.captured_remote_dyn_filters.lock().unwrap()[pushdown_phase_slot(phase)].clear(); let new_self = Arc::new(self.clone()); return Ok(FilterPushdownPropagation { @@ -1120,7 +1141,7 @@ impl ExecutionPlan for MergeScanExec { common_telemetry::warn!( "MergeScan remote dynamic filter producer id is not assigned; skipping remote dynamic filter pushdown" ); - self.captured_remote_dyn_filters.lock().unwrap().clear(); + self.captured_remote_dyn_filters.lock().unwrap()[pushdown_phase_slot(phase)].clear(); let new_self = Arc::new(self.clone()); return Ok(FilterPushdownPropagation { @@ -1128,24 +1149,20 @@ impl ExecutionPlan for MergeScanExec { updated_node: Some(new_self), }); }; - let remote_dyn_filter_pushdown = - capture_remote_dyn_filters_for_pushdown(remote_dyn_filter_producer_id, parent_filters); - *self.captured_remote_dyn_filters.lock().unwrap() = - remote_dyn_filter_pushdown.captured_dyn_filters; + let slot = pushdown_phase_slot(phase); + let mut captured_remote_dyn_filters = self.captured_remote_dyn_filters.lock().unwrap(); + let remote_dyn_filter_pushdown = capture_remote_dyn_filters_for_pushdown( + remote_dyn_filter_producer_id, + first_producer_local_ordinal(phase), + &captured_remote_dyn_filters[1 - slot], + parent_filters, + ); + captured_remote_dyn_filters[slot] = remote_dyn_filter_pushdown.captured_dyn_filters; + drop(captured_remote_dyn_filters); let new_self = Arc::new(self.clone()); Ok(FilterPushdownPropagation { - filters: remote_dyn_filter_pushdown - .pushed_down - .into_iter() - .map(|pushdown_ready| { - if pushdown_ready { - PushedDown::Yes - } else { - PushedDown::No - } - }) - .collect(), + filters: remote_dyn_filter_pushdown.pushed_down, updated_node: Some(new_self), }) } @@ -1713,6 +1730,8 @@ mod tests { )) as Arc; let captured = capture_remote_dyn_filters_for_pushdown( RemoteDynFilterProducerId::new(42), + 0, + &[], vec![dyn_filter], ); assert_eq!(captured.captured_dyn_filters.len(), 1); diff --git a/tests/cases/distributed/optimizer/time_index_filter_pushdown.result b/tests/cases/distributed/optimizer/time_index_filter_pushdown.result index f03818154c1..5530edbcead 100644 --- a/tests/cases/distributed/optimizer/time_index_filter_pushdown.result +++ b/tests/cases/distributed/optimizer/time_index_filter_pushdown.result @@ -160,6 +160,88 @@ WHERE | 8 | windows | 2023-06-12T02:10:00 | +------+---------+---------------------+ +-- The resolved scalar subquery reaches the region scans as a dynamic filter. +-- SQLNESS REPLACE (-+) - +-- SQLNESS REPLACE (\s\s+) _ +-- SQLNESS REPLACE (metrics.*) REDACTED +-- SQLNESS REPLACE (peers.*) REDACTED +-- SQLNESS REPLACE region=\d+\(\d+,\s+\d+\) region=REDACTED +-- SQLNESS REPLACE "partition_count":\{(.*?)\} "partition_count":REDACTED +-- SQLNESS REPLACE "flat_format":\s\w+, "flat_format": REDACTED, +-- SQLNESS REPLACE (?m)^\|_\|_\|_SortExec:.*\n +-- SQLNESS REPLACE FilterExec:\sDynamicFilter\s\[[^\]]*\] FilterExec: DynamicFilter [ REDACTED ] +-- SQLNESS REPLACE "files":\s\[.*?\],\s +-- SQLNESS REPLACE IS\sNULL\sOR\sgreptime_timestamp@0\s>\s\d+ IS NULL OR greptime_timestamp@0 > REDACTED +-- SQLNESS REPLACE (RoundRobinBatch.*) REDACTED +-- SQLNESS REPLACE Total\s+rows:\s+\d+ Total rows: REDACTED +-- SQLNESS REPLACE UnorderedScan:.*"dyn_filters":\s(\["DynamicFilter\s\[[^\]]*\]"\]).* UnorderedScan: REDACTED, "dyn_filters": $1 +EXPLAIN ANALYZE VERBOSE SELECT + rack, + os, + greptime_timestamp +FROM + cpu +WHERE + greptime_timestamp = ( + SELECT + greptime_timestamp + FROM + cpu + ORDER BY + greptime_timestamp DESC + LIMIT + 1 + ); + ++-+-+-+ +| stage | node | plan_| ++-+-+-+ +| 0_| 0_|_ScalarSubqueryExec REDACTED +|_|_|_FilterExec: greptime_timestamp@2 = scalar_subquery(1686535800000000000) REDACTED +|_|_|_ProjectionExec: expr=[rack@0 as rack, os@1 as os, greptime_timestamp@3 as greptime_timestamp] REDACTED +|_|_|_CooperativeExec REDACTED +|_|_|_MergeScanExec: REDACTED +|_|_|_MergeSortExec: [greptime_timestamp@0 DESC], fetch=1 REDACTED +|_|_|_MergeScanExec: REDACTED +|_|_|_| +| 1_| 0_|_CooperativeExec REDACTED +|_|_|_UnorderedScan: REDACTED, "dyn_filters": ["DynamicFilter [ greptime_timestamp@3 = 1686535800000000000 ]"] +|_|_|_| +| 1_| 1_|_CooperativeExec REDACTED +|_|_|_UnorderedScan: REDACTED, "dyn_filters": ["DynamicFilter [ greptime_timestamp@3 = 1686535800000000000 ]"] +|_|_|_| +| 1_| 2_|_CooperativeExec REDACTED +|_|_|_UnorderedScan: REDACTED, "dyn_filters": ["DynamicFilter [ greptime_timestamp@3 = 1686535800000000000 ]"] +|_|_|_| +| 1_| 3_|_CooperativeExec REDACTED +|_|_|_UnorderedScan: REDACTED, "dyn_filters": ["DynamicFilter [ greptime_timestamp@3 = 1686535800000000000 ]"] +|_|_|_| +| 1_| 4_|_CooperativeExec REDACTED +|_|_|_UnorderedScan: REDACTED, "dyn_filters": ["DynamicFilter [ greptime_timestamp@3 = 1686535800000000000 ]"] +|_|_|_| +| 1_| 0_|_FilterExec: DynamicFilter [ REDACTED ] REDACTED +|_|_|_SortPreservingMergeExec: [greptime_timestamp@0 DESC], fetch=1 REDACTED +|_|_|_UnorderedScan: REDACTED, "dyn_filters": ["DynamicFilter [ greptime_timestamp@0 IS NULL OR greptime_timestamp@0 > REDACTED ]"] +|_|_|_| +| 1_| 1_|_FilterExec: DynamicFilter [ REDACTED ] REDACTED +|_|_|_SortPreservingMergeExec: [greptime_timestamp@0 DESC], fetch=1 REDACTED +|_|_|_UnorderedScan: REDACTED, "dyn_filters": ["DynamicFilter [ greptime_timestamp@0 IS NULL OR greptime_timestamp@0 > REDACTED ]"] +|_|_|_| +| 1_| 2_|_FilterExec: DynamicFilter [ REDACTED ] REDACTED +|_|_|_SortPreservingMergeExec: [greptime_timestamp@0 DESC], fetch=1 REDACTED +|_|_|_UnorderedScan: REDACTED, "dyn_filters": ["DynamicFilter [ greptime_timestamp@0 IS NULL OR greptime_timestamp@0 > REDACTED ]"] +|_|_|_| +| 1_| 3_|_FilterExec: DynamicFilter [ REDACTED ] REDACTED +|_|_|_SortPreservingMergeExec: [greptime_timestamp@0 DESC], fetch=1 REDACTED +|_|_|_UnorderedScan: REDACTED, "dyn_filters": ["DynamicFilter [ greptime_timestamp@0 IS NULL OR greptime_timestamp@0 > REDACTED ]"] +|_|_|_| +| 1_| 4_|_FilterExec: DynamicFilter [ REDACTED ] REDACTED +|_|_|_SortPreservingMergeExec: [greptime_timestamp@0 DESC], fetch=1 REDACTED +|_|_|_UnorderedScan: REDACTED, "dyn_filters": ["DynamicFilter [ greptime_timestamp@0 IS NULL OR greptime_timestamp@0 > REDACTED ]"] +|_|_|_| +|_|_| Total rows: REDACTED_| ++-+-+-+ + drop table cpu; Affected Rows: 0 diff --git a/tests/cases/distributed/optimizer/time_index_filter_pushdown.sql b/tests/cases/distributed/optimizer/time_index_filter_pushdown.sql index c9c80c24a61..41e85a4f40a 100644 --- a/tests/cases/distributed/optimizer/time_index_filter_pushdown.sql +++ b/tests/cases/distributed/optimizer/time_index_filter_pushdown.sql @@ -102,4 +102,37 @@ WHERE 1 ); +-- The resolved scalar subquery reaches the region scans as a dynamic filter. +-- SQLNESS REPLACE (-+) - +-- SQLNESS REPLACE (\s\s+) _ +-- SQLNESS REPLACE (metrics.*) REDACTED +-- SQLNESS REPLACE (peers.*) REDACTED +-- SQLNESS REPLACE region=\d+\(\d+,\s+\d+\) region=REDACTED +-- SQLNESS REPLACE "partition_count":\{(.*?)\} "partition_count":REDACTED +-- SQLNESS REPLACE "flat_format":\s\w+, "flat_format": REDACTED, +-- SQLNESS REPLACE (?m)^\|_\|_\|_SortExec:.*\n +-- SQLNESS REPLACE FilterExec:\sDynamicFilter\s\[[^\]]*\] FilterExec: DynamicFilter [ REDACTED ] +-- SQLNESS REPLACE "files":\s\[.*?\],\s +-- SQLNESS REPLACE IS\sNULL\sOR\sgreptime_timestamp@0\s>\s\d+ IS NULL OR greptime_timestamp@0 > REDACTED +-- SQLNESS REPLACE (RoundRobinBatch.*) REDACTED +-- SQLNESS REPLACE Total\s+rows:\s+\d+ Total rows: REDACTED +-- SQLNESS REPLACE UnorderedScan:.*"dyn_filters":\s(\["DynamicFilter\s\[[^\]]*\]"\]).* UnorderedScan: REDACTED, "dyn_filters": $1 +EXPLAIN ANALYZE VERBOSE SELECT + rack, + os, + greptime_timestamp +FROM + cpu +WHERE + greptime_timestamp = ( + SELECT + greptime_timestamp + FROM + cpu + ORDER BY + greptime_timestamp DESC + LIMIT + 1 + ); + drop table cpu; diff --git a/tests/cases/standalone/optimizer/time_index_filter_pushdown.result b/tests/cases/standalone/optimizer/time_index_filter_pushdown.result index 50f39764ca2..106a78c49d2 100644 --- a/tests/cases/standalone/optimizer/time_index_filter_pushdown.result +++ b/tests/cases/standalone/optimizer/time_index_filter_pushdown.result @@ -223,6 +223,88 @@ WHERE | 8 | windows | 2023-06-12T02:10:00 | +------+---------+---------------------+ +-- The resolved scalar subquery reaches the region scans as a dynamic filter. +-- SQLNESS REPLACE (-+) - +-- SQLNESS REPLACE (\s\s+) _ +-- SQLNESS REPLACE (metrics.*) REDACTED +-- SQLNESS REPLACE (peers.*) REDACTED +-- SQLNESS REPLACE region=\d+\(\d+,\s+\d+\) region=REDACTED +-- SQLNESS REPLACE "partition_count":\{(.*?)\} "partition_count":REDACTED +-- SQLNESS REPLACE "flat_format":\s\w+, "flat_format": REDACTED, +-- SQLNESS REPLACE (?m)^\|_\|_\|_SortExec:.*\n +-- SQLNESS REPLACE FilterExec:\sDynamicFilter\s\[[^\]]*\] FilterExec: DynamicFilter [ REDACTED ] +-- SQLNESS REPLACE "files":\s\[.*?\],\s +-- SQLNESS REPLACE IS\sNULL\sOR\sgreptime_timestamp@0\s>\s\d+ IS NULL OR greptime_timestamp@0 > REDACTED +-- SQLNESS REPLACE (RoundRobinBatch.*) REDACTED +-- SQLNESS REPLACE Total\s+rows:\s+\d+ Total rows: REDACTED +-- SQLNESS REPLACE UnorderedScan:.*"dyn_filters":\s(\["DynamicFilter\s\[[^\]]*\]"\]).* UnorderedScan: REDACTED, "dyn_filters": $1 +EXPLAIN ANALYZE VERBOSE SELECT + rack, + os, + greptime_timestamp +FROM + cpu +WHERE + greptime_timestamp = ( + SELECT + greptime_timestamp + FROM + cpu + ORDER BY + greptime_timestamp DESC + LIMIT + 1 + ); + ++-+-+-+ +| stage | node | plan_| ++-+-+-+ +| 0_| 0_|_ScalarSubqueryExec REDACTED +|_|_|_FilterExec: greptime_timestamp@2 = scalar_subquery(1686535800000000000) REDACTED +|_|_|_ProjectionExec: expr=[rack@0 as rack, os@1 as os, greptime_timestamp@4 as greptime_timestamp] REDACTED +|_|_|_CooperativeExec REDACTED +|_|_|_MergeScanExec: REDACTED +|_|_|_MergeSortExec: [greptime_timestamp@0 DESC], fetch=1 REDACTED +|_|_|_MergeScanExec: REDACTED +|_|_|_| +| 1_| 0_|_CooperativeExec REDACTED +|_|_|_UnorderedScan: REDACTED, "dyn_filters": ["DynamicFilter [ greptime_timestamp@4 = 1686535800000000000 ]"] +|_|_|_| +| 1_| 1_|_CooperativeExec REDACTED +|_|_|_UnorderedScan: REDACTED, "dyn_filters": ["DynamicFilter [ greptime_timestamp@4 = 1686535800000000000 ]"] +|_|_|_| +| 1_| 2_|_CooperativeExec REDACTED +|_|_|_UnorderedScan: REDACTED, "dyn_filters": ["DynamicFilter [ greptime_timestamp@4 = 1686535800000000000 ]"] +|_|_|_| +| 1_| 3_|_CooperativeExec REDACTED +|_|_|_UnorderedScan: REDACTED, "dyn_filters": ["DynamicFilter [ greptime_timestamp@4 = 1686535800000000000 ]"] +|_|_|_| +| 1_| 4_|_CooperativeExec REDACTED +|_|_|_UnorderedScan: REDACTED, "dyn_filters": ["DynamicFilter [ greptime_timestamp@4 = 1686535800000000000 ]"] +|_|_|_| +| 1_| 0_|_FilterExec: DynamicFilter [ REDACTED ] REDACTED +|_|_|_SortPreservingMergeExec: [greptime_timestamp@0 DESC], fetch=1 REDACTED +|_|_|_UnorderedScan: REDACTED, "dyn_filters": ["DynamicFilter [ greptime_timestamp@0 IS NULL OR greptime_timestamp@0 > REDACTED ]"] +|_|_|_| +| 1_| 1_|_FilterExec: DynamicFilter [ REDACTED ] REDACTED +|_|_|_SortPreservingMergeExec: [greptime_timestamp@0 DESC], fetch=1 REDACTED +|_|_|_UnorderedScan: REDACTED, "dyn_filters": ["DynamicFilter [ greptime_timestamp@0 IS NULL OR greptime_timestamp@0 > REDACTED ]"] +|_|_|_| +| 1_| 2_|_FilterExec: DynamicFilter [ REDACTED ] REDACTED +|_|_|_SortPreservingMergeExec: [greptime_timestamp@0 DESC], fetch=1 REDACTED +|_|_|_UnorderedScan: REDACTED, "dyn_filters": ["DynamicFilter [ greptime_timestamp@0 IS NULL OR greptime_timestamp@0 > REDACTED ]"] +|_|_|_| +| 1_| 3_|_FilterExec: DynamicFilter [ REDACTED ] REDACTED +|_|_|_SortPreservingMergeExec: [greptime_timestamp@0 DESC], fetch=1 REDACTED +|_|_|_UnorderedScan: REDACTED, "dyn_filters": ["DynamicFilter [ greptime_timestamp@0 IS NULL OR greptime_timestamp@0 > REDACTED ]"] +|_|_|_| +| 1_| 4_|_FilterExec: DynamicFilter [ REDACTED ] REDACTED +|_|_|_SortPreservingMergeExec: [greptime_timestamp@0 DESC], fetch=1 REDACTED +|_|_|_UnorderedScan: REDACTED, "dyn_filters": ["DynamicFilter [ greptime_timestamp@0 IS NULL OR greptime_timestamp@0 > REDACTED ]"] +|_|_|_| +|_|_| Total rows: REDACTED_| ++-+-+-+ + drop table cpu; Affected Rows: 0 diff --git a/tests/cases/standalone/optimizer/time_index_filter_pushdown.sql b/tests/cases/standalone/optimizer/time_index_filter_pushdown.sql index 1dc010e1bf4..64f0fbc0e26 100644 --- a/tests/cases/standalone/optimizer/time_index_filter_pushdown.sql +++ b/tests/cases/standalone/optimizer/time_index_filter_pushdown.sql @@ -140,4 +140,37 @@ WHERE 1 ); +-- The resolved scalar subquery reaches the region scans as a dynamic filter. +-- SQLNESS REPLACE (-+) - +-- SQLNESS REPLACE (\s\s+) _ +-- SQLNESS REPLACE (metrics.*) REDACTED +-- SQLNESS REPLACE (peers.*) REDACTED +-- SQLNESS REPLACE region=\d+\(\d+,\s+\d+\) region=REDACTED +-- SQLNESS REPLACE "partition_count":\{(.*?)\} "partition_count":REDACTED +-- SQLNESS REPLACE "flat_format":\s\w+, "flat_format": REDACTED, +-- SQLNESS REPLACE (?m)^\|_\|_\|_SortExec:.*\n +-- SQLNESS REPLACE FilterExec:\sDynamicFilter\s\[[^\]]*\] FilterExec: DynamicFilter [ REDACTED ] +-- SQLNESS REPLACE "files":\s\[.*?\],\s +-- SQLNESS REPLACE IS\sNULL\sOR\sgreptime_timestamp@0\s>\s\d+ IS NULL OR greptime_timestamp@0 > REDACTED +-- SQLNESS REPLACE (RoundRobinBatch.*) REDACTED +-- SQLNESS REPLACE Total\s+rows:\s+\d+ Total rows: REDACTED +-- SQLNESS REPLACE UnorderedScan:.*"dyn_filters":\s(\["DynamicFilter\s\[[^\]]*\]"\]).* UnorderedScan: REDACTED, "dyn_filters": $1 +EXPLAIN ANALYZE VERBOSE SELECT + rack, + os, + greptime_timestamp +FROM + cpu +WHERE + greptime_timestamp = ( + SELECT + greptime_timestamp + FROM + cpu + ORDER BY + greptime_timestamp DESC + LIMIT + 1 + ); + drop table cpu;