docs: better comment, rm some slop

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

src/common/query/src/request/initial_remote_dyn_filter_reg.rs

@@ -30,9 +30,7 @@ pub const INITIAL_REMOTE_DYN_FILTER_REGISTRATIONS_EXTENSION_KEY: &str =
 pub const INITIAL_REMOTE_DYN_FILTER_REGS_MAX_COUNT: usize = 64;
 /// Raw encoded registration byte budget for initial remote dynamic filter registrations.
 ///
-/// This counts DataFusion physical expression proto bytes and optional initial snapshot payload
-/// bytes before serde JSON/base64 expansion. It is a payload budget, not an exact final
-/// QueryContext extension string-size limit.
+/// Counts proto payload bytes before JSON/base64 expansion, not the final extension size.
 pub const INITIAL_REMOTE_DYN_FILTER_REGS_MAX_TOTAL_PROTO_BYTES: usize = 64 * 1024;
 
 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
@@ -114,9 +112,8 @@ pub struct InitialDynFilterReg {
     pub child_exprs_datafusion_proto: Vec<Vec<u8>>,
     /// Optional producer-side predicate snapshot captured at initial registration time.
     ///
-    /// It is skipped when absent so older/no-snapshot registrations remain wire-compatible.
-    /// Datanode runtime code should treat this as an update that arrived before the runtime
-    /// `DynamicFilterPhysicalExpr` was created, not as registration identity metadata.
+    /// This is only an initial pending update for the remote runtime filter. It is not part of
+    /// registration identity; identity is carried by `filter_id` and child expressions.
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub initial_snapshot: Option<InitialDynFilterSnapshot>,
 }
