## Coverage Report

#### Created: 2025-11-22 06:00

next uncovered line (L), next uncovered region (R), next uncovered branch (B)

    /Users/andrewlamb/Software/datafusion/datafusion/physical-expr/src/expressions/dynamic_filters.rs

    Line
    Count
    Source
    1
    // Licensed to the Apache Software Foundation (ASF) under one
    2
    // or more contributor license agreements.  See the NOTICE file
    3
    // distributed with this work for additional information
    4
    // regarding copyright ownership.  The ASF licenses this file
    5
    // to you under the Apache License, Version 2.0 (the
    6
    // "License"); you may not use this file except in compliance
    7
    // with the License.  You may obtain a copy of the License at
    8
    //
    9
    //   http://www.apache.org/licenses/LICENSE-2.0
    10
    //
    11
    // Unless required by applicable law or agreed to in writing,
    12
    // software distributed under the License is distributed on an
    13
    // "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
    14
    // KIND, either express or implied.  See the License for the
    15
    // specific language governing permissions and limitations
    16
    // under the License.
    17
    18
    use parking_lot::RwLock;
    19
    use std::{any::Any, fmt::Display, hash::Hash, sync::Arc};
    20
    use tokio::sync::watch;
    21
    22
    use crate::PhysicalExpr;
    23
    use arrow::datatypes::{DataType, Schema};
    24
    use datafusion_common::{
    25
        tree_node::{Transformed, TransformedResult, TreeNode},
    26
        Result,
    27
    };
    28
    use datafusion_expr::ColumnarValue;
    29
    use datafusion_physical_expr_common::physical_expr::{DynEq, DynHash};
    30
    31
    /// State of a dynamic filter, tracking both updates and completion.
    32
    #[derive(Debug, Clone, Copy, PartialEq, Eq)]
    33
    enum FilterState {
    34
        /// Filter is in progress and may receive more updates.
    35
        InProgress { generation: u64 },
    36
        /// Filter is complete and will not receive further updates.
    37
        Complete { generation: u64 },
    38
    }
    39
    40
    impl FilterState {
    41
    0
        fn generation(&self) -> u64 {
    42
    0
            match self {
    43
    0
                FilterState::InProgress { generation }
    44
    0
                | FilterState::Complete { generation } => *generation,
    45
            }
    46
    0
        }
    47
    }
    48
    49
    /// A dynamic [`PhysicalExpr`] that can be updated by anyone with a reference to it.
    50
    ///
    51
    /// Any `ExecutionPlan` that uses this expression and holds a reference to it internally should probably also
    52
    /// implement `ExecutionPlan::reset_state` to remain compatible with recursive queries and other situations where
    53
    /// the same `ExecutionPlan` is reused with different data.
    54
    #[derive(Debug)]
    55
    pub struct DynamicFilterPhysicalExpr {
    56
        /// The original children of this PhysicalExpr, if any.
    57
        /// This is necessary because the dynamic filter may be initialized with a placeholder (e.g. `lit(true)`)
    58
        /// and later remapped to the actual expressions that are being filtered.
    59
        /// But we need to know the children (e.g. columns referenced in the expression) ahead of time to evaluate the expression correctly.
    60
        children: Vec<Arc<dyn PhysicalExpr>>,
    61
        /// If any of the children were remapped / modified (e.g. to adjust for projections) we need to keep track of the new children
    62
        /// so that when we update `current()` in subsequent iterations we can re-apply the replacements.
    63
        remapped_children: Option<Vec<Arc<dyn PhysicalExpr>>>,
    64
        /// The source of dynamic filters.
    65
        inner: Arc<RwLock<Inner>>,
    66
        /// Broadcasts filter state (updates and completion) to all waiters.
    67
        state_watch: watch::Sender<FilterState>,
    68
        /// For testing purposes track the data type and nullability to make sure they don't change.
    69
        /// If they do, there's a bug in the implementation.
    70
        /// But this can have overhead in production, so it's only included in our tests.
    71
        data_type: Arc<RwLock<Option<DataType>>>,
    72
        nullable: Arc<RwLock<Option<bool>>>,
    73
    }
    74
    75
    #[derive(Debug)]
    76
    struct Inner {
    77
        /// A counter that gets incremented every time the expression is updated so that we can track changes cheaply.
    78
        /// This is used for [`PhysicalExpr::snapshot_generation`] to have a cheap check for changes.
    79
        generation: u64,
    80
        expr: Arc<dyn PhysicalExpr>,
    81
        /// Flag for quick synchronous check if filter is complete.
    82
        /// This is redundant with the watch channel state, but allows us to return immediately
    83
        /// from `wait_complete()` without subscribing if already complete.
    84
        is_complete: bool,
    85
    }
    86
    87
    impl Inner {
    88
    0
        fn new(expr: Arc<dyn PhysicalExpr>) -> Self {
    89
    0
            Self {
    90
    0
                // Start with generation 1 which gives us a different result for [`PhysicalExpr::generation`] than the default 0.
    91
    0
                // This is not currently used anywhere but it seems useful to have this simple distinction.
    92
    0
                generation: 1,
    93
    0
                expr,
    94
    0
                is_complete: false,
    95
    0
            }
    96
    0
        }
    97
    98
        /// Clone the inner expression.
    99
    0
        fn expr(&self) -> &Arc<dyn PhysicalExpr> {
    100
    0
            &self.expr
    101
    0
        }
    102
    }
    103
    104
    impl Hash for DynamicFilterPhysicalExpr {
    105
    0
        fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
    106
    0
            let inner = self.current().expect("Failed to get current expression");
    107
    0
            inner.dyn_hash(state);
    108
    0
            self.children.dyn_hash(state);
    109
    0
            self.remapped_children.dyn_hash(state);
    110
    0
        }
    111
    }
    112
    113
    impl PartialEq for DynamicFilterPhysicalExpr {
    114
    0
        fn eq(&self, other: &Self) -> bool {
    115
    0
            let inner = self.current().expect("Failed to get current expression");
    116
    0
            let our_children = self.remapped_children.as_ref().unwrap_or(&self.children);
    117
    0
            let other_children = other.remapped_children.as_ref().unwrap_or(&other.children);
    118
    0
            let other = other.current().expect("Failed to get current expression");
    119
    0
            inner.dyn_eq(other.as_any()) && our_children == other_children
    120
    0
        }
    121
    }
    122
    123
    impl Eq for DynamicFilterPhysicalExpr {}
    124
    125
    impl Display for DynamicFilterPhysicalExpr {
    126
    0
        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
    127
    0
            self.render(f, |expr, f| write!(f, "{expr}"))
    128
    0
        }
    129
    }
    130
    131
    impl DynamicFilterPhysicalExpr {
    132
        /// Create a new [`DynamicFilterPhysicalExpr`]
    133
        /// from an initial expression and a list of children.
    134
        /// The list of children is provided separately because
    135
        /// the initial expression may not have the same children.
    136
        /// For example, if the initial expression is just `true`
    137
        /// it will not reference any columns, but we may know that
    138
        /// we are going to replace this expression with a real one
    139
        /// that does reference certain columns.
    140
        /// In this case you **must** pass in the columns that will be
    141
        /// used in the final expression as children to this function
    142
        /// since DataFusion is generally not compatible with dynamic
    143
        /// *children* in expressions.
    144
        ///
    145
        /// To determine the children you can:
    146
        ///
    147
        /// - Use [`collect_columns`] to collect the columns from the expression.
    148
        /// - Use existing information, such as the sort columns in a `SortExec`.
    149
        ///
    150
        /// Generally the important bit is that the *leaf children that reference columns
    151
        /// do not change* since those will be used to determine what columns need to read or projected
    152
        /// when evaluating the expression.
    153
        ///
    154
        /// Any `ExecutionPlan` that uses this expression and holds a reference to it internally should probably also
    155
        /// implement `ExecutionPlan::reset_state` to remain compatible with recursive queries and other situations where
    156
        /// the same `ExecutionPlan` is reused with different data.
    157
        ///
    158
        /// [`collect_columns`]: crate::utils::collect_columns
    159
    0
        pub fn new(
    160
    0
            children: Vec<Arc<dyn PhysicalExpr>>,
    161
    0
            inner: Arc<dyn PhysicalExpr>,
    162
    0
        ) -> Self {
    163
    0
            let (state_watch, _) = watch::channel(FilterState::InProgress { generation: 1 });
    164
    0
            Self {
    165
    0
                children,
    166
    0
                remapped_children: None, // Initially no remapped children
    167
    0
                inner: Arc::new(RwLock::new(Inner::new(inner))),
    168
    0
                state_watch,
    169
    0
                data_type: Arc::new(RwLock::new(None)),
    170
    0
                nullable: Arc::new(RwLock::new(None)),
    171
    0
            }
    172
    0
        }
    173
    174
    0
        fn remap_children(
    175
    0
            children: &[Arc<dyn PhysicalExpr>],
    176
    0
            remapped_children: Option<&Vec<Arc<dyn PhysicalExpr>>>,
    177
    0
            expr: Arc<dyn PhysicalExpr>,
    178
    0
        ) -> Result<Arc<dyn PhysicalExpr>> {
    179
    0
            if let Some(remapped_children) = remapped_children {
    180
                // Remap the children to the new children
    181
                // of the expression.
    182
    0
                expr.transform_up(|child| {
    183
                    // Check if this is any of our original children
    184
    0
                    if let Some(pos) =
    185
    0
                        children.iter().position(|c| c.as_ref() == child.as_ref())
    186
                    {
    187
                        // If so, remap it to the current children
    188
                        // of the expression.
    189
    0
                        let new_child = Arc::clone(&remapped_children[pos]);
    190
    0
                        Ok(Transformed::yes(new_child))
    191
                    } else {
    192
                        // Otherwise, just return the expression
    193
    0
                        Ok(Transformed::no(child))
    194
                    }
    195
    0
                })
    196
    0
                .data()
    197
            } else {
    198
                // If we don't have any remapped children, just return the expression
    199
    0
                Ok(Arc::clone(&expr))
    200
            }
    201
    0
        }
    202
    203
        /// Get the current generation of the expression.
    204
    0
        fn current_generation(&self) -> u64 {
    205
    0
            self.inner.read().generation
    206
    0
        }
    207
    208
        /// Get the current expression.
    209
        /// This will return the current expression with any children
    210
        /// remapped to match calls to [`PhysicalExpr::with_new_children`].
    211
    0
        pub fn current(&self) -> Result<Arc<dyn PhysicalExpr>> {
    212
    0
            let expr = Arc::clone(self.inner.read().expr());
    213
    0
            Self::remap_children(&self.children, self.remapped_children.as_ref(), expr)
    214
    0
        }
    215
    216
        /// Update the current expression and notify all waiters.
    217
        /// Any children of this expression must be a subset of the original children
    218
        /// passed to the constructor.
    219
        /// This should be called e.g.:
    220
        /// - When we've computed the probe side's hash table in a HashJoinExec
    221
        /// - After every batch is processed if we update the TopK heap in a SortExec using a TopK approach.
    222
    0
        pub fn update(&self, new_expr: Arc<dyn PhysicalExpr>) -> Result<()> {
    223
            // Remap the children of the new expression to match the original children
    224
            // We still do this again in `current()` but doing it preventively here
    225
            // reduces the work needed in some cases if `current()` is called multiple times
    226
            // and the same externally facing `PhysicalExpr` is used for both `with_new_children` and `update()`.`
    227
    0
            let new_expr = Self::remap_children(
    228
    0
                &self.children,
    229
    0
                self.remapped_children.as_ref(),
    230
    0
                new_expr,
    231
    0
            )?;
    232
    233
            // Load the current inner, increment generation, and store the new one
    234
    0
            let mut current = self.inner.write();
    235
    0
            let new_generation = current.generation + 1;
    236
    0
            *current = Inner {
    237
    0
                generation: new_generation,
    238
    0
                expr: new_expr,
    239
    0
                is_complete: current.is_complete,
    240
    0
            };
    241
    0
            drop(current); // Release the lock before broadcasting
    242
    243
            // Broadcast the new state to all waiters
    244
    0
            let _ = self.state_watch.send(FilterState::InProgress {
    245
    0
                generation: new_generation,
    246
    0
            });
    247
    0
            Ok(())
    248
    0
        }
    249
    250
        /// Mark this dynamic filter as complete and broadcast to all waiters.
    251
        ///
    252
        /// This signals that all expected updates have been received.
    253
        /// Waiters using [`Self::wait_complete`] will be notified.
    254
    0
        pub fn mark_complete(&self) {
    255
    0
            let mut current = self.inner.write();
    256
    0
            let current_generation = current.generation;
    257
    0
            current.is_complete = true;
    258
    0
            drop(current);
    259
    260
            // Broadcast completion to all waiters
    261
    0
            let _ = self.state_watch.send(FilterState::Complete {
    262
    0
                generation: current_generation,
    263
    0
            });
    264
    0
        }
    265
    266
        /// Wait asynchronously for any update to this filter.
    267
        ///
    268
        /// This method will return when [`Self::update`] is called and the generation increases.
    269
        /// It does not guarantee that the filter is complete.
    270
    0
        pub async fn wait_update(&self) {
    271
    0
            let mut rx = self.state_watch.subscribe();
    272
            // Get the current generation
    273
    0
            let current_gen = rx.borrow_and_update().generation();
    274
    275
            // Wait until generation increases
    276
    0
            let _ = rx.wait_for(|state| state.generation() > current_gen).await;
    277
    0
        }
    278
    279
        /// Wait asynchronously until this dynamic filter is marked as complete.
    280
        ///
    281
        /// This method returns immediately if the filter is already complete.
    282
        /// Otherwise, it waits until [`Self::mark_complete`] is called.
    283
        ///
    284
        /// Unlike [`Self::wait_update`], this method guarantees that when it returns,
    285
        /// the filter is fully complete with no more updates expected.
    286
    0
        pub async fn wait_complete(&self) {
    287
    0
            if self.inner.read().is_complete {
    288
    0
                return;
    289
    0
            }
    290
    291
    0
            let mut rx = self.state_watch.subscribe();
    292
    0
            let _ = rx
    293
    0
                .wait_for(|state| matches!(state, FilterState::Complete { .. }))
    294
    0
                .await;
    295
    0
        }
    296
    297
    0
        fn render(
    298
    0
            &self,
    299
    0
            f: &mut std::fmt::Formatter<'_>,
    300
    0
            render_expr: impl FnOnce(
    301
    0
                Arc<dyn PhysicalExpr>,
    302
    0
                &mut std::fmt::Formatter<'_>,
    303
    0
            ) -> std::fmt::Result,
    304
    0
        ) -> std::fmt::Result {
    305
    0
            let inner = self.current().map_err(|_| std::fmt::Error)?;
    306
    0
            let current_generation = self.current_generation();
    307
    0
            write!(f, "DynamicFilter [ ")?;
    308
    0
            if current_generation == 1 {
    309
    0
                write!(f, "empty")?;
    310
            } else {
    311
    0
                render_expr(inner, f)?;
    312
            }
    313
    314
    0
            write!(f, " ]")
    315
    0
        }
    316
    }
    317
    318
    impl PhysicalExpr for DynamicFilterPhysicalExpr {
    319
    0
        fn as_any(&self) -> &dyn Any {
    320
    0
            self
    321
    0
        }
    322
    323
    0
        fn children(&self) -> Vec<&Arc<dyn PhysicalExpr>> {
    324
    0
            self.remapped_children
    325
    0
                .as_ref()
    326
    0
                .unwrap_or(&self.children)
    327
    0
                .iter()
    328
    0
                .collect()
    329
    0
        }
    330
    331
    0
        fn with_new_children(
    332
    0
            self: Arc<Self>,
    333
    0
            children: Vec<Arc<dyn PhysicalExpr>>,
    334
    0
        ) -> Result<Arc<dyn PhysicalExpr>> {
    335
    0
            Ok(Arc::new(Self {
    336
    0
                children: self.children.clone(),
    337
    0
                remapped_children: Some(children),
    338
    0
                inner: Arc::clone(&self.inner),
    339
    0
                state_watch: self.state_watch.clone(),
    340
    0
                data_type: Arc::clone(&self.data_type),
    341
    0
                nullable: Arc::clone(&self.nullable),
    342
    0
            }))
    343
    0
        }
    344
    345
    0
        fn data_type(&self, input_schema: &Schema) -> Result<DataType> {
    346
    0
            let res = self.current()?.data_type(input_schema)?;
    347
            #[cfg(test)]
    348
            {
    349
                use datafusion_common::internal_err;
    350
                // Check if the data type has changed.
    351
    0
                let mut data_type_lock = self.data_type.write();
    352
    353
    0
                if let Some(existing) = &*data_type_lock {
    354
    0
                    if existing != &res {
    355
                        // If the data type has changed, we have a bug.
    356
    0
                        return internal_err!(
    357
    0
                            "DynamicFilterPhysicalExpr data type has changed unexpectedly. \
    358
    0
                            Expected: {existing:?}, Actual: {res:?}"
    359
                        );
    360
    0
                    }
    361
    0
                } else {
    362
    0
                    *data_type_lock = Some(res.clone());
    363
    0
                }
    364
            }
    365
    0
            Ok(res)
    366
    0
        }
    367
    368
    0
        fn nullable(&self, input_schema: &Schema) -> Result<bool> {
    369
    0
            let res = self.current()?.nullable(input_schema)?;
    370
            #[cfg(test)]
    371
            {
    372
                use datafusion_common::internal_err;
    373
                // Check if the nullability has changed.
    374
    0
                let mut nullable_lock = self.nullable.write();
    375
    0
                if let Some(existing) = *nullable_lock {
    376
    0
                    if existing != res {
    377
                        // If the nullability has changed, we have a bug.
    378
    0
                        return internal_err!(
    379
    0
                            "DynamicFilterPhysicalExpr nullability has changed unexpectedly. \
    380
    0
                            Expected: {existing}, Actual: {res}"
    381
                        );
    382
    0
                    }
    383
    0
                } else {
    384
    0
                    *nullable_lock = Some(res);
    385
    0
                }
    386
            }
    387
    0
            Ok(res)
    388
    0
        }
    389
    390
    0
        fn evaluate(
    391
    0
            &self,
    392
    0
            batch: &arrow::record_batch::RecordBatch,
    393
    0
        ) -> Result<ColumnarValue> {
    394
    0
            let current = self.current()?;
    395
            #[cfg(test)]
    396
            {
    397
                // Ensure that we are not evaluating after the expression has changed.
    398
    0
                let schema = batch.schema();
    399
    0
                self.nullable(&schema)?;
    400
    0
                self.data_type(&schema)?;
    401
            };
    402
    0
            current.evaluate(batch)
    403
    0
        }
    404
    405
    0
        fn fmt_sql(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
    406
    0
            self.render(f, |expr, f| expr.fmt_sql(f))
    407
    0
        }
    408
    409
    0
        fn snapshot(&self) -> Result<Option<Arc<dyn PhysicalExpr>>> {
    410
            // Return the current expression as a snapshot.
    411
    0
            Ok(Some(self.current()?))
    412
    0
        }
    413
    414
    0
        fn snapshot_generation(&self) -> u64 {
    415
            // Return the current generation of the expression.
    416
    0
            self.inner.read().generation
    417
    0
        }
    418
    }
    419
    420
    #[cfg(test)]
    421
    mod test {
    422
        use crate::{
    423
            expressions::{col, lit, BinaryExpr},
    424
            utils::reassign_expr_columns,
    425
        };
    426
        use arrow::{
    427
            array::RecordBatch,
    428
            datatypes::{DataType, Field, Schema},
    429
        };
    430
        use datafusion_common::ScalarValue;
    431
    432
        use super::*;
    433
    434
        #[test]
    435
    0
        fn test_remap_children() {
    436
    0
            let table_schema = Arc::new(Schema::new(vec![
    437
    0
                Field::new("a", DataType::Int32, false),
    438
    0
                Field::new("b", DataType::Int32, false),
    439
            ]));
    440
    0
            let expr = Arc::new(BinaryExpr::new(
    441
    0
                col("a", &table_schema).unwrap(),
    442
    0
                datafusion_expr::Operator::Eq,
    443
    0
                lit(42) as Arc<dyn PhysicalExpr>,
    444
            ));
    445
    0
            let dynamic_filter = Arc::new(DynamicFilterPhysicalExpr::new(
    446
    0
                vec![col("a", &table_schema).unwrap()],
    447
    0
                expr as Arc<dyn PhysicalExpr>,
    448
            ));
    449
            // Simulate two `ParquetSource` files with different filter schemas
    450
            // Both of these should hit the same inner `PhysicalExpr` even after `update()` is called
    451
            // and be able to remap children independently.
    452
    0
            let filter_schema_1 = Arc::new(Schema::new(vec![
    453
    0
                Field::new("a", DataType::Int32, false),
    454
    0
                Field::new("b", DataType::Int32, false),
    455
            ]));
    456
    0
            let filter_schema_2 = Arc::new(Schema::new(vec![
    457
    0
                Field::new("b", DataType::Int32, false),
    458
    0
                Field::new("a", DataType::Int32, false),
    459
            ]));
    460
            // Each ParquetExec calls `with_new_children` on the DynamicFilterPhysicalExpr
    461
            // and remaps the children to the file schema.
    462
    0
            let dynamic_filter_1 = reassign_expr_columns(
    463
    0
                Arc::clone(&dynamic_filter) as Arc<dyn PhysicalExpr>,
    464
    0
                &filter_schema_1,
    465
            )
    466
    0
            .unwrap();
    467
    0
            let snap = dynamic_filter_1.snapshot().unwrap().unwrap();
    468
    0
            insta::assert_snapshot!(format!("{snap:?}"), @r#"BinaryExpr { left: Column { name: "a", index: 0 }, op: Eq, right: Literal { value: Int32(42), field: Field { name: "lit", data_type: Int32 } }, fail_on_overflow: false }"#);
    469
    0
            let dynamic_filter_2 = reassign_expr_columns(
    470
    0
                Arc::clone(&dynamic_filter) as Arc<dyn PhysicalExpr>,
    471
    0
                &filter_schema_2,
    472
            )
    473
    0
            .unwrap();
    474
    0
            let snap = dynamic_filter_2.snapshot().unwrap().unwrap();
    475
    0
            insta::assert_snapshot!(format!("{snap:?}"), @r#"BinaryExpr { left: Column { name: "a", index: 1 }, op: Eq, right: Literal { value: Int32(42), field: Field { name: "lit", data_type: Int32 } }, fail_on_overflow: false }"#);
    476
            // Both filters allow evaluating the same expression
    477
    0
            let batch_1 = RecordBatch::try_new(
    478
    0
                Arc::clone(&filter_schema_1),
    479
    0
                vec![
    480
                    // a
    481
    0
                    ScalarValue::Int32(Some(42)).to_array_of_size(1).unwrap(),
    482
                    // b
    483
    0
                    ScalarValue::Int32(Some(43)).to_array_of_size(1).unwrap(),
    484
                ],
    485
            )
    486
    0
            .unwrap();
    487
    0
            let batch_2 = RecordBatch::try_new(
    488
    0
                Arc::clone(&filter_schema_2),
    489
    0
                vec![
    490
                    // b
    491
    0
                    ScalarValue::Int32(Some(43)).to_array_of_size(1).unwrap(),
    492
                    // a
    493
    0
                    ScalarValue::Int32(Some(42)).to_array_of_size(1).unwrap(),
    494
                ],
    495
            )
    496
    0
            .unwrap();
    497
            // Evaluate the expression on both batches
    498
    0
            let result_1 = dynamic_filter_1.evaluate(&batch_1).unwrap();
    499
    0
            let result_2 = dynamic_filter_2.evaluate(&batch_2).unwrap();
    500
            // Check that the results are the same
    501
    0
            let ColumnarValue::Array(arr_1) = result_1 else {
    502
    0
                panic!("Expected ColumnarValue::Array");
    503
            };
    504
    0
            let ColumnarValue::Array(arr_2) = result_2 else {
    505
    0
                panic!("Expected ColumnarValue::Array");
    506
            };
    507
    0
            assert!(arr_1.eq(&arr_2));
    508
    0
            let expected = ScalarValue::Boolean(Some(true))
    509
    0
                .to_array_of_size(1)
    510
    0
                .unwrap();
    511
    0
            assert!(arr_1.eq(&expected));
    512
            // Now lets update the expression
    513
            // Note that we update the *original* expression and that should be reflected in both the derived expressions
    514
    0
            let new_expr = Arc::new(BinaryExpr::new(
    515
    0
                col("a", &table_schema).unwrap(),
    516
    0
                datafusion_expr::Operator::Gt,
    517
    0
                lit(43) as Arc<dyn PhysicalExpr>,
    518
            ));
    519
    0
            dynamic_filter
    520
    0
                .update(Arc::clone(&new_expr) as Arc<dyn PhysicalExpr>)
    521
    0
                .expect("Failed to update expression");
    522
            // Now we should be able to evaluate the new expression on both batches
    523
    0
            let result_1 = dynamic_filter_1.evaluate(&batch_1).unwrap();
    524
    0
            let result_2 = dynamic_filter_2.evaluate(&batch_2).unwrap();
    525
            // Check that the results are the same
    526
    0
            let ColumnarValue::Array(arr_1) = result_1 else {
    527
    0
                panic!("Expected ColumnarValue::Array");
    528
            };
    529
    0
            let ColumnarValue::Array(arr_2) = result_2 else {
    530
    0
                panic!("Expected ColumnarValue::Array");
    531
            };
    532
    0
            assert!(arr_1.eq(&arr_2));
    533
    0
            let expected = ScalarValue::Boolean(Some(false))
    534
    0
                .to_array_of_size(1)
    535
    0
                .unwrap();
    536
    0
            assert!(arr_1.eq(&expected));
    537
    0
        }
    538
    539
        #[test]
    540
    0
        fn test_snapshot() {
    541
    0
            let expr = lit(42) as Arc<dyn PhysicalExpr>;
    542
    0
            let dynamic_filter = DynamicFilterPhysicalExpr::new(vec![], Arc::clone(&expr));
    543
    544
            // Take a snapshot of the current expression
    545
    0
            let snapshot = dynamic_filter.snapshot().unwrap();
    546
    0
            assert_eq!(snapshot, Some(expr));
    547
    548
            // Update the current expression
    549
    0
            let new_expr = lit(100) as Arc<dyn PhysicalExpr>;
    550
    0
            dynamic_filter.update(Arc::clone(&new_expr)).unwrap();
    551
            // Take another snapshot
    552
    0
            let snapshot = dynamic_filter.snapshot().unwrap();
    553
    0
            assert_eq!(snapshot, Some(new_expr));
    554
    0
        }
    555
    556
        #[test]
    557
    0
        fn test_dynamic_filter_physical_expr_misbehaves_data_type_nullable() {
    558
    0
            let dynamic_filter =
    559
    0
                DynamicFilterPhysicalExpr::new(vec![], lit(42) as Arc<dyn PhysicalExpr>);
    560
    561
            // First call to data_type and nullable should set the initial values.
    562
    0
            let initial_data_type = dynamic_filter.data_type(&Schema::empty()).unwrap();
    563
    0
            let initial_nullable = dynamic_filter.nullable(&Schema::empty()).unwrap();
    564
    565
            // Call again and expect no change.
    566
    0
            let second_data_type = dynamic_filter.data_type(&Schema::empty()).unwrap();
    567
    0
            let second_nullable = dynamic_filter.nullable(&Schema::empty()).unwrap();
    568
    0
            assert_eq!(
    569
                initial_data_type, second_data_type,
    570
    0
                "Data type should not change on second call."
    571
            );
    572
    0
            assert_eq!(
    573
                initial_nullable, second_nullable,
    574
    0
                "Nullability should not change on second call."
    575
            );
    576
    577
            // Now change the current expression to something else.
    578
    0
            dynamic_filter
    579
    0
                .update(lit(ScalarValue::Utf8(None)) as Arc<dyn PhysicalExpr>)
    580
    0
                .expect("Failed to update expression");
    581
            // Check that we error if we call data_type, nullable or evaluate after changing the expression.
    582
    0
            assert!(
    583
    0
                dynamic_filter.data_type(&Schema::empty()).is_err(),
    584
    0
                "Expected err when data_type is called after changing the expression."
    585
            );
    586
    0
            assert!(
    587
    0
                dynamic_filter.nullable(&Schema::empty()).is_err(),
    588
    0
                "Expected err when nullable is called after changing the expression."
    589
            );
    590
    0
            let batch = RecordBatch::new_empty(Arc::new(Schema::empty()));
    591
    0
            assert!(
    592
    0
                dynamic_filter.evaluate(&batch).is_err(),
    593
    0
                "Expected err when evaluate is called after changing the expression."
    594
            );
    595
    0
        }
    596
    597
        #[tokio::test]
    598
    0
        async fn test_wait_complete_already_complete() {
    599
    0
            let dynamic_filter = Arc::new(DynamicFilterPhysicalExpr::new(
    600
    0
                vec![],
    601
    0
                lit(42) as Arc<dyn PhysicalExpr>,
    602
            ));
    603
    604
            // Mark as complete immediately
    605
    0
            dynamic_filter.mark_complete();
    606
    607
            // wait_complete should return immediately
    608
    0
            dynamic_filter.wait_complete().await;
    609
    0
        }
    610
    }