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 <killme2008@gmail.com>

* 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 <killme2008@gmail.com>

* 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 <killme2008@gmail.com>

* chore(query): note the IN fallback restriction for scalar subquery filters

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* chore(query): document the scalar subquery dyn filter bridge removal condition

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* 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 <killme2008@gmail.com>

---------

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

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

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<DynamicFilterPhysicalExpr>,
+    /// 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<Arc<dyn PhysicalExpr>>,
 }
 
 #[derive(Debug, Clone)]
 pub(crate) struct RemoteDynFilterPushdown {
     pub(crate) captured_dyn_filters: Vec<CapturedDynFilter>,
     /// Preflight result per parent filter.
-    pub(crate) pushed_down: Vec<bool>,
+    pub(crate) pushed_down: Vec<PushedDown>,
 }
 
+/// 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<Arc<dyn datafusion::physical_plan::PhysicalExpr>>,
 ) -> 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::<Vec<_>>();
+    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<dyn PhysicalExpr>) -> bool {
+    let mut has_scalar_subquery = false;
+    let mut supported = true;
+    let _ = filter.apply(|expr| {
+        if expr.is::<ScalarSubqueryExpr>() {
+            has_scalar_subquery = true;
+        } else if expr.is::<InListExpr>()
+            || (expr.children().is_empty() && !expr.is::<Column>() && !expr.is::<Literal>())
+        {
+            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<dyn PhysicalExpr>,
+) -> Result<CapturedDynFilter> {
+    let mut columns = collect_columns(&filter).into_iter().collect::<Vec<_>>();
+    columns.sort_by_key(|column| column.index());
+    let children = columns
+        .into_iter()
+        .map(|column| Arc::new(column) as Arc<dyn PhysicalExpr>)
+        .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<dyn PhysicalExpr>) -> Result<Arc<dyn PhysicalExpr>> {
+    let empty_batch =
+        arrow::record_batch::RecordBatch::new_empty(Arc::new(arrow_schema::Schema::empty()));
+    filter
+        .clone()
+        .transform_up(|expr| {
+            if !expr.is::<ScalarSubqueryExpr>() {
+                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<dyn datafusion::physical_plan::PhysicalExpr>,
 ) -> Option<Arc<DynamicFilterPhysicalExpr>> {
@@ -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<bool> {
+        pushdown
+            .pushed_down
+            .iter()
+            .map(|pushed_down| matches!(pushed_down, PushedDown::Yes))
+            .collect()
+    }
+
+    fn host_eq_scalar_subquery(results: &ScalarSubqueryResults) -> Arc<dyn PhysicalExpr> {
+        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<dyn PhysicalExpr>,
+        ];
+
+        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<dyn PhysicalExpr>;
+
+        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<dyn PhysicalExpr>
+                })
+                .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]

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<usize> {
     }
 }
 
+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<bool> {
     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<RemoteDynFilterProducerId>,
-    captured_remote_dyn_filters: Arc<Mutex<Vec<CapturedDynFilter>>>,
+    /// 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<Mutex<[Vec<CapturedDynFilter>; 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<CapturedDynFilter> {
-        self.captured_remote_dyn_filters.lock().unwrap().clone()
+        self.captured_remote_dyn_filters.lock().unwrap().concat()
     }
 
     pub fn sub_stage_metrics(&self) -> Vec<RecordBatchMetrics> {
@@ -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<FilterPushdownPropagation<Arc<dyn ExecutionPlan>>> {
@@ -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<dyn datafusion_physical_expr::PhysicalExpr>;
         let captured = capture_remote_dyn_filters_for_pushdown(
             RemoteDynFilterProducerId::new(42),
+            0,
+            &[],
             vec![dyn_filter],
         );
         assert_eq!(captured.captured_dyn_filters.len(), 1);

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

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;

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

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;