diff --git a/src/common/query/src/request/initial_remote_dyn_filter_reg.rs b/src/common/query/src/request/initial_remote_dyn_filter_reg.rs index 01f193384b..16968e849b 100644 --- a/src/common/query/src/request/initial_remote_dyn_filter_reg.rs +++ b/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>, /// 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, } @@ -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, } diff --git a/src/query/src/dist_plan/dyn_filter_bridge.rs b/src/query/src/dist_plan/dyn_filter_bridge.rs index fe9ec39e3f..546aa20778 100644 --- a/src/query/src/dist_plan/dyn_filter_bridge.rs +++ b/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, - /// 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, } @@ -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 { - // 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( ¤t, @@ -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, @@ -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) diff --git a/src/query/src/dist_plan/filter_id.rs b/src/query/src/dist_plan/filter_id.rs index ba22907cbf..d3e7929336 100644 --- a/src/query/src/dist_plan/filter_id.rs +++ b/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, diff --git a/src/query/src/dist_plan/merge_scan.rs b/src/query/src/dist_plan/merge_scan.rs index 9d0526c769..2201d0c52f 100644 --- a/src/query/src/dist_plan/merge_scan.rs +++ b/src/query/src/dist_plan/merge_scan.rs @@ -193,11 +193,7 @@ pub struct MergeScanExec { /// Metrics for each partition partition_metrics: Arc>>, 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, captured_remote_dyn_filters: Arc>>, target_partition: usize, @@ -728,8 +724,7 @@ impl ExecutionPlan for MergeScanExec { .map(|filter| filter.filter) .collect::>(); 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(), diff --git a/src/query/src/dist_plan/remote_dyn_filter_registry.rs b/src/query/src/dist_plan/remote_dyn_filter_registry.rs index 4108c03c11..4c33ddd960 100644 --- a/src/query/src/dist_plan/remote_dyn_filter_registry.rs +++ b/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, /// 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>, } @@ -216,9 +213,7 @@ impl Drop for RemoteDynFilterRegistryLease { let query_id = registry.query_id(); let registry_weak = Arc::downgrade(®istry); - // 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>>, @@ -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, 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()