RFC: Remote Dynamic Filter Propagation

Summary

This RFC proposes a remote dynamic filter propagation mechanism for distributed queries. It lets frontend-produced dynamic filters reach remote datanode scans through a lightweight control plane, while preserving one rule: remote dynamic filters are an optimization only, never a correctness dependency.

Motivation

Today, dynamic filters can improve local execution, but they do not automatically propagate to remote datanode scans in distributed queries. As a result, the frontend may already know that a probe-side scan can be narrowed, while the remote scan still runs with a weaker predicate and loses pruning opportunities.

We want a minimal design that:

• propagates dynamic filter updates to remote scans,
• keeps filter identity and lifecycle stable across register/update/unregister,
• and safely degrades when encoding, routing, RPC, or apply logic fails.

Details

The high-level flow is:

1. A join on the frontend produces an alive dynamic filter.
2. MergeScanExec identifies the remote subscribers, generates a stable filter_id, and registers the alive filter into a query-scoped registry.
3. The initial remote read establishes the corresponding registration on the datanode side.
4. The frontend registry watches for dynamic filter updates via wait_update or generation changes.
5. Later updates and unregister messages are sent through the existing region unary RPC path.
6. The datanode applies these updates to query-scoped remote filter state and scan wrappers.
7. Query finish, cancel, or no-consumer conditions trigger unregister and cleanup.

Identity

The logical identity of a remote dynamic filter is query_id + filter_id.

Region and scan metadata are routing information, not part of filter-state identity. filter_id only needs to be stable and unique within a single query.

The current recommendation is to derive filter_id from:

• region_id
• producer-local ordinal
• canonicalized children fingerprint

The following should not be included:

• partition
• transport metadata
• memory addresses or temporary runtime object ids

Transport

This design reuses the existing region unary gRPC path:

• RegionRequest.body.remote_dyn_filter
• RemoteDynFilterRequest.oneof action
  • update
  • unregister

The initial remote read is responsible for register and scan setup. The unary RPC path is only for later update and unregister messages.

Frontend registry

The frontend uses a query-engine runtime map:

• implementation near src/query/src/dist_plan/remote_dyn_filter_registry.rs
• storage model: query_id -> Arc<RemoteDynFilterRegistry>

This registry should not live on a single MergeScanExec, and it should not be stored in QueryContext.mutable_session_data. It is a query execution runtime object that owns watcher tasks, cleanup tail, and fanout state.

The registry lifecycle has three states:

• Active: accepts registrations and sends updates
• Closing: query ended; stop new registrations, send final cleanup messages, drain in-flight RPCs
• Closed: watchers stopped, state removable from the runtime map

The registry may outlive the main query execution briefly for cleanup, but it is not intended to be a long-lived global object.

Propagation policy

Remote dynamic filters should remain a selective optimization, not an automatic fanout for every filter update.

The frontend may skip remote propagation when the encoded filter becomes too large, fanout cost is too high, or the expected pruning benefit is too small. In those cases, execution should continue with local-only dynamic filtering semantics.

Responsibilities

On the frontend:

• the join produces alive dynamic filters,
• MergeScanExec bridges producers to remote subscribers,
• the registry watches updates and fans out RPCs.

On the datanode:

• the unary handler receives update and unregister,
• query-scoped remote filter state is keyed by query_id + filter_id,
• remote wrappers apply updates through existing predicate and scan refresh paths.

Failure semantics

All failures must degrade safely:

• encode failure -> local-only filter
• RPC failure -> log/metric and degrade
• early update or missing target -> explicit buffer, drop+metric, or retry policy
• decode or remap failure -> disable remote optimization only

Alternatives

Registry on MergeScanExec

Rejected because lifecycle and cleanup would become fragmented across multiple bridge or exec instances in the same query.

Registry in QueryContext.mutable_session_data

Rejected because this is the wrong ownership model. The registry is not session metadata; it is a query runtime object with watcher tasks and cleanup behavior.

Long-lived global manager

Rejected for now because it is heavier than necessary. A query-engine runtime map is sufficient for the current design.

Drawbacks

• The design introduces extra query runtime state and cleanup logic on both frontend and datanode.
• The initial version only covers the current minimal filter form and leaves larger membership propagation to later work.
• A clear policy is still required for updates that arrive before scan registration.

Unresolved questions

1. Should children fingerprint canonicalization become a shared helper?
2. What is the strict timing relationship between is_complete and final unregister?
3. Does the runtime map need a background sweep task, or is explicit reap enough?
4. How should large build-side membership evolve beyond IN in later work?