## 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/intervals/cp_solver.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
    //! Constraint propagator/solver for custom [`PhysicalExpr`] graphs.
    19
    //!
    20
    //! The constraint propagator/solver in DataFusion uses interval arithmetic to
    21
    //! perform mathematical operations on intervals, which represent a range of
    22
    //! possible values rather than a single point value. This allows for the
    23
    //! propagation of ranges through mathematical operations, and can be used to
    24
    //! compute bounds for a complicated expression. The key idea is that by
    25
    //! breaking down a complicated expression into simpler terms, and then
    26
    //! combining the bounds for those simpler terms, one can obtain bounds for the
    27
    //! overall expression.
    28
    //!
    29
    //! This way of using interval arithmetic to compute bounds for a complex
    30
    //! expression by combining the bounds for the constituent terms within the
    31
    //! original expression allows us to reason about the range of possible values
    32
    //! of the expression. This information later can be used in range pruning of
    33
    //! the provably unnecessary parts of `RecordBatch`es.
    34
    //!
    35
    //! # Example
    36
    //!
    37
    //! For example, consider a mathematical expression such as `x^2 + y = 4` \[1\].
    38
    //! Since this expression would be a binary tree in [`PhysicalExpr`] notation,
    39
    //! this type of an hierarchical computation is well-suited for a graph based
    40
    //! implementation. In such an implementation, an equation system `f(x) = 0` is
    41
    //! represented by a directed acyclic expression graph (DAEG).
    42
    //!
    43
    //! In order to use interval arithmetic to compute bounds for this expression,
    44
    //! one would first determine intervals that represent the possible values of
    45
    //! `x` and `y` Let's say that the interval for `x` is `[1, 2]` and the interval
    46
    //! for `y` is `[-3, 1]`. In the chart below, you can see how the computation
    47
    //! takes place.
    48
    //!
    49
    //! # References
    50
    //!
    51
    //! 1. Kabak, Mehmet Ozan. Analog Circuit Start-Up Behavior Analysis: An Interval
    52
    //!    Arithmetic Based Approach, Chapter 4. Stanford University, 2015.
    53
    //! 2. Moore, Ramon E. Interval analysis. Vol. 4. Englewood Cliffs: Prentice-Hall, 1966.
    54
    //! 3. F. Messine, "Deterministic global optimization using interval constraint
    55
    //!    propagation techniques," RAIRO-Operations Research, vol. 38, no. 04,
    56
    //!    pp. 277-293, 2004.
    57
    //!
    58
    //! # Illustration
    59
    //!
    60
    //! ## Computing bounds for an expression using interval arithmetic
    61
    //!
    62
    //! ```text
    63
    //!             +-----+                         +-----+
    64
    //!        +----|  +  |----+               +----|  +  |----+
    65
    //!        |    |     |    |               |    |     |    |
    66
    //!        |    +-----+    |               |    +-----+    |
    67
    //!        |               |               |               |
    68
    //!    +-----+           +-----+       +-----+           +-----+
    69
    //!    |   2 |           |  y  |       |   2 | [1, 4]    |  y  |
    70
    //!    |[.]  |           |     |       |[.]  |           |     |
    71
    //!    +-----+           +-----+       +-----+           +-----+
    72
    //!       |                               |
    73
    //!       |                               |
    74
    //!     +---+                           +---+
    75
    //!     | x | [1, 2]                    | x | [1, 2]
    76
    //!     +---+                           +---+
    77
    //!
    78
    //!  (a) Bottom-up evaluation: Step 1 (b) Bottom up evaluation: Step 2
    79
    //!
    80
    //!                                      [1 - 3, 4 + 1] = [-2, 5]
    81
    //!             +-----+                         +-----+
    82
    //!        +----|  +  |----+               +----|  +  |----+
    83
    //!        |    |     |    |               |    |     |    |
    84
    //!        |    +-----+    |               |    +-----+    |
    85
    //!        |               |               |               |
    86
    //!    +-----+           +-----+       +-----+           +-----+
    87
    //!    |   2 |[1, 4]     |  y  |       |   2 |[1, 4]     |  y  |
    88
    //!    |[.]  |           |     |       |[.]  |           |     |
    89
    //!    +-----+           +-----+       +-----+           +-----+
    90
    //!       |              [-3, 1]          |              [-3, 1]
    91
    //!       |                               |
    92
    //!     +---+                           +---+
    93
    //!     | x | [1, 2]                    | x | [1, 2]
    94
    //!     +---+                           +---+
    95
    //!
    96
    //!  (c) Bottom-up evaluation: Step 3 (d) Bottom-up evaluation: Step 4
    97
    //! ```
    98
    //!
    99
    //! ## Top-down constraint propagation using inverse semantics
    100
    //!
    101
    //! ```text
    102
    //!    [-2, 5] ∩ [4, 4] = [4, 4]               [4, 4]
    103
    //!            +-----+                         +-----+
    104
    //!       +----|  +  |----+               +----|  +  |----+
    105
    //!       |    |     |    |               |    |     |    |
    106
    //!       |    +-----+    |               |    +-----+    |
    107
    //!       |               |               |               |
    108
    //!    +-----+           +-----+       +-----+           +-----+
    109
    //!    |   2 | [1, 4]    |  y  |       |   2 | [1, 4]    |  y  | [0, 1]*
    110
    //!    |[.]  |           |     |       |[.]  |           |     |
    111
    //!    +-----+           +-----+       +-----+           +-----+
    112
    //!      |              [-3, 1]          |
    113
    //!      |                               |
    114
    //!    +---+                           +---+
    115
    //!    | x | [1, 2]                    | x | [1, 2]
    116
    //!    +---+                           +---+
    117
    //!
    118
    //!  (a) Top-down propagation: Step 1 (b) Top-down propagation: Step 2
    119
    //!
    120
    //!                                     [1 - 3, 4 + 1] = [-2, 5]
    121
    //!            +-----+                         +-----+
    122
    //!       +----|  +  |----+               +----|  +  |----+
    123
    //!       |    |     |    |               |    |     |    |
    124
    //!       |    +-----+    |               |    +-----+    |
    125
    //!       |               |               |               |
    126
    //!    +-----+           +-----+       +-----+           +-----+
    127
    //!    |   2 |[3, 4]**   |  y  |       |   2 |[3, 4]     |  y  |
    128
    //!    |[.]  |           |     |       |[.]  |           |     |
    129
    //!    +-----+           +-----+       +-----+           +-----+
    130
    //!      |              [0, 1]           |              [-3, 1]
    131
    //!      |                               |
    132
    //!    +---+                           +---+
    133
    //!    | x | [1, 2]                    | x | [sqrt(3), 2]***
    134
    //!    +---+                           +---+
    135
    //!
    136
    //!  (c) Top-down propagation: Step 3  (d) Top-down propagation: Step 4
    137
    //!
    138
    //!    * [-3, 1] ∩ ([4, 4] - [1, 4]) = [0, 1]
    139
    //!    ** [1, 4] ∩ ([4, 4] - [0, 1]) = [3, 4]
    140
    //!    *** [1, 2] ∩ [sqrt(3), sqrt(4)] = [sqrt(3), 2]
    141
    //! ```
    142
    143
    use std::collections::HashSet;
    144
    use std::fmt::{Display, Formatter};
    145
    use std::mem::{size_of, size_of_val};
    146
    use std::sync::Arc;
    147
    148
    use super::utils::{
    149
        convert_duration_type_to_interval, convert_interval_type_to_duration, get_inverse_op,
    150
    };
    151
    use crate::expressions::{BinaryExpr, Literal};
    152
    use crate::utils::{build_dag, ExprTreeNode};
    153
    use crate::PhysicalExpr;
    154
    155
    use arrow::datatypes::{DataType, Schema};
    156
    use datafusion_common::{internal_err, not_impl_err, Result};
    157
    use datafusion_expr::interval_arithmetic::{apply_operator, satisfy_greater, Interval};
    158
    use datafusion_expr::Operator;
    159
    160
    use petgraph::graph::NodeIndex;
    161
    use petgraph::stable_graph::{DefaultIx, StableGraph};
    162
    use petgraph::visit::{Bfs, Dfs, DfsPostOrder, EdgeRef};
    163
    use petgraph::Outgoing;
    164
    165
    /// This object implements a directed acyclic expression graph (DAEG) that
    166
    /// is used to compute ranges for expressions through interval arithmetic.
    167
    #[derive(Clone, Debug)]
    168
    pub struct ExprIntervalGraph {
    169
        graph: StableGraph<ExprIntervalGraphNode, usize>,
    170
        root: NodeIndex,
    171
    }
    172
    173
    /// This object encapsulates all possible constraint propagation results.
    174
    #[derive(PartialEq, Debug)]
    175
    pub enum PropagationResult {
    176
        CannotPropagate,
    177
        Infeasible,
    178
        Success,
    179
    }
    180
    181
    /// This is a node in the DAEG; it encapsulates a reference to the actual
    182
    /// [`PhysicalExpr`] as well as an interval containing expression bounds.
    183
    #[derive(Clone, Debug)]
    184
    pub struct ExprIntervalGraphNode {
    185
        expr: Arc<dyn PhysicalExpr>,
    186
        interval: Interval,
    187
    }
    188
    189
    impl PartialEq for ExprIntervalGraphNode {
    190
    0
        fn eq(&self, other: &Self) -> bool {
    191
    0
            self.expr.eq(&other.expr)
    192
    0
        }
    193
    }
    194
    195
    impl Display for ExprIntervalGraphNode {
    196
    0
        fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
    197
    0
            write!(f, "{}", self.expr)
    198
    0
        }
    199
    }
    200
    201
    impl ExprIntervalGraphNode {
    202
        /// Constructs a new DAEG node with an `[-∞, ∞]` range.
    203
    0
        pub fn new_unbounded(expr: Arc<dyn PhysicalExpr>, dt: &DataType) -> Result<Self> {
    204
    0
            Interval::make_unbounded(dt)
    205
    0
                .map(|interval| ExprIntervalGraphNode { expr, interval })
    206
    0
        }
    207
    208
        /// Constructs a new DAEG node with the given range.
    209
    0
        pub fn new_with_interval(expr: Arc<dyn PhysicalExpr>, interval: Interval) -> Self {
    210
    0
            ExprIntervalGraphNode { expr, interval }
    211
    0
        }
    212
    213
        /// Get the interval object representing the range of the expression.
    214
    0
        pub fn interval(&self) -> &Interval {
    215
    0
            &self.interval
    216
    0
        }
    217
    218
        /// This function creates a DAEG node from DataFusion's [`ExprTreeNode`]
    219
        /// object. Literals are created with definite, singleton intervals while
    220
        /// any other expression starts with an indefinite interval (`[-∞, ∞]`).
    221
    0
        pub fn make_node(node: &ExprTreeNode<NodeIndex>, schema: &Schema) -> Result<Self> {
    222
    0
            let expr = Arc::clone(&node.expr);
    223
    0
            if let Some(literal) = expr.as_any().downcast_ref::<Literal>() {
    224
    0
                let value = literal.value();
    225
    0
                Interval::try_new(value.clone(), value.clone())
    226
    0
                    .map(|interval| Self::new_with_interval(expr, interval))
    227
            } else {
    228
    0
                expr.data_type(schema)
    229
    0
                    .and_then(|dt| Self::new_unbounded(expr, &dt))
    230
            }
    231
    0
        }
    232
    }
    233
    234
    /// This function refines intervals `left_child` and `right_child` by applying
    235
    /// constraint propagation through `parent` via operation. The main idea is
    236
    /// that we can shrink ranges of variables x and y using parent interval p.
    237
    ///
    238
    /// Assuming that x,y and p has ranges `[xL, xU]`, `[yL, yU]`, and `[pL, pU]`, we
    239
    /// apply the following operations:
    240
    /// - For plus operation, specifically, we would first do
    241
    ///     - `[xL, xU]` <- (`[pL, pU]` - `[yL, yU]`) ∩ `[xL, xU]`, and then
    242
    ///     - `[yL, yU]` <- (`[pL, pU]` - `[xL, xU]`) ∩ `[yL, yU]`.
    243
    /// - For minus operation, specifically, we would first do
    244
    ///     - `[xL, xU]` <- (`[yL, yU]` + `[pL, pU]`) ∩ `[xL, xU]`, and then
    245
    ///     - `[yL, yU]` <- (`[xL, xU]` - `[pL, pU]`) ∩ `[yL, yU]`.
    246
    /// - For multiplication operation, specifically, we would first do
    247
    ///     - `[xL, xU]` <- (`[pL, pU]` / `[yL, yU]`) ∩ `[xL, xU]`, and then
    248
    ///     - `[yL, yU]` <- (`[pL, pU]` / `[xL, xU]`) ∩ `[yL, yU]`.
    249
    /// - For division operation, specifically, we would first do
    250
    ///     - `[xL, xU]` <- (`[yL, yU]` * `[pL, pU]`) ∩ `[xL, xU]`, and then
    251
    ///     - `[yL, yU]` <- (`[xL, xU]` / `[pL, pU]`) ∩ `[yL, yU]`.
    252
    0
    pub fn propagate_arithmetic(
    253
    0
        op: &Operator,
    254
    0
        parent: &Interval,
    255
    0
        left_child: &Interval,
    256
    0
        right_child: &Interval,
    257
    0
    ) -> Result<Option<(Interval, Interval)>> {
    258
    0
        let inverse_op = get_inverse_op(*op)?;
    259
    0
        match (left_child.data_type(), right_child.data_type()) {
    260
            // If we have a child whose type is a time interval (i.e. DataType::Interval),
    261
            // we need special handling since timestamp differencing results in a
    262
            // Duration type.
    263
            (DataType::Timestamp(..), DataType::Interval(_)) => {
    264
    0
                propagate_time_interval_at_right(
    265
    0
                    left_child,
    266
    0
                    right_child,
    267
    0
                    parent,
    268
    0
                    op,
    269
    0
                    &inverse_op,
    270
                )
    271
            }
    272
            (DataType::Interval(_), DataType::Timestamp(..)) => {
    273
    0
                propagate_time_interval_at_left(
    274
    0
                    left_child,
    275
    0
                    right_child,
    276
    0
                    parent,
    277
    0
                    op,
    278
    0
                    &inverse_op,
    279
                )
    280
            }
    281
            _ => {
    282
                // First, propagate to the left:
    283
    0
                match apply_operator(&inverse_op, parent, right_child)?
    284
    0
                    .intersect(left_child)?
    285
                {
    286
                    // Left is feasible:
    287
    0
                    Some(value) => Ok(
    288
                        // Propagate to the right using the new left.
    289
    0
                        propagate_right(&value, parent, right_child, op, &inverse_op)?
    290
    0
                            .map(|right| (value, right)),
    291
                    ),
    292
                    // If the left child is infeasible, short-circuit.
    293
    0
                    None => Ok(None),
    294
                }
    295
            }
    296
        }
    297
    0
    }
    298
    299
    /// This function refines intervals `left_child` and `right_child` by applying
    300
    /// comparison propagation through `parent` via operation. The main idea is
    301
    /// that we can shrink ranges of variables x and y using parent interval p.
    302
    /// Two intervals can be ordered in 6 ways for a Gt `>` operator:
    303
    /// ```text
    304
    ///                           (1): Infeasible, short-circuit
    305
    /// left:   |        ================                                               |
    306
    /// right:  |                           ========================                    |
    307
    ///
    308
    ///                             (2): Update both interval
    309
    /// left:   |              ======================                                   |
    310
    /// right:  |                             ======================                    |
    311
    ///                                          |
    312
    ///                                          V
    313
    /// left:   |                             =======                                   |
    314
    /// right:  |                             =======                                   |
    315
    ///
    316
    ///                             (3): Update left interval
    317
    /// left:   |                  ==============================                       |
    318
    /// right:  |                           ==========                                  |
    319
    ///                                          |
    320
    ///                                          V
    321
    /// left:   |                           =====================                       |
    322
    /// right:  |                           ==========                                  |
    323
    ///
    324
    ///                             (4): Update right interval
    325
    /// left:   |                           ==========                                  |
    326
    /// right:  |                   ===========================                         |
    327
    ///                                          |
    328
    ///                                          V
    329
    /// left:   |                           ==========                                  |
    330
    /// right   |                   ==================                                  |
    331
    ///
    332
    ///                                   (5): No change
    333
    /// left:   |                       ============================                    |
    334
    /// right:  |               ===================                                     |
    335
    ///
    336
    ///                                   (6): No change
    337
    /// left:   |                                    ====================               |
    338
    /// right:  |                ===============                                        |
    339
    ///
    340
    ///         -inf --------------------------------------------------------------- +inf
    341
    /// ```
    342
    0
    pub fn propagate_comparison(
    343
    0
        op: &Operator,
    344
    0
        parent: &Interval,
    345
    0
        left_child: &Interval,
    346
    0
        right_child: &Interval,
    347
    0
    ) -> Result<Option<(Interval, Interval)>> {
    348
    0
        if parent == &Interval::TRUE {
    349
    0
            match op {
    350
    0
                Operator::Eq => left_child.intersect(right_child).map(|result| {
    351
    0
                    result.map(|intersection| (intersection.clone(), intersection))
    352
    0
                }),
    353
    0
                Operator::Gt => satisfy_greater(left_child, right_child, true),
    354
    0
                Operator::GtEq => satisfy_greater(left_child, right_child, false),
    355
    0
                Operator::Lt => satisfy_greater(right_child, left_child, true)
    356
    0
                    .map(|t| t.map(reverse_tuple)),
    357
    0
                Operator::LtEq => satisfy_greater(right_child, left_child, false)
    358
    0
                    .map(|t| t.map(reverse_tuple)),
    359
    0
                _ => internal_err!(
    360
    0
                    "The operator must be a comparison operator to propagate intervals"
    361
                ),
    362
            }
    363
    0
        } else if parent == &Interval::FALSE {
    364
    0
            match op {
    365
                Operator::Eq => {
    366
                    // TODO: Propagation is not possible until we support interval sets.
    367
    0
                    Ok(None)
    368
                }
    369
    0
                Operator::Gt => satisfy_greater(right_child, left_child, false),
    370
    0
                Operator::GtEq => satisfy_greater(right_child, left_child, true),
    371
    0
                Operator::Lt => satisfy_greater(left_child, right_child, false)
    372
    0
                    .map(|t| t.map(reverse_tuple)),
    373
    0
                Operator::LtEq => satisfy_greater(left_child, right_child, true)
    374
    0
                    .map(|t| t.map(reverse_tuple)),
    375
    0
                _ => internal_err!(
    376
    0
                    "The operator must be a comparison operator to propagate intervals"
    377
                ),
    378
            }
    379
        } else {
    380
            // Uncertainty cannot change any end-point of the intervals.
    381
    0
            Ok(None)
    382
        }
    383
    0
    }
    384
    385
    impl ExprIntervalGraph {
    386
    0
        pub fn try_new(expr: Arc<dyn PhysicalExpr>, schema: &Schema) -> Result<Self> {
    387
            // Build the full graph:
    388
    0
            let (root, graph) =
    389
    0
                build_dag(expr, &|node| ExprIntervalGraphNode::make_node(node, schema))?;
    390
    0
            Ok(Self { graph, root })
    391
    0
        }
    392
    393
    0
        pub fn node_count(&self) -> usize {
    394
    0
            self.graph.node_count()
    395
    0
        }
    396
    397
        /// Estimate size of bytes including `Self`.
    398
    0
        pub fn size(&self) -> usize {
    399
    0
            let node_memory_usage = self.graph.node_count()
    400
    0
                * (size_of::<ExprIntervalGraphNode>() + size_of::<NodeIndex>());
    401
    0
            let edge_memory_usage =
    402
    0
                self.graph.edge_count() * (size_of::<usize>() + size_of::<NodeIndex>() * 2);
    403
    404
    0
            size_of_val(self) + node_memory_usage + edge_memory_usage
    405
    0
        }
    406
    407
        // Sometimes, we do not want to calculate and/or propagate intervals all
    408
        // way down to leaf expressions. For example, assume that we have a
    409
        // `SymmetricHashJoin` which has a child with an output ordering like:
    410
        //
    411
        // ```text
    412
        // PhysicalSortExpr {
    413
        //     expr: BinaryExpr('a', +, 'b'),
    414
        //     sort_option: ..
    415
        // }
    416
        // ```
    417
        //
    418
        // i.e. its output order comes from a clause like `ORDER BY a + b`. In such
    419
        // a case, we must calculate the interval for the `BinaryExpr(a, +, b)`
    420
        // instead of the columns inside this `BinaryExpr`, because this interval
    421
        // decides whether we prune or not. Therefore, children `PhysicalExpr`s of
    422
        // this `BinaryExpr` may be pruned for performance. The figure below
    423
        // explains this example visually.
    424
        //
    425
        // Note that we just remove the nodes from the DAEG, do not make any change
    426
        // to the plan itself.
    427
        //
    428
        // ```text
    429
        //
    430
        //                                  +-----+                                          +-----+
    431
        //                                  | GT  |                                          | GT  |
    432
        //                         +--------|     |-------+                         +--------|     |-------+
    433
        //                         |        +-----+       |                         |        +-----+       |
    434
        //                         |                      |                         |                      |
    435
        //                      +-----+                   |                      +-----+                   |
    436
        //                      |Cast |                   |                      |Cast |                   |
    437
        //                      |     |                   |             --\      |     |                   |
    438
        //                      +-----+                   |       ----------     +-----+                   |
    439
        //                         |                      |             --/         |                      |
    440
        //                         |                      |                         |                      |
    441
        //                      +-----+                +-----+                   +-----+                +-----+
    442
        //                   +--|Plus |--+          +--|Plus |--+                |Plus |             +--|Plus |--+
    443
        //                   |  |     |  |          |  |     |  |                |     |             |  |     |  |
    444
        //  Prune from here  |  +-----+  |          |  +-----+  |                +-----+             |  +-----+  |
    445
        //  ------------------------------------    |           |                                    |           |
    446
        //                   |           |          |           |                                    |           |
    447
        //                +-----+     +-----+    +-----+     +-----+                              +-----+     +-----+
    448
        //                | a   |     |  b  |    |  c  |     |  2  |                              |  c  |     |  2  |
    449
        //                |     |     |     |    |     |     |     |                              |     |     |     |
    450
        //                +-----+     +-----+    +-----+     +-----+                              +-----+     +-----+
    451
        //
    452
        // ```
    453
    454
        /// This function associates stable node indices with [`PhysicalExpr`]s so
    455
        /// that we can match `Arc<dyn PhysicalExpr>` and NodeIndex objects during
    456
        /// membership tests.
    457
    0
        pub fn gather_node_indices(
    458
    0
            &mut self,
    459
    0
            exprs: &[Arc<dyn PhysicalExpr>],
    460
    0
        ) -> Vec<(Arc<dyn PhysicalExpr>, usize)> {
    461
    0
            let graph = &self.graph;
    462
    0
            let mut bfs = Bfs::new(graph, self.root);
    463
            // We collect the node indices (usize) of [PhysicalExpr]s in the order
    464
            // given by argument `exprs`. To preserve this order, we initialize each
    465
            // expression's node index with usize::MAX, and then find the corresponding
    466
            // node indices by traversing the graph.
    467
    0
            let mut removals = vec![];
    468
    0
            let mut expr_node_indices = exprs
    469
    0
                .iter()
    470
    0
                .map(|e| (Arc::clone(e), usize::MAX))
    471
    0
                .collect::<Vec<_>>();
    472
    0
            while let Some(node) = bfs.next(graph) {
    473
                // Get the plan corresponding to this node:
    474
    0
                let expr = &graph[node].expr;
    475
                // If the current expression is among `exprs`, slate its children
    476
                // for removal:
    477
    0
                if let Some(value) = exprs.iter().position(|e| expr.eq(e)) {
    478
                    // Update the node index of the associated `PhysicalExpr`:
    479
    0
                    expr_node_indices[value].1 = node.index();
    480
    0
                    for edge in graph.edges_directed(node, Outgoing) {
    481
    0
                        // Slate the child for removal, do not remove immediately.
    482
    0
                        removals.push(edge.id());
    483
    0
                    }
    484
    0
                }
    485
            }
    486
    0
            for edge_idx in removals {
    487
    0
                self.graph.remove_edge(edge_idx);
    488
    0
            }
    489
            // Get the set of node indices reachable from the root node:
    490
    0
            let connected_nodes = self.connected_nodes();
    491
            // Remove nodes not connected to the root node:
    492
    0
            self.graph
    493
    0
                .retain_nodes(|_, index| connected_nodes.contains(&index));
    494
    0
            expr_node_indices
    495
    0
        }
    496
    497
        /// Returns the set of node indices reachable from the root node via a
    498
        /// simple depth-first search.
    499
    0
        fn connected_nodes(&self) -> HashSet<NodeIndex> {
    500
    0
            let mut nodes = HashSet::new();
    501
    0
            let mut dfs = Dfs::new(&self.graph, self.root);
    502
    0
            while let Some(node) = dfs.next(&self.graph) {
    503
    0
                nodes.insert(node);
    504
    0
            }
    505
    0
            nodes
    506
    0
        }
    507
    508
        /// Updates intervals for all expressions in the DAEG by successive
    509
        /// bottom-up and top-down traversals.
    510
    0
        pub fn update_ranges(
    511
    0
            &mut self,
    512
    0
            leaf_bounds: &mut [(usize, Interval)],
    513
    0
            given_range: Interval,
    514
    0
        ) -> Result<PropagationResult> {
    515
    0
            self.assign_intervals(leaf_bounds);
    516
    0
            let bounds = self.evaluate_bounds()?;
    517
            // There are three possible cases to consider:
    518
            // (1) given_range ⊇ bounds => Nothing to propagate
    519
            // (2) ∅ ⊂ (given_range ∩ bounds) ⊂ bounds => Can propagate
    520
            // (3) Disjoint sets => Infeasible
    521
    0
            if given_range.contains(bounds)? == Interval::TRUE {
    522
                // First case:
    523
    0
                Ok(PropagationResult::CannotPropagate)
    524
    0
            } else if bounds.contains(&given_range)? != Interval::FALSE {
    525
                // Second case:
    526
    0
                let result = self.propagate_constraints(given_range);
    527
    0
                self.update_intervals(leaf_bounds);
    528
    0
                result
    529
            } else {
    530
                // Third case:
    531
    0
                Ok(PropagationResult::Infeasible)
    532
            }
    533
    0
        }
    534
    535
        /// This function assigns given ranges to expressions in the DAEG.
    536
        /// The argument `assignments` associates indices of sought expressions
    537
        /// with their corresponding new ranges.
    538
    0
        pub fn assign_intervals(&mut self, assignments: &[(usize, Interval)]) {
    539
    0
            for (index, interval) in assignments {
    540
    0
                let node_index = NodeIndex::from(*index as DefaultIx);
    541
    0
                self.graph[node_index].interval = interval.clone();
    542
    0
            }
    543
    0
        }
    544
    545
        /// This function fetches ranges of expressions from the DAEG. The argument
    546
        /// `assignments` associates indices of sought expressions with their ranges,
    547
        /// which this function modifies to reflect the intervals in the DAEG.
    548
    0
        pub fn update_intervals(&self, assignments: &mut [(usize, Interval)]) {
    549
    0
            for (index, interval) in assignments.iter_mut() {
    550
    0
                let node_index = NodeIndex::from(*index as DefaultIx);
    551
    0
                *interval = self.graph[node_index].interval.clone();
    552
    0
            }
    553
    0
        }
    554
    555
        /// Computes bounds for an expression using interval arithmetic via a
    556
        /// bottom-up traversal.
    557
        ///
    558
        /// # Examples
    559
        ///
    560
        /// ```
    561
        /// use arrow::datatypes::DataType;
    562
        /// use arrow::datatypes::Field;
    563
        /// use arrow::datatypes::Schema;
    564
        /// use datafusion_common::ScalarValue;
    565
        /// use datafusion_expr::interval_arithmetic::Interval;
    566
        /// use datafusion_expr::Operator;
    567
        /// use datafusion_physical_expr::expressions::{BinaryExpr, Column, Literal};
    568
        /// use datafusion_physical_expr::intervals::cp_solver::ExprIntervalGraph;
    569
        /// use datafusion_physical_expr::PhysicalExpr;
    570
        /// use std::sync::Arc;
    571
        ///
    572
        /// let expr = Arc::new(BinaryExpr::new(
    573
        ///     Arc::new(Column::new("gnz", 0)),
    574
        ///     Operator::Plus,
    575
        ///     Arc::new(Literal::new(ScalarValue::Int32(Some(10)))),
    576
        /// ));
    577
        ///
    578
        /// let schema = Schema::new(vec![Field::new("gnz".to_string(), DataType::Int32, true)]);
    579
        ///
    580
        /// let mut graph = ExprIntervalGraph::try_new(expr, &schema).unwrap();
    581
        /// // Do it once, while constructing.
    582
        /// let node_indices = graph.gather_node_indices(&[Arc::new(Column::new("gnz", 0))]);
    583
        /// let left_index = node_indices.get(0).unwrap().1;
    584
        ///
    585
        /// // Provide intervals for leaf variables (here, there is only one).
    586
        /// let intervals = vec![(left_index, Interval::make(Some(10), Some(20)).unwrap())];
    587
        ///
    588
        /// // Evaluate bounds for the composite expression:
    589
        /// graph.assign_intervals(&intervals);
    590
        /// assert_eq!(
    591
        ///     graph.evaluate_bounds().unwrap(),
    592
        ///     &Interval::make(Some(20), Some(30)).unwrap(),
    593
        /// )
    594
        /// ```
    595
    0
        pub fn evaluate_bounds(&mut self) -> Result<&Interval> {
    596
    0
            let mut dfs = DfsPostOrder::new(&self.graph, self.root);
    597
    0
            while let Some(node) = dfs.next(&self.graph) {
    598
    0
                let neighbors = self.graph.neighbors_directed(node, Outgoing);
    599
    0
                let mut children_intervals = neighbors
    600
    0
                    .map(|child| self.graph[child].interval())
    601
    0
                    .collect::<Vec<_>>();
    602
                // If the current expression is a leaf, its interval should already
    603
                // be set externally, just continue with the evaluation procedure:
    604
    0
                if !children_intervals.is_empty() {
    605
                    // Reverse to align with `PhysicalExpr`'s children:
    606
    0
                    children_intervals.reverse();
    607
    0
                    self.graph[node].interval =
    608
    0
                        self.graph[node].expr.evaluate_bounds(&children_intervals)?;
    609
    0
                }
    610
            }
    611
    0
            Ok(self.graph[self.root].interval())
    612
    0
        }
    613
    614
        /// Updates/shrinks bounds for leaf expressions using interval arithmetic
    615
        /// via a top-down traversal.
    616
    0
        fn propagate_constraints(
    617
    0
            &mut self,
    618
    0
            given_range: Interval,
    619
    0
        ) -> Result<PropagationResult> {
    620
            // Adjust the root node with the given range:
    621
    0
            if let Some(interval) = self.graph[self.root].interval.intersect(given_range)? {
    622
    0
                self.graph[self.root].interval = interval;
    623
    0
            } else {
    624
    0
                return Ok(PropagationResult::Infeasible);
    625
            }
    626
    627
    0
            let mut bfs = Bfs::new(&self.graph, self.root);
    628
    629
    0
            while let Some(node) = bfs.next(&self.graph) {
    630
    0
                let neighbors = self.graph.neighbors_directed(node, Outgoing);
    631
    0
                let mut children = neighbors.collect::<Vec<_>>();
    632
                // If the current expression is a leaf, its range is now final.
    633
                // So, just continue with the propagation procedure:
    634
    0
                if children.is_empty() {
    635
    0
                    continue;
    636
    0
                }
    637
                // Reverse to align with `PhysicalExpr`'s children:
    638
    0
                children.reverse();
    639
    0
                let children_intervals = children
    640
    0
                    .iter()
    641
    0
                    .map(|child| self.graph[*child].interval())
    642
    0
                    .collect::<Vec<_>>();
    643
    0
                let node_interval = self.graph[node].interval();
    644
                // Special case: true OR could in principle be propagated by 3 interval sets,
    645
                // (i.e. left true, or right true, or both true) however we do not support this yet.
    646
    0
                if node_interval == &Interval::TRUE
    647
    0
                    && self.graph[node]
    648
    0
                        .expr
    649
    0
                        .as_any()
    650
    0
                        .downcast_ref::<BinaryExpr>()
    651
    0
                        .is_some_and(|expr| expr.op() == &Operator::Or)
    652
                {
    653
    0
                    return not_impl_err!("OR operator cannot yet propagate true intervals");
    654
    0
                }
    655
    0
                let propagated_intervals = self.graph[node]
    656
    0
                    .expr
    657
    0
                    .propagate_constraints(node_interval, &children_intervals)?;
    658
    0
                if let Some(propagated_intervals) = propagated_intervals {
    659
    0
                    for (child, interval) in children.into_iter().zip(propagated_intervals) {
    660
    0
                        self.graph[child].interval = interval;
    661
    0
                    }
    662
                } else {
    663
                    // The constraint is infeasible, report:
    664
    0
                    return Ok(PropagationResult::Infeasible);
    665
                }
    666
            }
    667
    0
            Ok(PropagationResult::Success)
    668
    0
        }
    669
    670
        /// Returns the interval associated with the node at the given `index`.
    671
    0
        pub fn get_interval(&self, index: usize) -> Interval {
    672
    0
            self.graph[NodeIndex::new(index)].interval.clone()
    673
    0
        }
    674
    }
    675
    676
    /// This is a subfunction of the `propagate_arithmetic` function that propagates to the right child.
    677
    0
    fn propagate_right(
    678
    0
        left: &Interval,
    679
    0
        parent: &Interval,
    680
    0
        right: &Interval,
    681
    0
        op: &Operator,
    682
    0
        inverse_op: &Operator,
    683
    0
    ) -> Result<Option<Interval>> {
    684
    0
        match op {
    685
    0
            Operator::Minus => apply_operator(op, left, parent),
    686
    0
            Operator::Plus => apply_operator(inverse_op, parent, left),
    687
    0
            Operator::Divide => apply_operator(op, left, parent),
    688
    0
            Operator::Multiply => apply_operator(inverse_op, parent, left),
    689
    0
            _ => internal_err!("Interval arithmetic does not support the operator {}", op),
    690
    0
        }?
    691
    0
        .intersect(right)
    692
    0
    }
    693
    694
    /// During the propagation of [`Interval`] values on an [`ExprIntervalGraph`],
    695
    /// if there exists a `timestamp - timestamp` operation, the result would be
    696
    /// of type `Duration`. However, we may encounter a situation where a time interval
    697
    /// is involved in an arithmetic operation with a `Duration` type. This function
    698
    /// offers special handling for such cases, where the time interval resides on
    699
    /// the left side of the operation.
    700
    0
    fn propagate_time_interval_at_left(
    701
    0
        left_child: &Interval,
    702
    0
        right_child: &Interval,
    703
    0
        parent: &Interval,
    704
    0
        op: &Operator,
    705
    0
        inverse_op: &Operator,
    706
    0
    ) -> Result<Option<(Interval, Interval)>> {
    707
        // We check if the child's time interval(s) has a non-zero month or day field(s).
    708
        // If so, we return it as is without propagating. Otherwise, we first convert
    709
        // the time intervals to the `Duration` type, then propagate, and then convert
    710
        // the bounds to time intervals again.
    711
    0
        let result = if let Some(duration) = convert_interval_type_to_duration(left_child) {
    712
    0
            match apply_operator(inverse_op, parent, right_child)?.intersect(duration)? {
    713
    0
                Some(value) => {
    714
    0
                    let left = convert_duration_type_to_interval(&value);
    715
    0
                    let right = propagate_right(&value, parent, right_child, op, inverse_op)?;
    716
    0
                    match (left, right) {
    717
    0
                        (Some(left), Some(right)) => Some((left, right)),
    718
    0
                        _ => None,
    719
                    }
    720
                }
    721
    0
                None => None,
    722
            }
    723
        } else {
    724
    0
            propagate_right(left_child, parent, right_child, op, inverse_op)?
    725
    0
                .map(|right| (left_child.clone(), right))
    726
        };
    727
    0
        Ok(result)
    728
    0
    }
    729
    730
    /// During the propagation of [`Interval`] values on an [`ExprIntervalGraph`],
    731
    /// if there exists a `timestamp - timestamp` operation, the result would be
    732
    /// of type `Duration`. However, we may encounter a situation where a time interval
    733
    /// is involved in an arithmetic operation with a `Duration` type. This function
    734
    /// offers special handling for such cases, where the time interval resides on
    735
    /// the right side of the operation.
    736
    0
    fn propagate_time_interval_at_right(
    737
    0
        left_child: &Interval,
    738
    0
        right_child: &Interval,
    739
    0
        parent: &Interval,
    740
    0
        op: &Operator,
    741
    0
        inverse_op: &Operator,
    742
    0
    ) -> Result<Option<(Interval, Interval)>> {
    743
        // We check if the child's time interval(s) has a non-zero month or day field(s).
    744
        // If so, we return it as is without propagating. Otherwise, we first convert
    745
        // the time intervals to the `Duration` type, then propagate, and then convert
    746
        // the bounds to time intervals again.
    747
    0
        let result = if let Some(duration) = convert_interval_type_to_duration(right_child) {
    748
    0
            match apply_operator(inverse_op, parent, &duration)?.intersect(left_child)? {
    749
    0
                Some(value) => {
    750
    0
                    propagate_right(left_child, parent, &duration, op, inverse_op)?
    751
    0
                        .and_then(|right| convert_duration_type_to_interval(&right))
    752
    0
                        .map(|right| (value, right))
    753
                }
    754
    0
                None => None,
    755
            }
    756
        } else {
    757
    0
            apply_operator(inverse_op, parent, right_child)?
    758
    0
                .intersect(left_child)?
    759
    0
                .map(|value| (value, right_child.clone()))
    760
        };
    761
    0
        Ok(result)
    762
    0
    }
    763
    764
    0
    fn reverse_tuple<T, U>((first, second): (T, U)) -> (U, T) {
    765
    0
        (second, first)
    766
    0
    }
    767
    768
    #[cfg(test)]
    769
    mod tests {
    770
        use super::*;
    771
        use crate::expressions::{BinaryExpr, Column};
    772
        use crate::intervals::test_utils::gen_conjunctive_numerical_expr;
    773
    774
        use arrow::array::types::{IntervalDayTime, IntervalMonthDayNano};
    775
        use arrow::datatypes::{Field, TimeUnit};
    776
        use datafusion_common::ScalarValue;
    777
    778
        use itertools::Itertools;
    779
        use rand::rngs::StdRng;
    780
        use rand::{Rng, SeedableRng};
    781
        use rstest::*;
    782
    783
        #[allow(clippy::too_many_arguments)]
    784
    0
        fn experiment(
    785
    0
            expr: Arc<dyn PhysicalExpr>,
    786
    0
            exprs_with_interval: (Arc<dyn PhysicalExpr>, Arc<dyn PhysicalExpr>),
    787
    0
            left_interval: Interval,
    788
    0
            right_interval: Interval,
    789
    0
            left_expected: Interval,
    790
    0
            right_expected: Interval,
    791
    0
            result: PropagationResult,
    792
    0
            schema: &Schema,
    793
    0
        ) -> Result<()> {
    794
    0
            let col_stats = [
    795
    0
                (Arc::clone(&exprs_with_interval.0), left_interval),
    796
    0
                (Arc::clone(&exprs_with_interval.1), right_interval),
    797
    0
            ];
    798
    0
            let expected = [
    799
    0
                (Arc::clone(&exprs_with_interval.0), left_expected),
    800
    0
                (Arc::clone(&exprs_with_interval.1), right_expected),
    801
    0
            ];
    802
    0
            let mut graph = ExprIntervalGraph::try_new(expr, schema)?;
    803
    0
            let expr_indexes = graph.gather_node_indices(
    804
    0
                &col_stats.iter().map(|(e, _)| Arc::clone(e)).collect_vec(),
    805
            );
    806
    807
    0
            let mut col_stat_nodes = col_stats
    808
    0
                .iter()
    809
    0
                .zip(expr_indexes.iter())
    810
    0
                .map(|((_, interval), (_, index))| (*index, interval.clone()))
    811
    0
                .collect_vec();
    812
    0
            let expected_nodes = expected
    813
    0
                .iter()
    814
    0
                .zip(expr_indexes.iter())
    815
    0
                .map(|((_, interval), (_, index))| (*index, interval.clone()))
    816
    0
                .collect_vec();
    817
    818
    0
            let exp_result = graph.update_ranges(&mut col_stat_nodes[..], Interval::TRUE)?;
    819
    0
            assert_eq!(exp_result, result);
    820
    0
            col_stat_nodes.iter().zip(expected_nodes.iter()).for_each(
    821
    0
                |((_, calculated_interval_node), (_, expected))| {
    822
                    // NOTE: These randomized tests only check for conservative containment,
    823
                    // not openness/closedness of endpoints.
    824
    825
                    // Calculated bounds are relaxed by 1 to cover all strict and
    826
                    // and non-strict comparison cases since we have only closed bounds.
    827
    0
                    let one = ScalarValue::new_one(&expected.data_type()).unwrap();
    828
    0
                    assert!(
    829
    0
                        calculated_interval_node.lower()
    830
    0
                            <= &expected.lower().add(&one).unwrap(),
    831
                        "{}",
    832
    0
                        format!(
    833
    0
                            "Calculated {} must be less than or equal {}",
    834
    0
                            calculated_interval_node.lower(),
    835
    0
                            expected.lower()
    836
                        )
    837
                    );
    838
    0
                    assert!(
    839
    0
                        calculated_interval_node.upper()
    840
    0
                            >= &expected.upper().sub(&one).unwrap(),
    841
                        "{}",
    842
    0
                        format!(
    843
    0
                            "Calculated {} must be greater than or equal {}",
    844
    0
                            calculated_interval_node.upper(),
    845
    0
                            expected.upper()
    846
                        )
    847
                    );
    848
    0
                },
    849
            );
    850
    0
            Ok(())
    851
    0
        }
    852
    853
        macro_rules! generate_cases {
    854
            ($FUNC_NAME:ident, $TYPE:ty, $SCALAR:ident) => {
    855
    0
                fn $FUNC_NAME<const ASC: bool>(
    856
    0
                    expr: Arc<dyn PhysicalExpr>,
    857
    0
                    left_col: Arc<dyn PhysicalExpr>,
    858
    0
                    right_col: Arc<dyn PhysicalExpr>,
    859
    0
                    seed: u64,
    860
    0
                    expr_left: $TYPE,
    861
    0
                    expr_right: $TYPE,
    862
    0
                ) -> Result<()> {
    863
    0
                    let mut r = StdRng::seed_from_u64(seed);
    864
    865
    0
                    let (left_given, right_given, left_expected, right_expected) = if ASC {
    866
    0
                        let left = r.random_range((0 as $TYPE)..(1000 as $TYPE));
    867
    0
                        let right = r.random_range((0 as $TYPE)..(1000 as $TYPE));
    868
    0
                        (
    869
    0
                            (Some(left), None),
    870
    0
                            (Some(right), None),
    871
    0
                            (Some(<$TYPE>::max(left, right + expr_left)), None),
    872
    0
                            (Some(<$TYPE>::max(right, left + expr_right)), None),
    873
    0
                        )
    874
                    } else {
    875
    0
                        let left = r.random_range((0 as $TYPE)..(1000 as $TYPE));
    876
    0
                        let right = r.random_range((0 as $TYPE)..(1000 as $TYPE));
    877
    0
                        (
    878
    0
                            (None, Some(left)),
    879
    0
                            (None, Some(right)),
    880
    0
                            (None, Some(<$TYPE>::min(left, right + expr_left))),
    881
    0
                            (None, Some(<$TYPE>::min(right, left + expr_right))),
    882
    0
                        )
    883
                    };
    884
    885
    0
                    experiment(
    886
    0
                        expr,
    887
    0
                        (left_col.clone(), right_col.clone()),
    888
    0
                        Interval::make(left_given.0, left_given.1).unwrap(),
    889
    0
                        Interval::make(right_given.0, right_given.1).unwrap(),
    890
    0
                        Interval::make(left_expected.0, left_expected.1).unwrap(),
    891
    0
                        Interval::make(right_expected.0, right_expected.1).unwrap(),
    892
    0
                        PropagationResult::Success,
    893
    0
                        &Schema::new(vec![
    894
    0
                            Field::new(
    895
    0
                                left_col.as_any().downcast_ref::<Column>().unwrap().name(),
    896
    0
                                DataType::$SCALAR,
    897
    0
                                true,
    898
    0
                            ),
    899
    0
                            Field::new(
    900
    0
                                right_col.as_any().downcast_ref::<Column>().unwrap().name(),
    901
    0
                                DataType::$SCALAR,
    902
    0
                                true,
    903
    0
                            ),
    904
    0
                        ]),
    905
                    )
    906
    0
                }
    907
            };
    908
        }
    909
        generate_cases!(generate_case_i32, i32, Int32);
    910
        generate_cases!(generate_case_i64, i64, Int64);
    911
        generate_cases!(generate_case_f32, f32, Float32);
    912
        generate_cases!(generate_case_f64, f64, Float64);
    913
    914
        #[test]
    915
    0
        fn testing_not_possible() -> Result<()> {
    916
    0
            let left_col = Arc::new(Column::new("left_watermark", 0));
    917
    0
            let right_col = Arc::new(Column::new("right_watermark", 0));
    918
    919
            // left_watermark > right_watermark + 5
    920
    0
            let left_and_1 = Arc::new(BinaryExpr::new(
    921
    0
                Arc::clone(&left_col) as Arc<dyn PhysicalExpr>,
    922
    0
                Operator::Plus,
    923
    0
                Arc::new(Literal::new(ScalarValue::Int32(Some(5)))),
    924
            ));
    925
    0
            let expr = Arc::new(BinaryExpr::new(
    926
    0
                left_and_1,
    927
    0
                Operator::Gt,
    928
    0
                Arc::clone(&right_col) as Arc<dyn PhysicalExpr>,
    929
            ));
    930
    0
            experiment(
    931
    0
                expr,
    932
    0
                (
    933
    0
                    Arc::clone(&left_col) as Arc<dyn PhysicalExpr>,
    934
    0
                    Arc::clone(&right_col) as Arc<dyn PhysicalExpr>,
    935
    0
                ),
    936
    0
                Interval::make(Some(10_i32), Some(20_i32))?,
    937
    0
                Interval::make(Some(100), None)?,
    938
    0
                Interval::make(Some(10), Some(20))?,
    939
    0
                Interval::make(Some(100), None)?,
    940
    0
                PropagationResult::Infeasible,
    941
    0
                &Schema::new(vec![
    942
    0
                    Field::new(
    943
    0
                        left_col.as_any().downcast_ref::<Column>().unwrap().name(),
    944
    0
                        DataType::Int32,
    945
    0
                        true,
    946
    0
                    ),
    947
    0
                    Field::new(
    948
    0
                        right_col.as_any().downcast_ref::<Column>().unwrap().name(),
    949
    0
                        DataType::Int32,
    950
    0
                        true,
    951
    0
                    ),
    952
    0
                ]),
    953
            )
    954
    0
        }
    955
    956
        macro_rules! integer_float_case_1 {
    957
            ($TEST_FUNC_NAME:ident, $GENERATE_CASE_FUNC_NAME:ident, $TYPE:ty, $SCALAR:ident) => {
    958
                #[rstest]
    959
                #[test]
    960
                fn $TEST_FUNC_NAME(
    961
                    #[values(0, 1, 2, 3, 4, 12, 32, 314, 3124, 123, 125, 211, 215, 4123)]
    962
                    seed: u64,
    963
                    #[values(Operator::Gt, Operator::GtEq)] greater_op: Operator,
    964
                    #[values(Operator::Lt, Operator::LtEq)] less_op: Operator,
    965
                ) -> Result<()> {
    966
                    let left_col = Arc::new(Column::new("left_watermark", 0));
    967
                    let right_col = Arc::new(Column::new("right_watermark", 0));
    968
    969
                    // left_watermark + 1 > right_watermark + 11 AND left_watermark + 3 < right_watermark + 33
    970
                    let expr = gen_conjunctive_numerical_expr(
    971
                        left_col.clone(),
    972
                        right_col.clone(),
    973
                        (
    974
                            Operator::Plus,
    975
                            Operator::Plus,
    976
                            Operator::Plus,
    977
                            Operator::Plus,
    978
                        ),
    979
                        ScalarValue::$SCALAR(Some(1 as $TYPE)),
    980
                        ScalarValue::$SCALAR(Some(11 as $TYPE)),
    981
                        ScalarValue::$SCALAR(Some(3 as $TYPE)),
    982
                        ScalarValue::$SCALAR(Some(33 as $TYPE)),
    983
                        (greater_op, less_op),
    984
                    );
    985
                    // l > r + 10 AND r > l - 30
    986
                    let l_gt_r = 10 as $TYPE;
    987
                    let r_gt_l = -30 as $TYPE;
    988
                    $GENERATE_CASE_FUNC_NAME::<true>(
    989
                        expr.clone(),
    990
                        left_col.clone(),
    991
                        right_col.clone(),
    992
                        seed,
    993
                        l_gt_r,
    994
                        r_gt_l,
    995
                    )?;
    996
                    // Descending tests
    997
                    // r < l - 10 AND l < r + 30
    998
                    let r_lt_l = -l_gt_r;
    999
                    let l_lt_r = -r_gt_l;
    1000
                    $GENERATE_CASE_FUNC_NAME::<false>(
    1001
                        expr, left_col, right_col, seed, l_lt_r, r_lt_l,
    1002
                    )
    1003
                }
    1004
            };
    1005
        }
    1006
    1007
        integer_float_case_1!(case_1_i32, generate_case_i32, i32, Int32);
    1008
        integer_float_case_1!(case_1_i64, generate_case_i64, i64, Int64);
    1009
        integer_float_case_1!(case_1_f64, generate_case_f64, f64, Float64);
    1010
        integer_float_case_1!(case_1_f32, generate_case_f32, f32, Float32);
    1011
    1012
        macro_rules! integer_float_case_2 {
    1013
            ($TEST_FUNC_NAME:ident, $GENERATE_CASE_FUNC_NAME:ident, $TYPE:ty, $SCALAR:ident) => {
    1014
                #[rstest]
    1015
                #[test]
    1016
                fn $TEST_FUNC_NAME(
    1017
                    #[values(0, 1, 2, 3, 4, 12, 32, 314, 3124, 123, 125, 211, 215, 4123)]
    1018
                    seed: u64,
    1019
                    #[values(Operator::Gt, Operator::GtEq)] greater_op: Operator,
    1020
                    #[values(Operator::Lt, Operator::LtEq)] less_op: Operator,
    1021
                ) -> Result<()> {
    1022
                    let left_col = Arc::new(Column::new("left_watermark", 0));
    1023
                    let right_col = Arc::new(Column::new("right_watermark", 0));
    1024
    1025
                    // left_watermark - 1 > right_watermark + 5 AND left_watermark + 3 < right_watermark + 10
    1026
                    let expr = gen_conjunctive_numerical_expr(
    1027
                        left_col.clone(),
    1028
                        right_col.clone(),
    1029
                        (
    1030
                            Operator::Minus,
    1031
                            Operator::Plus,
    1032
                            Operator::Plus,
    1033
                            Operator::Plus,
    1034
                        ),
    1035
                        ScalarValue::$SCALAR(Some(1 as $TYPE)),
    1036
                        ScalarValue::$SCALAR(Some(5 as $TYPE)),
    1037
                        ScalarValue::$SCALAR(Some(3 as $TYPE)),
    1038
                        ScalarValue::$SCALAR(Some(10 as $TYPE)),
    1039
                        (greater_op, less_op),
    1040
                    );
    1041
                    // l > r + 6 AND r > l - 7
    1042
                    let l_gt_r = 6 as $TYPE;
    1043
                    let r_gt_l = -7 as $TYPE;
    1044
                    $GENERATE_CASE_FUNC_NAME::<true>(
    1045
                        expr.clone(),
    1046
                        left_col.clone(),
    1047
                        right_col.clone(),
    1048
                        seed,
    1049
                        l_gt_r,
    1050
                        r_gt_l,
    1051
                    )?;
    1052
                    // Descending tests
    1053
                    // r < l - 6 AND l < r + 7
    1054
                    let r_lt_l = -l_gt_r;
    1055
                    let l_lt_r = -r_gt_l;
    1056
                    $GENERATE_CASE_FUNC_NAME::<false>(
    1057
                        expr, left_col, right_col, seed, l_lt_r, r_lt_l,
    1058
                    )
    1059
                }
    1060
            };
    1061
        }
    1062
    1063
        integer_float_case_2!(case_2_i32, generate_case_i32, i32, Int32);
    1064
        integer_float_case_2!(case_2_i64, generate_case_i64, i64, Int64);
    1065
        integer_float_case_2!(case_2_f64, generate_case_f64, f64, Float64);
    1066
        integer_float_case_2!(case_2_f32, generate_case_f32, f32, Float32);
    1067
    1068
        macro_rules! integer_float_case_3 {
    1069
            ($TEST_FUNC_NAME:ident, $GENERATE_CASE_FUNC_NAME:ident, $TYPE:ty, $SCALAR:ident) => {
    1070
                #[rstest]
    1071
                #[test]
    1072
                fn $TEST_FUNC_NAME(
    1073
                    #[values(0, 1, 2, 3, 4, 12, 32, 314, 3124, 123, 125, 211, 215, 4123)]
    1074
                    seed: u64,
    1075
                    #[values(Operator::Gt, Operator::GtEq)] greater_op: Operator,
    1076
                    #[values(Operator::Lt, Operator::LtEq)] less_op: Operator,
    1077
                ) -> Result<()> {
    1078
                    let left_col = Arc::new(Column::new("left_watermark", 0));
    1079
                    let right_col = Arc::new(Column::new("right_watermark", 0));
    1080
    1081
                    // left_watermark - 1 > right_watermark + 5 AND left_watermark - 3 < right_watermark + 10
    1082
                    let expr = gen_conjunctive_numerical_expr(
    1083
                        left_col.clone(),
    1084
                        right_col.clone(),
    1085
                        (
    1086
                            Operator::Minus,
    1087
                            Operator::Plus,
    1088
                            Operator::Minus,
    1089
                            Operator::Plus,
    1090
                        ),
    1091
                        ScalarValue::$SCALAR(Some(1 as $TYPE)),
    1092
                        ScalarValue::$SCALAR(Some(5 as $TYPE)),
    1093
                        ScalarValue::$SCALAR(Some(3 as $TYPE)),
    1094
                        ScalarValue::$SCALAR(Some(10 as $TYPE)),
    1095
                        (greater_op, less_op),
    1096
                    );
    1097
                    // l > r + 6 AND r > l - 13
    1098
                    let l_gt_r = 6 as $TYPE;
    1099
                    let r_gt_l = -13 as $TYPE;
    1100
                    $GENERATE_CASE_FUNC_NAME::<true>(
    1101
                        expr.clone(),
    1102
                        left_col.clone(),
    1103
                        right_col.clone(),
    1104
                        seed,
    1105
                        l_gt_r,
    1106
                        r_gt_l,
    1107
                    )?;
    1108
                    // Descending tests
    1109
                    // r < l - 6 AND l < r + 13
    1110
                    let r_lt_l = -l_gt_r;
    1111
                    let l_lt_r = -r_gt_l;
    1112
                    $GENERATE_CASE_FUNC_NAME::<false>(
    1113
                        expr, left_col, right_col, seed, l_lt_r, r_lt_l,
    1114
                    )
    1115
                }
    1116
            };
    1117
        }
    1118
    1119
        integer_float_case_3!(case_3_i32, generate_case_i32, i32, Int32);
    1120
        integer_float_case_3!(case_3_i64, generate_case_i64, i64, Int64);
    1121
        integer_float_case_3!(case_3_f64, generate_case_f64, f64, Float64);
    1122
        integer_float_case_3!(case_3_f32, generate_case_f32, f32, Float32);
    1123
    1124
        macro_rules! integer_float_case_4 {
    1125
            ($TEST_FUNC_NAME:ident, $GENERATE_CASE_FUNC_NAME:ident, $TYPE:ty, $SCALAR:ident) => {
    1126
                #[rstest]
    1127
                #[test]
    1128
                fn $TEST_FUNC_NAME(
    1129
                    #[values(0, 1, 2, 3, 4, 12, 32, 314, 3124, 123, 125, 211, 215, 4123)]
    1130
                    seed: u64,
    1131
                    #[values(Operator::Gt, Operator::GtEq)] greater_op: Operator,
    1132
                    #[values(Operator::Lt, Operator::LtEq)] less_op: Operator,
    1133
                ) -> Result<()> {
    1134
                    let left_col = Arc::new(Column::new("left_watermark", 0));
    1135
                    let right_col = Arc::new(Column::new("right_watermark", 0));
    1136
    1137
                    // left_watermark - 10 > right_watermark - 5 AND left_watermark - 30 < right_watermark - 3
    1138
                    let expr = gen_conjunctive_numerical_expr(
    1139
                        left_col.clone(),
    1140
                        right_col.clone(),
    1141
                        (
    1142
                            Operator::Minus,
    1143
                            Operator::Minus,
    1144
                            Operator::Minus,
    1145
                            Operator::Plus,
    1146
                        ),
    1147
                        ScalarValue::$SCALAR(Some(10 as $TYPE)),
    1148
                        ScalarValue::$SCALAR(Some(5 as $TYPE)),
    1149
                        ScalarValue::$SCALAR(Some(3 as $TYPE)),
    1150
                        ScalarValue::$SCALAR(Some(10 as $TYPE)),
    1151
                        (greater_op, less_op),
    1152
                    );
    1153
                    // l > r + 5 AND r > l - 13
    1154
                    let l_gt_r = 5 as $TYPE;
    1155
                    let r_gt_l = -13 as $TYPE;
    1156
                    $GENERATE_CASE_FUNC_NAME::<true>(
    1157
                        expr.clone(),
    1158
                        left_col.clone(),
    1159
                        right_col.clone(),
    1160
                        seed,
    1161
                        l_gt_r,
    1162
                        r_gt_l,
    1163
                    )?;
    1164
                    // Descending tests
    1165
                    // r < l - 5 AND l < r + 13
    1166
                    let r_lt_l = -l_gt_r;
    1167
                    let l_lt_r = -r_gt_l;
    1168
                    $GENERATE_CASE_FUNC_NAME::<false>(
    1169
                        expr, left_col, right_col, seed, l_lt_r, r_lt_l,
    1170
                    )
    1171
                }
    1172
            };
    1173
        }
    1174
    1175
        integer_float_case_4!(case_4_i32, generate_case_i32, i32, Int32);
    1176
        integer_float_case_4!(case_4_i64, generate_case_i64, i64, Int64);
    1177
        integer_float_case_4!(case_4_f64, generate_case_f64, f64, Float64);
    1178
        integer_float_case_4!(case_4_f32, generate_case_f32, f32, Float32);
    1179
    1180
        macro_rules! integer_float_case_5 {
    1181
            ($TEST_FUNC_NAME:ident, $GENERATE_CASE_FUNC_NAME:ident, $TYPE:ty, $SCALAR:ident) => {
    1182
                #[rstest]
    1183
                #[test]
    1184
                fn $TEST_FUNC_NAME(
    1185
                    #[values(0, 1, 2, 3, 4, 12, 32, 314, 3124, 123, 125, 211, 215, 4123)]
    1186
                    seed: u64,
    1187
                    #[values(Operator::Gt, Operator::GtEq)] greater_op: Operator,
    1188
                    #[values(Operator::Lt, Operator::LtEq)] less_op: Operator,
    1189
                ) -> Result<()> {
    1190
                    let left_col = Arc::new(Column::new("left_watermark", 0));
    1191
                    let right_col = Arc::new(Column::new("right_watermark", 0));
    1192
    1193
                    // left_watermark - 10 > right_watermark - 5 AND left_watermark - 30 < right_watermark - 3
    1194
                    let expr = gen_conjunctive_numerical_expr(
    1195
                        left_col.clone(),
    1196
                        right_col.clone(),
    1197
                        (
    1198
                            Operator::Minus,
    1199
                            Operator::Minus,
    1200
                            Operator::Minus,
    1201
                            Operator::Minus,
    1202
                        ),
    1203
                        ScalarValue::$SCALAR(Some(10 as $TYPE)),
    1204
                        ScalarValue::$SCALAR(Some(5 as $TYPE)),
    1205
                        ScalarValue::$SCALAR(Some(30 as $TYPE)),
    1206
                        ScalarValue::$SCALAR(Some(3 as $TYPE)),
    1207
                        (greater_op, less_op),
    1208
                    );
    1209
                    // l > r + 5 AND r > l - 27
    1210
                    let l_gt_r = 5 as $TYPE;
    1211
                    let r_gt_l = -27 as $TYPE;
    1212
                    $GENERATE_CASE_FUNC_NAME::<true>(
    1213
                        expr.clone(),
    1214
                        left_col.clone(),
    1215
                        right_col.clone(),
    1216
                        seed,
    1217
                        l_gt_r,
    1218
                        r_gt_l,
    1219
                    )?;
    1220
                    // Descending tests
    1221
                    // r < l - 5 AND l < r + 27
    1222
                    let r_lt_l = -l_gt_r;
    1223
                    let l_lt_r = -r_gt_l;
    1224
                    $GENERATE_CASE_FUNC_NAME::<false>(
    1225
                        expr, left_col, right_col, seed, l_lt_r, r_lt_l,
    1226
                    )
    1227
                }
    1228
            };
    1229
        }
    1230
    1231
        integer_float_case_5!(case_5_i32, generate_case_i32, i32, Int32);
    1232
        integer_float_case_5!(case_5_i64, generate_case_i64, i64, Int64);
    1233
        integer_float_case_5!(case_5_f64, generate_case_f64, f64, Float64);
    1234
        integer_float_case_5!(case_5_f32, generate_case_f32, f32, Float32);
    1235
    1236
        #[test]
    1237
    0
        fn test_gather_node_indices_dont_remove() -> Result<()> {
    1238
            // Expression: a@0 + b@1 + 1 > a@0 - b@1, given a@0 + b@1.
    1239
            // Do not remove a@0 or b@1, only remove edges since a@0 - b@1 also
    1240
            // depends on leaf nodes a@0 and b@1.
    1241
    0
            let left_expr = Arc::new(BinaryExpr::new(
    1242
    0
                Arc::new(BinaryExpr::new(
    1243
    0
                    Arc::new(Column::new("a", 0)),
    1244
    0
                    Operator::Plus,
    1245
    0
                    Arc::new(Column::new("b", 1)),
    1246
    0
                )),
    1247
    0
                Operator::Plus,
    1248
    0
                Arc::new(Literal::new(ScalarValue::Int32(Some(1)))),
    1249
            ));
    1250
    1251
    0
            let right_expr = Arc::new(BinaryExpr::new(
    1252
    0
                Arc::new(Column::new("a", 0)),
    1253
    0
                Operator::Minus,
    1254
    0
                Arc::new(Column::new("b", 1)),
    1255
            ));
    1256
    0
            let expr = Arc::new(BinaryExpr::new(left_expr, Operator::Gt, right_expr));
    1257
    0
            let mut graph = ExprIntervalGraph::try_new(
    1258
    0
                expr,
    1259
    0
                &Schema::new(vec![
    1260
    0
                    Field::new("a", DataType::Int32, true),
    1261
    0
                    Field::new("b", DataType::Int32, true),
    1262
    0
                ]),
    1263
            )
    1264
    0
            .unwrap();
    1265
            // Define a test leaf node.
    1266
    0
            let leaf_node = Arc::new(BinaryExpr::new(
    1267
    0
                Arc::new(Column::new("a", 0)),
    1268
    0
                Operator::Plus,
    1269
    0
                Arc::new(Column::new("b", 1)),
    1270
            ));
    1271
            // Store the current node count.
    1272
    0
            let prev_node_count = graph.node_count();
    1273
            // Gather the index of node in the expression graph that match the test leaf node.
    1274
    0
            graph.gather_node_indices(&[leaf_node]);
    1275
            // Store the final node count.
    1276
    0
            let final_node_count = graph.node_count();
    1277
            // Assert that the final node count is equal the previous node count.
    1278
            // This means we did not remove any node.
    1279
    0
            assert_eq!(prev_node_count, final_node_count);
    1280
    0
            Ok(())
    1281
    0
        }
    1282
    1283
        #[test]
    1284
    0
        fn test_gather_node_indices_remove() -> Result<()> {
    1285
            // Expression: a@0 + b@1 + 1 > y@0 - z@1, given a@0 + b@1.
    1286
            // We expect to remove two nodes since we do not need a@ and b@.
    1287
    0
            let left_expr = Arc::new(BinaryExpr::new(
    1288
    0
                Arc::new(BinaryExpr::new(
    1289
    0
                    Arc::new(Column::new("a", 0)),
    1290
    0
                    Operator::Plus,
    1291
    0
                    Arc::new(Column::new("b", 1)),
    1292
    0
                )),
    1293
    0
                Operator::Plus,
    1294
    0
                Arc::new(Literal::new(ScalarValue::Int32(Some(1)))),
    1295
            ));
    1296
    1297
    0
            let right_expr = Arc::new(BinaryExpr::new(
    1298
    0
                Arc::new(Column::new("y", 0)),
    1299
    0
                Operator::Minus,
    1300
    0
                Arc::new(Column::new("z", 1)),
    1301
            ));
    1302
    0
            let expr = Arc::new(BinaryExpr::new(left_expr, Operator::Gt, right_expr));
    1303
    0
            let mut graph = ExprIntervalGraph::try_new(
    1304
    0
                expr,
    1305
    0
                &Schema::new(vec![
    1306
    0
                    Field::new("a", DataType::Int32, true),
    1307
    0
                    Field::new("b", DataType::Int32, true),
    1308
    0
                    Field::new("y", DataType::Int32, true),
    1309
    0
                    Field::new("z", DataType::Int32, true),
    1310
    0
                ]),
    1311
            )
    1312
    0
            .unwrap();
    1313
            // Define a test leaf node.
    1314
    0
            let leaf_node = Arc::new(BinaryExpr::new(
    1315
    0
                Arc::new(Column::new("a", 0)),
    1316
    0
                Operator::Plus,
    1317
    0
                Arc::new(Column::new("b", 1)),
    1318
            ));
    1319
            // Store the current node count.
    1320
    0
            let prev_node_count = graph.node_count();
    1321
            // Gather the index of node in the expression graph that match the test leaf node.
    1322
    0
            graph.gather_node_indices(&[leaf_node]);
    1323
            // Store the final node count.
    1324
    0
            let final_node_count = graph.node_count();
    1325
            // Assert that the final node count is two less than the previous node
    1326
            // count; i.e. that we did remove two nodes.
    1327
    0
            assert_eq!(prev_node_count, final_node_count + 2);
    1328
    0
            Ok(())
    1329
    0
        }
    1330
    1331
        #[test]
    1332
    0
        fn test_gather_node_indices_remove_one() -> Result<()> {
    1333
            // Expression: a@0 + b@1 + 1 > a@0 - z@1, given a@0 + b@1.
    1334
            // We expect to remove one nodesince we still need a@ but not b@.
    1335
    0
            let left_expr = Arc::new(BinaryExpr::new(
    1336
    0
                Arc::new(BinaryExpr::new(
    1337
    0
                    Arc::new(Column::new("a", 0)),
    1338
    0
                    Operator::Plus,
    1339
    0
                    Arc::new(Column::new("b", 1)),
    1340
    0
                )),
    1341
    0
                Operator::Plus,
    1342
    0
                Arc::new(Literal::new(ScalarValue::Int32(Some(1)))),
    1343
            ));
    1344
    1345
    0
            let right_expr = Arc::new(BinaryExpr::new(
    1346
    0
                Arc::new(Column::new("a", 0)),
    1347
    0
                Operator::Minus,
    1348
    0
                Arc::new(Column::new("z", 1)),
    1349
            ));
    1350
    0
            let expr = Arc::new(BinaryExpr::new(left_expr, Operator::Gt, right_expr));
    1351
    0
            let mut graph = ExprIntervalGraph::try_new(
    1352
    0
                expr,
    1353
    0
                &Schema::new(vec![
    1354
    0
                    Field::new("a", DataType::Int32, true),
    1355
    0
                    Field::new("b", DataType::Int32, true),
    1356
    0
                    Field::new("z", DataType::Int32, true),
    1357
    0
                ]),
    1358
            )
    1359
    0
            .unwrap();
    1360
            // Define a test leaf node.
    1361
    0
            let leaf_node = Arc::new(BinaryExpr::new(
    1362
    0
                Arc::new(Column::new("a", 0)),
    1363
    0
                Operator::Plus,
    1364
    0
                Arc::new(Column::new("b", 1)),
    1365
            ));
    1366
            // Store the current node count.
    1367
    0
            let prev_node_count = graph.node_count();
    1368
            // Gather the index of node in the expression graph that match the test leaf node.
    1369
    0
            graph.gather_node_indices(&[leaf_node]);
    1370
            // Store the final node count.
    1371
    0
            let final_node_count = graph.node_count();
    1372
            // Assert that the final node count is one less than the previous node
    1373
            // count; i.e. that we did remove two nodes.
    1374
    0
            assert_eq!(prev_node_count, final_node_count + 1);
    1375
    0
            Ok(())
    1376
    0
        }
    1377
    1378
        #[test]
    1379
    0
        fn test_gather_node_indices_cannot_provide() -> Result<()> {
    1380
            // Expression: a@0 + 1 + b@1 > y@0 - z@1 -> provide a@0 + b@1
    1381
            // TODO: We expect nodes a@0 and b@1 to be pruned, and intervals to be provided from the a@0 + b@1 node.
    1382
            // However, we do not have an exact node for a@0 + b@1 due to the binary tree structure of the expressions.
    1383
            // Pruning and interval providing for BinaryExpr expressions are more challenging without exact matches.
    1384
            // Currently, we only support exact matches for BinaryExprs, but we plan to extend support beyond exact matches in the future.
    1385
    0
            let left_expr = Arc::new(BinaryExpr::new(
    1386
    0
                Arc::new(BinaryExpr::new(
    1387
    0
                    Arc::new(Column::new("a", 0)),
    1388
    0
                    Operator::Plus,
    1389
    0
                    Arc::new(Literal::new(ScalarValue::Int32(Some(1)))),
    1390
    0
                )),
    1391
    0
                Operator::Plus,
    1392
    0
                Arc::new(Column::new("b", 1)),
    1393
            ));
    1394
    1395
    0
            let right_expr = Arc::new(BinaryExpr::new(
    1396
    0
                Arc::new(Column::new("y", 0)),
    1397
    0
                Operator::Minus,
    1398
    0
                Arc::new(Column::new("z", 1)),
    1399
            ));
    1400
    0
            let expr = Arc::new(BinaryExpr::new(left_expr, Operator::Gt, right_expr));
    1401
    0
            let mut graph = ExprIntervalGraph::try_new(
    1402
    0
                expr,
    1403
    0
                &Schema::new(vec![
    1404
    0
                    Field::new("a", DataType::Int32, true),
    1405
    0
                    Field::new("b", DataType::Int32, true),
    1406
    0
                    Field::new("y", DataType::Int32, true),
    1407
    0
                    Field::new("z", DataType::Int32, true),
    1408
    0
                ]),
    1409
            )
    1410
    0
            .unwrap();
    1411
            // Define a test leaf node.
    1412
    0
            let leaf_node = Arc::new(BinaryExpr::new(
    1413
    0
                Arc::new(Column::new("a", 0)),
    1414
    0
                Operator::Plus,
    1415
    0
                Arc::new(Column::new("b", 1)),
    1416
            ));
    1417
            // Store the current node count.
    1418
    0
            let prev_node_count = graph.node_count();
    1419
            // Gather the index of node in the expression graph that match the test leaf node.
    1420
    0
            graph.gather_node_indices(&[leaf_node]);
    1421
            // Store the final node count.
    1422
    0
            let final_node_count = graph.node_count();
    1423
            // Assert that the final node count is equal the previous node count (i.e., no node was pruned).
    1424
    0
            assert_eq!(prev_node_count, final_node_count);
    1425
    0
            Ok(())
    1426
    0
        }
    1427
    1428
        #[test]
    1429
    0
        fn test_propagate_constraints_singleton_interval_at_right() -> Result<()> {
    1430
    0
            let expression = BinaryExpr::new(
    1431
    0
                Arc::new(Column::new("ts_column", 0)),
    1432
    0
                Operator::Plus,
    1433
    0
                Arc::new(Literal::new(ScalarValue::new_interval_mdn(0, 1, 321))),
    1434
            );
    1435
    0
            let parent = Interval::try_new(
    1436
                // 15.10.2020 - 10:11:12.000_000_321 AM
    1437
    0
                ScalarValue::TimestampNanosecond(Some(1_602_756_672_000_000_321), None),
    1438
                // 16.10.2020 - 10:11:12.000_000_321 AM
    1439
    0
                ScalarValue::TimestampNanosecond(Some(1_602_843_072_000_000_321), None),
    1440
    0
            )?;
    1441
    0
            let left_child = Interval::try_new(
    1442
                // 10.10.2020 - 10:11:12 AM
    1443
    0
                ScalarValue::TimestampNanosecond(Some(1_602_324_672_000_000_000), None),
    1444
                // 20.10.2020 - 10:11:12 AM
    1445
    0
                ScalarValue::TimestampNanosecond(Some(1_603_188_672_000_000_000), None),
    1446
    0
            )?;
    1447
    0
            let right_child = Interval::try_new(
    1448
                // 1 day 321 ns
    1449
    0
                ScalarValue::IntervalMonthDayNano(Some(IntervalMonthDayNano {
    1450
    0
                    months: 0,
    1451
    0
                    days: 1,
    1452
    0
                    nanoseconds: 321,
    1453
    0
                })),
    1454
                // 1 day 321 ns
    1455
    0
                ScalarValue::IntervalMonthDayNano(Some(IntervalMonthDayNano {
    1456
    0
                    months: 0,
    1457
    0
                    days: 1,
    1458
    0
                    nanoseconds: 321,
    1459
    0
                })),
    1460
    0
            )?;
    1461
    0
            let children = vec![&left_child, &right_child];
    1462
    0
            let result = expression
    1463
    0
                .propagate_constraints(&parent, &children)?
    1464
    0
                .unwrap();
    1465
    1466
    0
            assert_eq!(
    1467
    0
                vec![
    1468
    0
                    Interval::try_new(
    1469
                        // 14.10.2020 - 10:11:12 AM
    1470
    0
                        ScalarValue::TimestampNanosecond(
    1471
    0
                            Some(1_602_670_272_000_000_000),
    1472
    0
                            None
    1473
    0
                        ),
    1474
                        // 15.10.2020 - 10:11:12 AM
    1475
    0
                        ScalarValue::TimestampNanosecond(
    1476
    0
                            Some(1_602_756_672_000_000_000),
    1477
    0
                            None
    1478
    0
                        ),
    1479
    0
                    )?,
    1480
    0
                    Interval::try_new(
    1481
                        // 1 day 321 ns in Duration type
    1482
    0
                        ScalarValue::IntervalMonthDayNano(Some(IntervalMonthDayNano {
    1483
    0
                            months: 0,
    1484
    0
                            days: 1,
    1485
    0
                            nanoseconds: 321,
    1486
    0
                        })),
    1487
                        // 1 day 321 ns in Duration type
    1488
    0
                        ScalarValue::IntervalMonthDayNano(Some(IntervalMonthDayNano {
    1489
    0
                            months: 0,
    1490
    0
                            days: 1,
    1491
    0
                            nanoseconds: 321,
    1492
    0
                        })),
    1493
    0
                    )?
    1494
                ],
    1495
                result
    1496
            );
    1497
    1498
    0
            Ok(())
    1499
    0
        }
    1500
    1501
        #[test]
    1502
    0
        fn test_propagate_constraints_column_interval_at_left() -> Result<()> {
    1503
    0
            let expression = BinaryExpr::new(
    1504
    0
                Arc::new(Column::new("interval_column", 1)),
    1505
    0
                Operator::Plus,
    1506
    0
                Arc::new(Column::new("ts_column", 0)),
    1507
            );
    1508
    0
            let parent = Interval::try_new(
    1509
                // 15.10.2020 - 10:11:12 AM
    1510
    0
                ScalarValue::TimestampMillisecond(Some(1_602_756_672_000), None),
    1511
                // 16.10.2020 - 10:11:12 AM
    1512
    0
                ScalarValue::TimestampMillisecond(Some(1_602_843_072_000), None),
    1513
    0
            )?;
    1514
    0
            let right_child = Interval::try_new(
    1515
                // 10.10.2020 - 10:11:12 AM
    1516
    0
                ScalarValue::TimestampMillisecond(Some(1_602_324_672_000), None),
    1517
                // 20.10.2020 - 10:11:12 AM
    1518
    0
                ScalarValue::TimestampMillisecond(Some(1_603_188_672_000), None),
    1519
    0
            )?;
    1520
    0
            let left_child = Interval::try_new(
    1521
                // 2 days in millisecond
    1522
    0
                ScalarValue::IntervalDayTime(Some(IntervalDayTime {
    1523
    0
                    days: 0,
    1524
    0
                    milliseconds: 172_800_000,
    1525
    0
                })),
    1526
                // 10 days in millisecond
    1527
    0
                ScalarValue::IntervalDayTime(Some(IntervalDayTime {
    1528
    0
                    days: 0,
    1529
    0
                    milliseconds: 864_000_000,
    1530
    0
                })),
    1531
    0
            )?;
    1532
    0
            let children = vec![&left_child, &right_child];
    1533
    0
            let result = expression
    1534
    0
                .propagate_constraints(&parent, &children)?
    1535
    0
                .unwrap();
    1536
    1537
    0
            assert_eq!(
    1538
    0
                vec![
    1539
    0
                    Interval::try_new(
    1540
                        // 2 days in millisecond
    1541
    0
                        ScalarValue::IntervalDayTime(Some(IntervalDayTime {
    1542
    0
                            days: 0,
    1543
    0
                            milliseconds: 172_800_000,
    1544
    0
                        })),
    1545
                        // 6 days
    1546
    0
                        ScalarValue::IntervalDayTime(Some(IntervalDayTime {
    1547
    0
                            days: 0,
    1548
    0
                            milliseconds: 518_400_000,
    1549
    0
                        })),
    1550
    0
                    )?,
    1551
    0
                    Interval::try_new(
    1552
                        // 10.10.2020 - 10:11:12 AM
    1553
    0
                        ScalarValue::TimestampMillisecond(Some(1_602_324_672_000), None),
    1554
                        // 14.10.2020 - 10:11:12 AM
    1555
    0
                        ScalarValue::TimestampMillisecond(Some(1_602_670_272_000), None),
    1556
    0
                    )?
    1557
                ],
    1558
                result
    1559
            );
    1560
    1561
    0
            Ok(())
    1562
    0
        }
    1563
    1564
        #[test]
    1565
    0
        fn test_propagate_comparison() -> Result<()> {
    1566
            // In the examples below:
    1567
            // `left` is unbounded: [?, ?],
    1568
            // `right` is known to be [1000,1000]
    1569
            // so `left` < `right` results in no new knowledge of `right` but knowing that `left` is now < 1000:` [?, 999]
    1570
    0
            let left = Interval::make_unbounded(&DataType::Int64)?;
    1571
    0
            let right = Interval::make(Some(1000_i64), Some(1000_i64))?;
    1572
    0
            assert_eq!(
    1573
                (Some((
    1574
    0
                    Interval::make(None, Some(999_i64))?,
    1575
    0
                    Interval::make(Some(1000_i64), Some(1000_i64))?,
    1576
                ))),
    1577
    0
                propagate_comparison(&Operator::Lt, &Interval::TRUE, &left, &right)?
    1578
            );
    1579
    1580
    0
            let left =
    1581
    0
                Interval::make_unbounded(&DataType::Timestamp(TimeUnit::Nanosecond, None))?;
    1582
    0
            let right = Interval::try_new(
    1583
    0
                ScalarValue::TimestampNanosecond(Some(1000), None),
    1584
    0
                ScalarValue::TimestampNanosecond(Some(1000), None),
    1585
    0
            )?;
    1586
    0
            assert_eq!(
    1587
                (Some((
    1588
    0
                    Interval::try_new(
    1589
    0
                        ScalarValue::try_from(&DataType::Timestamp(
    1590
    0
                            TimeUnit::Nanosecond,
    1591
    0
                            None
    1592
    0
                        ))
    1593
    0
                        .unwrap(),
    1594
    0
                        ScalarValue::TimestampNanosecond(Some(999), None),
    1595
    0
                    )?,
    1596
    0
                    Interval::try_new(
    1597
    0
                        ScalarValue::TimestampNanosecond(Some(1000), None),
    1598
    0
                        ScalarValue::TimestampNanosecond(Some(1000), None),
    1599
    0
                    )?
    1600
                ))),
    1601
    0
                propagate_comparison(&Operator::Lt, &Interval::TRUE, &left, &right)?
    1602
            );
    1603
    1604
    0
            let left = Interval::make_unbounded(&DataType::Timestamp(
    1605
    0
                TimeUnit::Nanosecond,
    1606
    0
                Some("+05:00".into()),
    1607
    0
            ))?;
    1608
    0
            let right = Interval::try_new(
    1609
    0
                ScalarValue::TimestampNanosecond(Some(1000), Some("+05:00".into())),
    1610
    0
                ScalarValue::TimestampNanosecond(Some(1000), Some("+05:00".into())),
    1611
    0
            )?;
    1612
    0
            assert_eq!(
    1613
                (Some((
    1614
    0
                    Interval::try_new(
    1615
    0
                        ScalarValue::try_from(&DataType::Timestamp(
    1616
    0
                            TimeUnit::Nanosecond,
    1617
    0
                            Some("+05:00".into()),
    1618
    0
                        ))
    1619
    0
                        .unwrap(),
    1620
    0
                        ScalarValue::TimestampNanosecond(Some(999), Some("+05:00".into())),
    1621
    0
                    )?,
    1622
    0
                    Interval::try_new(
    1623
    0
                        ScalarValue::TimestampNanosecond(Some(1000), Some("+05:00".into())),
    1624
    0
                        ScalarValue::TimestampNanosecond(Some(1000), Some("+05:00".into())),
    1625
    0
                    )?
    1626
                ))),
    1627
    0
                propagate_comparison(&Operator::Lt, &Interval::TRUE, &left, &right)?
    1628
            );
    1629
    1630
    0
            Ok(())
    1631
    0
        }
    1632
    1633
        #[test]
    1634
    0
        fn test_propagate_or() -> Result<()> {
    1635
    0
            let expr = Arc::new(BinaryExpr::new(
    1636
    0
                Arc::new(Column::new("a", 0)),
    1637
    0
                Operator::Or,
    1638
    0
                Arc::new(Column::new("b", 1)),
    1639
            ));
    1640
    0
            let parent = Interval::FALSE;
    1641
    0
            let children_set = vec![
    1642
    0
                vec![&Interval::FALSE, &Interval::TRUE_OR_FALSE],
    1643
    0
                vec![&Interval::TRUE_OR_FALSE, &Interval::FALSE],
    1644
    0
                vec![&Interval::FALSE, &Interval::FALSE],
    1645
    0
                vec![&Interval::TRUE_OR_FALSE, &Interval::TRUE_OR_FALSE],
    1646
            ];
    1647
    0
            for children in children_set {
    1648
    0
                assert_eq!(
    1649
    0
                    expr.propagate_constraints(&parent, &children)?.unwrap(),
    1650
    0
                    vec![Interval::FALSE, Interval::FALSE],
    1651
                );
    1652
            }
    1653
    1654
    0
            let parent = Interval::FALSE;
    1655
    0
            let children_set = vec![
    1656
    0
                vec![&Interval::TRUE, &Interval::TRUE_OR_FALSE],
    1657
    0
                vec![&Interval::TRUE_OR_FALSE, &Interval::TRUE],
    1658
            ];
    1659
    0
            for children in children_set {
    1660
    0
                assert_eq!(expr.propagate_constraints(&parent, &children)?, None,);
    1661
            }
    1662
    1663
    0
            let parent = Interval::TRUE;
    1664
    0
            let children = vec![&Interval::FALSE, &Interval::TRUE_OR_FALSE];
    1665
    0
            assert_eq!(
    1666
    0
                expr.propagate_constraints(&parent, &children)?.unwrap(),
    1667
    0
                vec![Interval::FALSE, Interval::TRUE]
    1668
            );
    1669
    1670
    0
            let parent = Interval::TRUE;
    1671
    0
            let children = vec![&Interval::TRUE_OR_FALSE, &Interval::TRUE_OR_FALSE];
    1672
    0
            assert_eq!(
    1673
    0
                expr.propagate_constraints(&parent, &children)?.unwrap(),
    1674
                // Empty means unchanged intervals.
    1675
    0
                vec![]
    1676
            );
    1677
    1678
    0
            Ok(())
    1679
    0
        }
    1680
    1681
        #[test]
    1682
    0
        fn test_propagate_certainly_false_and() -> Result<()> {
    1683
    0
            let expr = Arc::new(BinaryExpr::new(
    1684
    0
                Arc::new(Column::new("a", 0)),
    1685
    0
                Operator::And,
    1686
    0
                Arc::new(Column::new("b", 1)),
    1687
            ));
    1688
    0
            let parent = Interval::FALSE;
    1689
    0
            let children_and_results_set = vec![
    1690
    0
                (
    1691
    0
                    vec![&Interval::TRUE, &Interval::TRUE_OR_FALSE],
    1692
    0
                    vec![Interval::TRUE, Interval::FALSE],
    1693
    0
                ),
    1694
    0
                (
    1695
    0
                    vec![&Interval::TRUE_OR_FALSE, &Interval::TRUE],
    1696
    0
                    vec![Interval::FALSE, Interval::TRUE],
    1697
    0
                ),
    1698
    0
                (
    1699
    0
                    vec![&Interval::TRUE_OR_FALSE, &Interval::TRUE_OR_FALSE],
    1700
    0
                    // Empty means unchanged intervals.
    1701
    0
                    vec![],
    1702
    0
                ),
    1703
    0
                (vec![&Interval::FALSE, &Interval::TRUE_OR_FALSE], vec![]),
    1704
            ];
    1705
    0
            for (children, result) in children_and_results_set {
    1706
    0
                assert_eq!(
    1707
    0
                    expr.propagate_constraints(&parent, &children)?.unwrap(),
    1708
                    result
    1709
                );
    1710
            }
    1711
    1712
    0
            Ok(())
    1713
    0
        }
    1714
    }