Struct DynamicFilterPhysicalExpr 

pub struct DynamicFilterPhysicalExpr { /* private fields */ }

A dynamic PhysicalExpr that can be updated by anyone with a reference to it.

Any ExecutionPlan that uses this expression and holds a reference to it internally should probably also implement ExecutionPlan::reset_state to remain compatible with recursive queries and other situations where the same ExecutionPlan is reused with different data.

Implementations

impl DynamicFilterPhysicalExpr

pub fn new( children: Vec<Arc<dyn PhysicalExpr>>, inner: Arc<dyn PhysicalExpr>, ) -> Self

Create a new DynamicFilterPhysicalExpr from an initial expression and a list of children. The list of children is provided separately because the initial expression may not have the same children. For example, if the initial expression is just true it will not reference any columns, but we may know that we are going to replace this expression with a real one that does reference certain columns. In this case you must pass in the columns that will be used in the final expression as children to this function since DataFusion is generally not compatible with dynamic children in expressions.

To determine the children you can:

• Use collect_columns to collect the columns from the expression.
• Use existing information, such as the sort columns in a SortExec.

Generally the important bit is that the leaf children that reference columns do not change since those will be used to determine what columns need to read or projected when evaluating the expression.

Any ExecutionPlan that uses this expression and holds a reference to it internally should probably also implement ExecutionPlan::reset_state to remain compatible with recursive queries and other situations where the same ExecutionPlan is reused with different data.

pub fn current(&self) -> Result<Arc<dyn PhysicalExpr>>

Get the current expression. This will return the current expression with any children remapped to match calls to PhysicalExpr::with_new_children.

pub fn update(&self, new_expr: Arc<dyn PhysicalExpr>) -> Result<()>

Update the current expression and notify all waiters. Any children of this expression must be a subset of the original children passed to the constructor. This should be called e.g.:

• When we’ve computed the probe side’s hash table in a HashJoinExec
• After every batch is processed if we update the TopK heap in a SortExec using a TopK approach.

pub fn mark_complete(&self)

Mark this dynamic filter as complete and broadcast to all waiters.

This signals that all expected updates have been received. Waiters using Self::wait_complete will be notified.

pub async fn wait_update(&self)

Wait asynchronously for any update to this filter.

This method will return when Self::update is called and the generation increases. It does not guarantee that the filter is complete.

pub async fn wait_complete(self: &Arc<Self>)

Wait asynchronously until this dynamic filter is marked as complete.

This method returns immediately if the filter is already complete or if the filter is not being used by any consumers. Otherwise, it waits until Self::mark_complete is called.

Unlike Self::wait_update, this method guarantees that when it returns, the filter is fully complete with no more updates expected.

pub fn is_used(self: &Arc<Self>) -> bool

Check if this dynamic filter is being actively used by any consumers.

Returns true if there are references beyond the producer (e.g., the HashJoinExec that created the filter). This is useful to avoid computing expensive filter expressions when no consumer will actually use them.

Note: We check the inner Arc’s strong_count, not the outer Arc’s count, because when filters are transformed (e.g., via reassign_expr_columns during filter pushdown), new outer Arc instances are created via with_new_children(), but they all share the same inner Arc<RwLock<Inner>>. This is what allows filter updates to propagate to consumers even after transformation.

Trait Implementations

impl Debug for DynamicFilterPhysicalExpr

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl Display for DynamicFilterPhysicalExpr

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl Hash for DynamicFilterPhysicalExpr

fn hash<H: Hasher>(&self, state: &mut H)

Feeds this value into the given Hasher.

fn hash_slice<H>(data: &[Self], state: &mut H)

where H: Hasher, Self: Sized,

Feeds a slice of this type into the given Hasher.

impl PartialEq for DynamicFilterPhysicalExpr

fn eq(&self, other: &Self) -> bool

Tests for self and other values to be equal, and is used by ==.

fn ne(&self, other: &Rhs) -> bool

Tests for !=. The default implementation is almost always sufficient, and should not be overridden without very good reason.

impl PhysicalExpr for DynamicFilterPhysicalExpr

fn as_any(&self) -> &dyn Any

Returns the physical expression as Any so that it can be downcast to a specific implementation.

fn children(&self) -> Vec<&Arc<dyn PhysicalExpr>>

Get a list of child PhysicalExpr that provide the input for this expr.

fn with_new_children( self: Arc<Self>, children: Vec<Arc<dyn PhysicalExpr>>, ) -> Result<Arc<dyn PhysicalExpr>>

Returns a new PhysicalExpr where all children were replaced by new exprs.

fn data_type(&self, input_schema: &Schema) -> Result<DataType>

Get the data type of this expression, given the schema of the input

fn nullable(&self, input_schema: &Schema) -> Result<bool>

Determine whether this expression is nullable, given the schema of the input

fn evaluate(&self, batch: &RecordBatch) -> Result<ColumnarValue>

Evaluate an expression against a RecordBatch

fn fmt_sql(&self, f: &mut Formatter<'_>) -> Result

Format this PhysicalExpr in nice human readable “SQL” format

fn snapshot(&self) -> Result<Option<Arc<dyn PhysicalExpr>>>

Take a snapshot of this PhysicalExpr, if it is dynamic.

fn snapshot_generation(&self) -> u64

Returns the generation of this PhysicalExpr for snapshotting purposes. The generation is an arbitrary u64 that can be used to track changes in the state of the PhysicalExpr over time without having to do an exhaustive comparison. This is useful to avoid unnecessary computation or serialization if there are no changes to the expression. In particular, dynamic expressions that may change over time; this allows cheap checks for changes. Static expressions that do not change over time should return 0, as does the default implementation. You should not call this method directly as it does not handle recursion. Instead use snapshot_generation to handle recursion and capture the full state of the PhysicalExpr.

fn return_field( &self, input_schema: &Schema, ) -> Result<Arc<Field>, DataFusionError>

The output field associated with this expression

fn evaluate_selection( &self, batch: &RecordBatch, selection: &BooleanArray, ) -> Result<ColumnarValue, DataFusionError>

Evaluate an expression against a RecordBatch after first applying a validity array

fn evaluate_bounds( &self, _children: &[&Interval], ) -> Result<Interval, DataFusionError>

Computes the output interval for the expression, given the input intervals.

fn propagate_constraints( &self, _interval: &Interval, _children: &[&Interval], ) -> Result<Option<Vec<Interval>>, DataFusionError>

Updates bounds for child expressions, given a known interval for this expression.

fn evaluate_statistics( &self, children: &[&Distribution], ) -> Result<Distribution, DataFusionError>

Computes the output statistics for the expression, given the input statistics.

fn propagate_statistics( &self, parent: &Distribution, children: &[&Distribution], ) -> Result<Option<Vec<Distribution>>, DataFusionError>

Updates children statistics using the given parent statistic for this expression.

fn get_properties( &self, _children: &[ExprProperties], ) -> Result<ExprProperties, DataFusionError>

Calculates the properties of this PhysicalExpr based on its children’s properties (i.e. order and range), recursively aggregating the information from its children. In cases where the PhysicalExpr has no children (e.g., Literal or Column), these properties should be specified externally, as the function defaults to unknown properties.

fn is_volatile_node(&self) -> bool

Returns true if the expression node is volatile, i.e. whether it can return different results when evaluated multiple times with the same input.

impl Eq for DynamicFilterPhysicalExpr