@@ -124,13 +121,9 @@ pub struct InitialDynFilterReg {
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
 pub struct InitialDynFilterSnapshot {
     pub payload: DynFilterPayload,
-    /// Producer-side snapshot generation. Receivers use this with normal update generation
-    /// ordering to ignore stale payloads and apply newer payloads.
+    /// Producer-side generation used to ignore stale snapshots.
     pub generation: u64,
     /// Whether this snapshot completes the dynamic filter stream.
-    ///
-    /// 3a currently sets this conservatively to false because DataFusion exposes synchronous
-    /// generation/current reads but not a synchronous public completion getter.
     pub is_complete: bool,
 }
 

src/query/src/dist_plan/dyn_filter_bridge.rs

@@ -39,9 +39,7 @@ pub(crate) struct CapturedDynFilter {
 #[derive(Debug, Clone)]
 pub(crate) struct RemoteDynFilterPushdown {
     pub(crate) captured_dyn_filters: Vec<CapturedDynFilter>,
-    /// Per-parent-filter preflight readiness. `true` means the initial registration can be
-    /// constructed and serialized; the caller may still choose to return `PushedDown::No` until
-    /// datanode-side consumption is available.
+    /// Preflight result per parent filter.
     pub(crate) pushed_down: Vec<bool>,
 }
 
@@ -74,17 +72,6 @@ pub(crate) fn capture_remote_dyn_filters_for_pushdown(
         }
     }
 
-    // Initial snapshots are an optional warm-start optimization. If validation fails while any
-    // snapshot is attached, retry once without snapshots before rejecting base registrations. This
-    // broad fallback is intentional: snapshot bytes are the only optional part here, and the second
-    // validation below still rejects non-snapshot errors such as duplicate ids or count limits.
-    if has_initial_snapshots(&captured_dyn_filters)
-        && let Err(error) = validate_initial_registrations_for_pushdown(&captured_dyn_filters)
-    {
-        common_telemetry::warn!(error; "Initial remote dyn filter registrations failed validation with optional snapshots; retrying without snapshots");
-        drop_initial_snapshots(&mut captured_dyn_filters);
-    }
-
     if let Err(error) = validate_initial_registrations_for_pushdown(&captured_dyn_filters) {
         common_telemetry::warn!(error; "Remote dyn filters are not pushed down because initial registrations are invalid");
         return RemoteDynFilterPushdown {
@@ -142,10 +129,7 @@ fn build_captured_dyn_filter(
 
     Ok(CapturedDynFilter {
         filter_id,
-        initial_registration: attach_stable_initial_snapshot(
-            initial_registration,
-            &alive_dyn_filter,
-        ),
+        initial_registration: attach_initial_snapshot(initial_registration, &alive_dyn_filter),
         alive_dyn_filter,
     })
 }
@@ -160,43 +144,21 @@ fn validate_initial_registrations_for_pushdown(
     Ok(())
 }
 
-fn drop_initial_snapshots(captured_dyn_filters: &mut [CapturedDynFilter]) {
-    for captured in captured_dyn_filters {
-        captured.initial_registration.initial_snapshot = None;
-    }
-}
-
-fn has_initial_snapshots(captured_dyn_filters: &[CapturedDynFilter]) -> bool {
-    captured_dyn_filters
-        .iter()
-        .any(|captured| captured.initial_registration.initial_snapshot.is_some())
-}
-
-fn attach_stable_initial_snapshot(
+fn attach_initial_snapshot(
     initial_registration: InitialDynFilterReg,
     alive_dyn_filter: &DynamicFilterPhysicalExpr,
 ) -> InitialDynFilterReg {
-    let Some(initial_snapshot) = stable_initial_snapshot(alive_dyn_filter) else {
+    let Some(initial_snapshot) = initial_snapshot(alive_dyn_filter) else {
         return initial_registration;
     };
 
     initial_registration.with_initial_snapshot(initial_snapshot)
 }
 
-fn stable_initial_snapshot(
+fn initial_snapshot(
     alive_dyn_filter: &DynamicFilterPhysicalExpr,
 ) -> Option<InitialDynFilterSnapshot> {
-    // DataFusion dynamic filters start at generation 1 with a neutral predicate such as `true`.
-    // Sending that initial value does not improve remote filtering, so only send snapshots after
-    // a real producer update has advanced the generation.
-    let generation_before = alive_dyn_filter.snapshot_generation();
-    if generation_before <= 1 {
-        return None;
-    }
-
-    // Read a stable snapshot by checking the generation before and after `current()`. This relies
-    // on DataFusion's generation being monotonic for each update; if it changes while reading,
-    // omit the optional snapshot and let later async updates carry the new predicate.
+    let generation = alive_dyn_filter.snapshot_generation();
     let current = match alive_dyn_filter.current() {
         Ok(current) => current,
         Err(error) => {
@@ -204,10 +166,6 @@ fn stable_initial_snapshot(
             return None;
         }
     };
-    let generation_after = alive_dyn_filter.snapshot_generation();
-    if generation_before != generation_after {
-        return None;
-    }
 
     let payload = match DynFilterPayload::from_datafusion_expr(
         &current,
@@ -220,12 +178,12 @@ fn stable_initial_snapshot(
         }
     };
 
+    // Current DataFusion exposes `wait_complete()`, but no non-blocking completion getter.
+    let is_complete = false;
     Some(InitialDynFilterSnapshot::new(
         payload,
-        generation_after,
-        // DataFusion does not expose a synchronous public completion getter. Treat the initial
-        // snapshot as a non-terminal pending update; later runtime/update handling owns completion.
-        false,
+        generation,
+        is_complete,
     ))
 }
 
@@ -252,7 +210,7 @@ pub(crate) fn query_context_with_initial_dyn_filter_regs(
     }
 
     if let Err(error) = regs.validate_default_bounds() {
-        common_telemetry::warn!(error; "Dropping initial remote dyn filter registrations for region {} that exceed Task 03 bounds", region_id);
+        common_telemetry::warn!(error; "Dropping initial remote dyn filter registrations for region {} that exceed configured bounds", region_id);
         return region_query_ctx;
     }
 
@@ -454,7 +412,7 @@ mod tests {
             pushdown.captured_dyn_filters[0]
                 .initial_registration
                 .initial_snapshot
-                .is_none()
+                .is_some()
         );
     }
 
@@ -476,7 +434,7 @@ mod tests {
     }
 
     #[test]
-    fn capture_remote_dyn_filters_for_pushdown_omits_neutral_initial_snapshot() {
+    fn capture_remote_dyn_filters_for_pushdown_attaches_initial_snapshot() {
         let parent_filters = vec![Arc::new(DynamicFilterPhysicalExpr::new(
             vec![Arc::new(Column::new("host", 1)) as Arc<_>],
             lit(true) as _,
@@ -493,12 +451,12 @@ mod tests {
             pushdown.captured_dyn_filters[0]
                 .initial_registration
                 .initial_snapshot
-                .is_none()
+                .is_some()
         );
     }
 
     #[test]
-    fn capture_remote_dyn_filters_for_pushdown_attaches_stable_initial_snapshot() {
+    fn capture_remote_dyn_filters_for_pushdown_attaches_initial_snapshot_after_update() {
         let dyn_filter = Arc::new(DynamicFilterPhysicalExpr::new(
             vec![Arc::new(Column::new("host", 1)) as Arc<_>],
             lit(true) as _,
@@ -526,7 +484,7 @@ mod tests {
     }
 
     #[test]
-    fn capture_remote_dyn_filters_for_pushdown_drops_oversized_snapshots_and_keeps_regs() {
+    fn capture_remote_dyn_filters_for_pushdown_rejects_oversized_snapshots() {
         let parent_filters = vec![
             test_dyn_filter_with_snapshot_payload("host", 0, 40 * 1024)
                 as Arc<dyn datafusion::physical_plan::PhysicalExpr>,
@@ -539,18 +497,12 @@ mod tests {
             parent_filters,
         );
 
-        assert_eq!(pushdown.pushed_down, vec![true, true]);
-        assert_eq!(pushdown.captured_dyn_filters.len(), 2);
-        assert!(
-            pushdown
-                .captured_dyn_filters
-                .iter()
-                .all(|captured| { captured.initial_registration.initial_snapshot.is_none() })
-        );
+        assert_eq!(pushdown.pushed_down, vec![false, false]);
+        assert!(pushdown.captured_dyn_filters.is_empty());
     }
 
     #[test]
-    fn capture_remote_dyn_filters_for_pushdown_rejects_regs_still_invalid_after_snapshot_drop() {
+    fn capture_remote_dyn_filters_for_pushdown_rejects_too_many_regs_with_snapshots() {
         const TOO_MANY_INITIAL_REGS: usize = 65;
 
         let parent_filters = (0..TOO_MANY_INITIAL_REGS)

src/query/src/dist_plan/filter_id.rs

@@ -53,9 +53,7 @@ impl FromStr for FilterFingerprint {
 
 /// Query-local identity for one remote dynamic filter producer.
 ///
-/// The analyzer assigns a new id to each frontend-side `MergeScan` rewrite so filters from
-/// independent producers cannot collide even when their producer-local ordinals and child
-/// fingerprints match.
+/// Distinguishes independent producers(MergeScan node) that may share ordinals and child fingerprints.
 #[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
 pub struct RemoteDynFilterProducerId(u64);
 
@@ -90,9 +88,8 @@ pub struct FilterId {
     children_fingerprint: FilterFingerprint,
 }
 
-// NOTE(remote-dyn-filter): FilterId is generated once by the source-side planner/runtime and then
-// propagated through the query lifecycle. Consumers should treat it as the canonical propagated
-// identifier instead of independently recomputing it from local state.
+// NOTE(remote-dyn-filter): FilterId is source-generated and propagated; consumers should not
+// recompute it from local scan state.
 
 impl FilterId {
     pub fn new(
@@ -173,10 +170,7 @@ impl Display for ParseFilterIdError {
 
 /// Builds the query-local remote dynamic filter identity.
 ///
-/// The identity is `remote dynamic filter producer id + producer-local ordinal + canonicalized child fingerprint`.
-/// Subscriber routing details such as `region_id` and `partition` stay outside this key so they
-/// can remain in the later fanout/subscriber map instead of splitting one shared remote filter state.
-/// See [`FilterId`] for the propagation contract.
+/// Identity is producer id + local ordinal + child fingerprint; routing stays outside.
 pub(crate) fn build_remote_dyn_filter_id(
     remote_dyn_filter_producer_id: RemoteDynFilterProducerId,
     producer_local_ordinal: usize,

src/query/src/dist_plan/merge_scan.rs

@@ -193,11 +193,7 @@ pub struct MergeScanExec {
     /// Metrics for each partition
     partition_metrics: Arc<Mutex<HashMap<usize, PartitionMetrics>>>,
     query_ctx: QueryContextRef,
-    /// Optional because remote dynamic filters must fail open.
-    ///
-    /// If producer id assignment is missing, `MergeScanExec` still executes the query normally and
-    /// only skips RDF pushdown/registration while keeping parent-side filters as the correctness
-    /// fallback.
+    /// 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>>>,
     target_partition: usize,
@@ -728,8 +724,7 @@ impl ExecutionPlan for MergeScanExec {
             .map(|filter| filter.filter)
             .collect::<Vec<_>>();
         let Some(remote_dyn_filter_producer_id) = self.remote_dyn_filter_producer_id else {
-            // RDF is an optimization: missing producer identity must only disable RDF, never block
-            // normal query execution or remove the parent-side filter fallback.
+            // Missing RDF identity disables only RDF, not normal execution.
             common_telemetry::warn!(
                 "MergeScan remote dynamic filter producer id is not assigned; skipping remote dynamic filter pushdown"
             );
@@ -752,11 +747,8 @@ impl ExecutionPlan for MergeScanExec {
                 .pushed_down
                 .into_iter()
                 .map(|_pushdown_ready| {
-                    // TODO(remote-dyn-filter): Return `PushedDown::Yes` for `_pushdown_ready`
-                    // filters after datanode-side initial registration consumption and runtime
-                    // pending-update application are implemented. Until then, keep the parent-side
-                    // filter as a correctness fallback because the remote side may ignore the
-                    // registration carried in QueryContext.
+                    // TODO(discord9): Return `PushedDown::Yes` after datanodes consume RDF
+                    // registrations and pending updates. Until then, keep the parent-side filter.
                     PushedDown::No
                 })
                 .collect(),

src/query/src/dist_plan/remote_dyn_filter_registry.rs

@@ -55,7 +55,8 @@ pub enum SubscriberRegistration {
 
 /// A registered query-local remote dynamic filter entry.
 ///
-/// This stores the alive DataFusion filter handle together with minimal subscriber state.
+/// The frontend query owns the strong DataFusion filter handle until the query finishes; the
+/// registry only keeps a weak reference for later updates.
 #[derive(Debug)]
 pub struct DynFilterEntry {
     filter_id: FilterId,
@@ -168,17 +169,13 @@ impl QueryDynFilterRegistry {
 
 /// Stream-scoped lease that keeps a query registry alive.
 ///
-/// The manager only stores a weak index. Holding this lease is the production path for owning a
-/// strong registry reference; this keeps cleanup tied to stream lifetime instead of a hand-rolled
-/// active-stream counter.
+/// Production code owns registries through this lease; the manager only keeps a weak index.
 #[derive(Debug)]
 pub struct RemoteDynFilterRegistryLease {
     registry_manager: Arc<DynFilterRegistryManager>,
     /// Always `Some` while the lease is alive.
     ///
-    /// This is wrapped in `Option` only so `Drop` can `take()` and release the strong `Arc` before
-    /// checking whether the manager's weak index became dead. Without this explicit drop order,
-    /// the field would remain alive until after `Drop::drop` returns.
+    /// `Option` lets `Drop` release the strong `Arc` before pruning the weak index.
     registry: Option<Arc<QueryDynFilterRegistry>>,
 }
 
@@ -216,9 +213,7 @@ impl Drop for RemoteDynFilterRegistryLease {
         let query_id = registry.query_id();
         let registry_weak = Arc::downgrade(&registry);
 
-        // Release this lease's strong reference before checking whether the manager's weak entry is
-        // dead. If two leases drop concurrently, checking `Arc::strong_count` before the fields are
-        // dropped can make both drops observe each other and both skip cleanup.
+        // Release this lease before pruning; concurrent drops must not observe each other's strong refs.
         drop(registry);
 
         let _ = self
@@ -229,9 +224,7 @@ impl Drop for RemoteDynFilterRegistryLease {
 
 /// Query-engine manager for query-scoped remote dynamic filter registries.
 ///
-/// This is an index, not an owner: the map stores `Weak` pointers and active streams own the
-/// registry through [`RemoteDynFilterRegistryLease`]. Keep production access lease-based so
-/// dropping a lease releases one strong owner before the manager prunes a dead weak entry.
+/// Weak index only; active streams own registries through [`RemoteDynFilterRegistryLease`].
 #[derive(Debug, Default)]
 pub struct DynFilterRegistryManager {
     registries: RwLock<HashMap<QueryId, Weak<QueryDynFilterRegistry>>>,
@@ -270,10 +263,7 @@ impl DynFilterRegistryManager {
             .unwrap_or_else(|poisoned| poisoned.into_inner());
         let current = registries.get(query_id)?;
 
-        // Compare `Weak` handles rather than raw pointers. The weak control block stays alive while
-        // either this drop's local weak handle or the manager's weak entry exists. `ptr_eq` prevents
-        // an old lease drop from removing a different registry installed for the same query id, and
-        // `upgrade().is_none()` ensures we only prune dead entries.
+        // `ptr_eq` protects a newer registry for the same query id; `upgrade` ensures it is dead.
         if current.ptr_eq(dropped_registry) && current.upgrade().is_none() {
             registries.remove(query_id)
         } else {
@@ -299,10 +289,7 @@ impl DynFilterRegistryManager {
 
     /// Acquires the stream-owned registry lease for `query_id`.
     ///
-    /// This is the only production API that returns a live registry handle. A concurrent final
-    /// lease drop cannot remove the map entry between lookup and ownership transfer because the
-    /// upgraded `Arc` returned by `get_or_init` is already a strong owner before the manager lock is
-    /// released.
+    /// Returns a lease holding a strong registry reference.
     pub fn acquire_lease(self: &Arc<Self>, query_id: QueryId) -> RemoteDynFilterRegistryLease {
         let registry = self.get_or_init(query_id);
         RemoteDynFilterRegistryLease::new(self.clone(), registry)
@@ -322,8 +309,7 @@ impl DynFilterRegistryManager {
 
     #[cfg(test)]
     pub fn registry_count(&self) -> usize {
-        // Snapshot helper for tests. Lifecycle decisions must not depend on this count; cleanup uses
-        // the lease-owned `Arc` and weak-entry pruning instead.
+        // Test snapshot helper; lifecycle decisions use lease-owned Arcs and weak pruning.
         self.registries
             .read()
             .unwrap()