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<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]
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<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);
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;