## 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/window/window_expr.rs

    Line
    Count
    Source
    1
    // Licensed to the Apache Software Foundation (ASF) under one
    2
    // or more contributor license agreements.  See the NOTICE file
    3
    // distributed with this work for additional information
    4
    // regarding copyright ownership.  The ASF licenses this file
    5
    // to you under the Apache License, Version 2.0 (the
    6
    // "License"); you may not use this file except in compliance
    7
    // with the License.  You may obtain a copy of the License at
    8
    //
    9
    //   http://www.apache.org/licenses/LICENSE-2.0
    10
    //
    11
    // Unless required by applicable law or agreed to in writing,
    12
    // software distributed under the License is distributed on an
    13
    // "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
    14
    // KIND, either express or implied.  See the License for the
    15
    // specific language governing permissions and limitations
    16
    // under the License.
    17
    18
    use std::any::Any;
    19
    use std::fmt::Debug;
    20
    use std::ops::Range;
    21
    use std::sync::Arc;
    22
    23
    use crate::PhysicalExpr;
    24
    25
    use arrow::array::BooleanArray;
    26
    use arrow::array::{new_empty_array, Array, ArrayRef};
    27
    use arrow::compute::filter as arrow_filter;
    28
    use arrow::compute::kernels::sort::SortColumn;
    29
    use arrow::compute::SortOptions;
    30
    use arrow::datatypes::FieldRef;
    31
    use arrow::record_batch::RecordBatch;
    32
    use datafusion_common::cast::as_boolean_array;
    33
    use datafusion_common::utils::compare_rows;
    34
    use datafusion_common::{
    35
        arrow_datafusion_err, exec_datafusion_err, internal_err, DataFusionError, Result,
    36
        ScalarValue,
    37
    };
    38
    use datafusion_expr::window_state::{
    39
        PartitionBatchState, WindowAggState, WindowFrameContext, WindowFrameStateGroups,
    40
    };
    41
    use datafusion_expr::{Accumulator, PartitionEvaluator, WindowFrame, WindowFrameBound};
    42
    use datafusion_physical_expr_common::sort_expr::PhysicalSortExpr;
    43
    44
    use datafusion_physical_expr_common::utils::evaluate_expressions_to_arrays;
    45
    use indexmap::IndexMap;
    46
    47
    /// Common trait for [window function] implementations
    48
    ///
    49
    /// # Aggregate Window Expressions
    50
    ///
    51
    /// These expressions take the form
    52
    ///
    53
    /// ```text
    54
    /// OVER({ROWS | RANGE| GROUPS} BETWEEN UNBOUNDED PRECEDING AND ...)
    55
    /// ```
    56
    ///
    57
    /// For example, cumulative window frames uses `PlainAggregateWindowExpr`.
    58
    ///
    59
    /// # Non Aggregate Window Expressions
    60
    ///
    61
    /// The expressions have the form
    62
    ///
    63
    /// ```text
    64
    /// OVER({ROWS | RANGE| GROUPS} BETWEEN M {PRECEDING| FOLLOWING} AND ...)
    65
    /// ```
    66
    ///
    67
    /// For example, sliding window frames use [`SlidingAggregateWindowExpr`].
    68
    ///
    69
    /// [window function]: https://en.wikipedia.org/wiki/Window_function_(SQL)
    70
    /// [`PlainAggregateWindowExpr`]: crate::window::PlainAggregateWindowExpr
    71
    /// [`SlidingAggregateWindowExpr`]: crate::window::SlidingAggregateWindowExpr
    72
    pub trait WindowExpr: Send + Sync + Debug {
    73
        /// Returns the window expression as [`Any`] so that it can be
    74
        /// downcast to a specific implementation.
    75
        fn as_any(&self) -> &dyn Any;
    76
    77
        /// The field of the final result of this window function.
    78
        fn field(&self) -> Result<FieldRef>;
    79
    80
        /// Human readable name such as `"MIN(c2)"` or `"RANK()"`. The default
    81
        /// implementation returns placeholder text.
    82
    0
        fn name(&self) -> &str {
    83
    0
            "WindowExpr: default name"
    84
    0
        }
    85
    86
        /// Expressions that are passed to the WindowAccumulator.
    87
        /// Functions which take a single input argument, such as `sum`, return a single [`datafusion_expr::expr::Expr`],
    88
        /// others (e.g. `cov`) return many.
    89
        fn expressions(&self) -> Vec<Arc<dyn PhysicalExpr>>;
    90
    91
        /// Evaluate the window function arguments against the batch and return
    92
        /// array ref, normally the resulting `Vec` is a single element one.
    93
    0
        fn evaluate_args(&self, batch: &RecordBatch) -> Result<Vec<ArrayRef>> {
    94
    0
            evaluate_expressions_to_arrays(&self.expressions(), batch)
    95
    0
        }
    96
    97
        /// Evaluate the window function values against the batch
    98
        fn evaluate(&self, batch: &RecordBatch) -> Result<ArrayRef>;
    99
    100
        /// Evaluate the window function against the batch. This function facilitates
    101
        /// stateful, bounded-memory implementations.
    102
    0
        fn evaluate_stateful(
    103
    0
            &self,
    104
    0
            _partition_batches: &PartitionBatches,
    105
    0
            _window_agg_state: &mut PartitionWindowAggStates,
    106
    0
        ) -> Result<()> {
    107
    0
            internal_err!("evaluate_stateful is not implemented for {}", self.name())
    108
    0
        }
    109
    110
        /// Expressions that's from the window function's partition by clause, empty if absent
    111
        fn partition_by(&self) -> &[Arc<dyn PhysicalExpr>];
    112
    113
        /// Expressions that's from the window function's order by clause, empty if absent
    114
        fn order_by(&self) -> &[PhysicalSortExpr];
    115
    116
        /// Get order by columns, empty if absent
    117
    0
        fn order_by_columns(&self, batch: &RecordBatch) -> Result<Vec<SortColumn>> {
    118
    0
            self.order_by()
    119
    0
                .iter()
    120
    0
                .map(|e| e.evaluate_to_sort_column(batch))
    121
    0
                .collect()
    122
    0
        }
    123
    124
        /// Get the window frame of this [WindowExpr].
    125
        fn get_window_frame(&self) -> &Arc<WindowFrame>;
    126
    127
        /// Return a flag indicating whether this [WindowExpr] can run with
    128
        /// bounded memory.
    129
        fn uses_bounded_memory(&self) -> bool;
    130
    131
        /// Get the reverse expression of this [WindowExpr].
    132
        fn get_reverse_expr(&self) -> Option<Arc<dyn WindowExpr>>;
    133
    134
        /// Creates a new instance of the window function evaluator.
    135
        ///
    136
        /// Returns `WindowFn::Builtin` for built-in window functions (e.g., ROW_NUMBER, RANK)
    137
        /// or `WindowFn::Aggregate` for aggregate window functions (e.g., SUM, AVG).
    138
        fn create_window_fn(&self) -> Result<WindowFn>;
    139
    140
        /// Returns all expressions used in the [`WindowExpr`].
    141
        /// These expressions are (1) function arguments, (2) partition by expressions, (3) order by expressions.
    142
    0
        fn all_expressions(&self) -> WindowPhysicalExpressions {
    143
    0
            let args = self.expressions();
    144
    0
            let partition_by_exprs = self.partition_by().to_vec();
    145
    0
            let order_by_exprs = self
    146
    0
                .order_by()
    147
    0
                .iter()
    148
    0
                .map(|sort_expr| Arc::clone(&sort_expr.expr))
    149
    0
                .collect();
    150
    0
            WindowPhysicalExpressions {
    151
    0
                args,
    152
    0
                partition_by_exprs,
    153
    0
                order_by_exprs,
    154
    0
            }
    155
    0
        }
    156
    157
        /// Rewrites [`WindowExpr`], with new expressions given. The argument should be consistent
    158
        /// with the return value of the [`WindowExpr::all_expressions`] method.
    159
        /// Returns `Some(Arc<dyn WindowExpr>)` if re-write is supported, otherwise returns `None`.
    160
    0
        fn with_new_expressions(
    161
    0
            &self,
    162
    0
            _args: Vec<Arc<dyn PhysicalExpr>>,
    163
    0
            _partition_bys: Vec<Arc<dyn PhysicalExpr>>,
    164
    0
            _order_by_exprs: Vec<Arc<dyn PhysicalExpr>>,
    165
    0
        ) -> Option<Arc<dyn WindowExpr>> {
    166
    0
            None
    167
    0
        }
    168
    }
    169
    170
    /// Stores the physical expressions used inside the `WindowExpr`.
    171
    pub struct WindowPhysicalExpressions {
    172
        /// Window function arguments
    173
        pub args: Vec<Arc<dyn PhysicalExpr>>,
    174
        /// PARTITION BY expressions
    175
        pub partition_by_exprs: Vec<Arc<dyn PhysicalExpr>>,
    176
        /// ORDER BY expressions
    177
        pub order_by_exprs: Vec<Arc<dyn PhysicalExpr>>,
    178
    }
    179
    180
    /// Extension trait that adds common functionality to [`AggregateWindowExpr`]s
    181
    pub trait AggregateWindowExpr: WindowExpr {
    182
        /// Get the accumulator for the window expression. Note that distinct
    183
        /// window expressions may return distinct accumulators; e.g. sliding
    184
        /// (non-sliding) expressions will return sliding (normal) accumulators.
    185
        fn get_accumulator(&self) -> Result<Box<dyn Accumulator>>;
    186
    187
        /// Optional FILTER (WHERE ...) predicate for this window aggregate.
    188
        fn filter_expr(&self) -> Option<&Arc<dyn PhysicalExpr>>;
    189
    190
        /// Given current range and the last range, calculates the accumulator
    191
        /// result for the range of interest.
    192
        fn get_aggregate_result_inside_range(
    193
            &self,
    194
            last_range: &Range<usize>,
    195
            cur_range: &Range<usize>,
    196
            value_slice: &[ArrayRef],
    197
            accumulator: &mut Box<dyn Accumulator>,
    198
            filter_mask: Option<&BooleanArray>,
    199
        ) -> Result<ScalarValue>;
    200
    201
        /// Indicates whether this window function always produces the same result
    202
        /// for all rows in the partition.
    203
        fn is_constant_in_partition(&self) -> bool;
    204
    205
        /// Evaluates the window function against the batch.
    206
    0
        fn aggregate_evaluate(&self, batch: &RecordBatch) -> Result<ArrayRef> {
    207
    0
            let mut accumulator = self.get_accumulator()?;
    208
    0
            let mut last_range = Range { start: 0, end: 0 };
    209
    0
            let sort_options = self.order_by().iter().map(|o| o.options).collect();
    210
    0
            let mut window_frame_ctx =
    211
    0
                WindowFrameContext::new(Arc::clone(self.get_window_frame()), sort_options);
    212
    0
            self.get_result_column(
    213
    0
                &mut accumulator,
    214
    0
                batch,
    215
    0
                None,
    216
    0
                &mut last_range,
    217
    0
                &mut window_frame_ctx,
    218
                0,
    219
                false,
    220
            )
    221
    0
        }
    222
    223
        /// Statefully evaluates the window function against the batch. Maintains
    224
        /// state so that it can work incrementally over multiple chunks.
    225
    0
        fn aggregate_evaluate_stateful(
    226
    0
            &self,
    227
    0
            partition_batches: &PartitionBatches,
    228
    0
            window_agg_state: &mut PartitionWindowAggStates,
    229
    0
        ) -> Result<()> {
    230
    0
            let field = self.field()?;
    231
    0
            let out_type = field.data_type();
    232
    0
            for (partition_row, partition_batch_state) in partition_batches.iter() {
    233
    0
                if !window_agg_state.contains_key(partition_row) {
    234
    0
                    let accumulator = self.get_accumulator()?;
    235
    0
                    window_agg_state.insert(
    236
    0
                        partition_row.clone(),
    237
                        WindowState {
    238
    0
                            state: WindowAggState::new(out_type)?,
    239
    0
                            window_fn: WindowFn::Aggregate(accumulator),
    240
                        },
    241
                    );
    242
    0
                };
    243
    0
                let window_state = window_agg_state
    244
    0
                    .get_mut(partition_row)
    245
    0
                    .ok_or_else(|| exec_datafusion_err!("Cannot find state"))?;
    246
    0
                let accumulator = match &mut window_state.window_fn {
    247
    0
                    WindowFn::Aggregate(accumulator) => accumulator,
    248
    0
                    _ => unreachable!(),
    249
                };
    250
    0
                let state = &mut window_state.state;
    251
    0
                let record_batch = &partition_batch_state.record_batch;
    252
    0
                let most_recent_row = partition_batch_state.most_recent_row.as_ref();
    253
    254
                // If there is no window state context, initialize it.
    255
    0
                let window_frame_ctx = state.window_frame_ctx.get_or_insert_with(|| {
    256
    0
                    let sort_options = self.order_by().iter().map(|o| o.options).collect();
    257
    0
                    WindowFrameContext::new(Arc::clone(self.get_window_frame()), sort_options)
    258
    0
                });
    259
    0
                let out_col = self.get_result_column(
    260
    0
                    accumulator,
    261
    0
                    record_batch,
    262
    0
                    most_recent_row,
    263
                    // Start search from the last range
    264
    0
                    &mut state.window_frame_range,
    265
    0
                    window_frame_ctx,
    266
    0
                    state.last_calculated_index,
    267
    0
                    !partition_batch_state.is_end,
    268
    0
                )?;
    269
    0
                state.update(&out_col, partition_batch_state)?;
    270
            }
    271
    0
            Ok(())
    272
    0
        }
    273
    274
        /// Calculates the window expression result for the given record batch.
    275
        /// Assumes that `record_batch` belongs to a single partition.
    276
        ///
    277
        /// # Arguments
    278
        /// * `accumulator`: The accumulator to use for the calculation.
    279
        /// * `record_batch`: batch belonging to the current partition (see [`PartitionBatchState`]).
    280
        /// * `most_recent_row`: the batch that contains the most recent row, if available (see [`PartitionBatchState`]).
    281
        /// * `last_range`: The last range of rows that were processed (see [`WindowAggState`]).
    282
        /// * `window_frame_ctx`: Details about the window frame (see [`WindowFrameContext`]).
    283
        /// * `idx`: The index of the current row in the record batch.
    284
        /// * `not_end`: is the current row not the end of the partition (see [`PartitionBatchState`]).
    285
        #[allow(clippy::too_many_arguments)]
    286
    0
        fn get_result_column(
    287
    0
            &self,
    288
    0
            accumulator: &mut Box<dyn Accumulator>,
    289
    0
            record_batch: &RecordBatch,
    290
    0
            most_recent_row: Option<&RecordBatch>,
    291
    0
            last_range: &mut Range<usize>,
    292
    0
            window_frame_ctx: &mut WindowFrameContext,
    293
    0
            mut idx: usize,
    294
    0
            not_end: bool,
    295
    0
        ) -> Result<ArrayRef> {
    296
    0
            let values = self.evaluate_args(record_batch)?;
    297
    298
            // Evaluate filter mask once per record batch if present
    299
    0
            let filter_mask_arr: Option<ArrayRef> = match self.filter_expr() {
    300
    0
                Some(expr) => {
    301
    0
                    let value = expr.evaluate(record_batch)?;
    302
    0
                    Some(value.into_array(record_batch.num_rows())?)
    303
                }
    304
    0
                None => None,
    305
            };
    306
    307
            // Borrow boolean view from the owned array
    308
    0
            let filter_mask: Option<&BooleanArray> = match filter_mask_arr.as_deref() {
    309
    0
                Some(arr) => Some(as_boolean_array(arr)?),
    310
    0
                None => None,
    311
            };
    312
    313
    0
            if self.is_constant_in_partition() {
    314
    0
                if not_end {
    315
    0
                    let field = self.field()?;
    316
    0
                    let out_type = field.data_type();
    317
    0
                    return Ok(new_empty_array(out_type));
    318
    0
                }
    319
    0
                let values = if let Some(mask) = filter_mask {
    320
                    // Apply mask to all argument arrays before a single update
    321
    0
                    filter_arrays(&values, mask)?
    322
                } else {
    323
    0
                    values
    324
                };
    325
    0
                accumulator.update_batch(&values)?;
    326
    0
                let value = accumulator.evaluate()?;
    327
    0
                return value.to_array_of_size(record_batch.num_rows());
    328
    0
            }
    329
    0
            let order_bys = get_orderby_values(self.order_by_columns(record_batch)?);
    330
    0
            let most_recent_row_order_bys = most_recent_row
    331
    0
                .map(|batch| self.order_by_columns(batch))
    332
    0
                .transpose()?
    333
    0
                .map(get_orderby_values);
    334
    335
            // We iterate on each row to perform a running calculation.
    336
    0
            let length = values[0].len();
    337
    0
            let mut row_wise_results: Vec<ScalarValue> = vec![];
    338
    0
            let is_causal = self.get_window_frame().is_causal();
    339
    0
            while idx < length {
    340
                // Start search from the last_range. This squeezes searched range.
    341
    0
                let cur_range =
    342
    0
                    window_frame_ctx.calculate_range(&order_bys, last_range, length, idx)?;
    343
                // Exit if the range is non-causal and extends all the way:
    344
    0
                if cur_range.end == length
    345
    0
                    && !is_causal
    346
    0
                    && not_end
    347
    0
                    && !is_end_bound_safe(
    348
    0
                        window_frame_ctx,
    349
    0
                        &order_bys,
    350
    0
                        most_recent_row_order_bys.as_deref(),
    351
    0
                        self.order_by(),
    352
    0
                        idx,
    353
    0
                    )?
    354
                {
    355
    0
                    break;
    356
    0
                }
    357
    0
                let value = self.get_aggregate_result_inside_range(
    358
    0
                    last_range,
    359
    0
                    &cur_range,
    360
    0
                    &values,
    361
    0
                    accumulator,
    362
    0
                    filter_mask,
    363
    0
                )?;
    364
                // Update last range
    365
    0
                *last_range = cur_range;
    366
    0
                row_wise_results.push(value);
    367
    0
                idx += 1;
    368
            }
    369
    370
    0
            if row_wise_results.is_empty() {
    371
    0
                let field = self.field()?;
    372
    0
                let out_type = field.data_type();
    373
    0
                Ok(new_empty_array(out_type))
    374
            } else {
    375
    0
                ScalarValue::iter_to_array(row_wise_results)
    376
            }
    377
    0
        }
    378
    }
    379
    380
    /// Filters a single array with the provided boolean mask.
    381
    0
    pub(crate) fn filter_array(array: &ArrayRef, mask: &BooleanArray) -> Result<ArrayRef> {
    382
    0
        arrow_filter(array.as_ref(), mask)
    383
    0
            .map(|a| a as ArrayRef)
    384
    0
            .map_err(|e| arrow_datafusion_err!(e))
    385
    0
    }
    386
    387
    /// Filters a list of arrays with the provided boolean mask.
    388
    0
    pub(crate) fn filter_arrays(
    389
    0
        arrays: &[ArrayRef],
    390
    0
        mask: &BooleanArray,
    391
    0
    ) -> Result<Vec<ArrayRef>> {
    392
    0
        arrays.iter().map(|arr| filter_array(arr, mask)).collect()
    393
    0
    }
    394
    395
    /// Determines whether the end bound calculation for a window frame context is
    396
    /// safe, meaning that the end bound stays the same, regardless of future data,
    397
    /// based on the current sort expressions and ORDER BY columns. This function
    398
    /// delegates work to specific functions for each frame type.
    399
    ///
    400
    /// # Parameters
    401
    ///
    402
    /// * `window_frame_ctx`: The context of the window frame being evaluated.
    403
    /// * `order_bys`: A slice of `ArrayRef` representing the ORDER BY columns.
    404
    /// * `most_recent_order_bys`: An optional reference to the most recent ORDER BY
    405
    ///   columns.
    406
    /// * `sort_exprs`: Defines the lexicographical ordering in question.
    407
    /// * `idx`: The current index in the window frame.
    408
    ///
    409
    /// # Returns
    410
    ///
    411
    /// A `Result` which is `Ok(true)` if the end bound is safe, `Ok(false)` otherwise.
    412
    0
    pub(crate) fn is_end_bound_safe(
    413
    0
        window_frame_ctx: &WindowFrameContext,
    414
    0
        order_bys: &[ArrayRef],
    415
    0
        most_recent_order_bys: Option<&[ArrayRef]>,
    416
    0
        sort_exprs: &[PhysicalSortExpr],
    417
    0
        idx: usize,
    418
    0
    ) -> Result<bool> {
    419
    0
        if sort_exprs.is_empty() {
    420
            // Early return if no sort expressions are present:
    421
    0
            return Ok(false);
    422
    0
        };
    423
    424
    0
        match window_frame_ctx {
    425
    0
            WindowFrameContext::Rows(window_frame) => {
    426
    0
                is_end_bound_safe_for_rows(&window_frame.end_bound)
    427
            }
    428
    0
            WindowFrameContext::Range { window_frame, .. } => is_end_bound_safe_for_range(
    429
    0
                &window_frame.end_bound,
    430
    0
                &order_bys[0],
    431
    0
                most_recent_order_bys.map(|items| &items[0]),
    432
    0
                &sort_exprs[0].options,
    433
    0
                idx,
    434
            ),
    435
            WindowFrameContext::Groups {
    436
    0
                window_frame,
    437
    0
                state,
    438
    0
            } => is_end_bound_safe_for_groups(
    439
    0
                &window_frame.end_bound,
    440
    0
                state,
    441
    0
                &order_bys[0],
    442
    0
                most_recent_order_bys.map(|items| &items[0]),
    443
    0
                &sort_exprs[0].options,
    444
            ),
    445
        }
    446
    0
    }
    447
    448
    /// For row-based window frames, determines whether the end bound calculation
    449
    /// is safe, which is trivially the case for `Preceding` and `CurrentRow` bounds.
    450
    /// For 'Following' bounds, it compares the bound value to zero to ensure that
    451
    /// it doesn't extend beyond the current row.
    452
    ///
    453
    /// # Parameters
    454
    ///
    455
    /// * `end_bound`: Reference to the window frame bound in question.
    456
    ///
    457
    /// # Returns
    458
    ///
    459
    /// A `Result` indicating whether the end bound is safe for row-based window frames.
    460
    0
    fn is_end_bound_safe_for_rows(end_bound: &WindowFrameBound) -> Result<bool> {
    461
    0
        if let WindowFrameBound::Following(value) = end_bound {
    462
    0
            let zero = ScalarValue::new_zero(&value.data_type());
    463
    0
            Ok(zero.map(|zero| value.eq(&zero)).unwrap_or(false))
    464
        } else {
    465
    0
            Ok(true)
    466
        }
    467
    0
    }
    468
    469
    /// For row-based window frames, determines whether the end bound calculation
    470
    /// is safe by comparing it against specific values (zero, current row). It uses
    471
    /// the `is_row_ahead` helper function to determine if the current row is ahead
    472
    /// of the most recent row based on the ORDER BY column and sorting options.
    473
    ///
    474
    /// # Parameters
    475
    ///
    476
    /// * `end_bound`: Reference to the window frame bound in question.
    477
    /// * `orderby_col`: Reference to the column used for ordering.
    478
    /// * `most_recent_ob_col`: Optional reference to the most recent order-by column.
    479
    /// * `sort_options`: The sorting options used in the window frame.
    480
    /// * `idx`: The current index in the window frame.
    481
    ///
    482
    /// # Returns
    483
    ///
    484
    /// A `Result` indicating whether the end bound is safe for range-based window frames.
    485
    0
    fn is_end_bound_safe_for_range(
    486
    0
        end_bound: &WindowFrameBound,
    487
    0
        orderby_col: &ArrayRef,
    488
    0
        most_recent_ob_col: Option<&ArrayRef>,
    489
    0
        sort_options: &SortOptions,
    490
    0
        idx: usize,
    491
    0
    ) -> Result<bool> {
    492
    0
        match end_bound {
    493
    0
            WindowFrameBound::Preceding(value) => {
    494
    0
                let zero = ScalarValue::new_zero(&value.data_type())?;
    495
    0
                if value.eq(&zero) {
    496
    0
                    is_row_ahead(orderby_col, most_recent_ob_col, sort_options)
    497
                } else {
    498
    0
                    Ok(true)
    499
                }
    500
            }
    501
            WindowFrameBound::CurrentRow => {
    502
    0
                is_row_ahead(orderby_col, most_recent_ob_col, sort_options)
    503
            }
    504
    0
            WindowFrameBound::Following(delta) => {
    505
    0
                let Some(most_recent_ob_col) = most_recent_ob_col else {
    506
    0
                    return Ok(false);
    507
                };
    508
    0
                let most_recent_row_value =
    509
    0
                    ScalarValue::try_from_array(most_recent_ob_col, 0)?;
    510
    0
                let current_row_value = ScalarValue::try_from_array(orderby_col, idx)?;
    511
    512
    0
                if sort_options.descending {
    513
    0
                    current_row_value
    514
    0
                        .sub(delta)
    515
    0
                        .map(|value| value > most_recent_row_value)
    516
                } else {
    517
    0
                    current_row_value
    518
    0
                        .add(delta)
    519
    0
                        .map(|value| most_recent_row_value > value)
    520
                }
    521
            }
    522
        }
    523
    0
    }
    524
    525
    /// For group-based window frames, determines whether the end bound calculation
    526
    /// is safe by considering the group offset and whether the current row is ahead
    527
    /// of the most recent row in terms of sorting. It checks if the end bound is
    528
    /// within the bounds of the current group based on group end indices.
    529
    ///
    530
    /// # Parameters
    531
    ///
    532
    /// * `end_bound`: Reference to the window frame bound in question.
    533
    /// * `state`: The state of the window frame for group calculations.
    534
    /// * `orderby_col`: Reference to the column used for ordering.
    535
    /// * `most_recent_ob_col`: Optional reference to the most recent order-by column.
    536
    /// * `sort_options`: The sorting options used in the window frame.
    537
    ///
    538
    /// # Returns
    539
    ///
    540
    /// A `Result` indicating whether the end bound is safe for group-based window frames.
    541
    0
    fn is_end_bound_safe_for_groups(
    542
    0
        end_bound: &WindowFrameBound,
    543
    0
        state: &WindowFrameStateGroups,
    544
    0
        orderby_col: &ArrayRef,
    545
    0
        most_recent_ob_col: Option<&ArrayRef>,
    546
    0
        sort_options: &SortOptions,
    547
    0
    ) -> Result<bool> {
    548
    0
        match end_bound {
    549
    0
            WindowFrameBound::Preceding(value) => {
    550
    0
                let zero = ScalarValue::new_zero(&value.data_type())?;
    551
    0
                if value.eq(&zero) {
    552
    0
                    is_row_ahead(orderby_col, most_recent_ob_col, sort_options)
    553
                } else {
    554
    0
                    Ok(true)
    555
                }
    556
            }
    557
            WindowFrameBound::CurrentRow => {
    558
    0
                is_row_ahead(orderby_col, most_recent_ob_col, sort_options)
    559
            }
    560
    0
            WindowFrameBound::Following(ScalarValue::UInt64(Some(offset))) => {
    561
    0
                let delta = state.group_end_indices.len() - state.current_group_idx;
    562
    0
                if delta == (*offset as usize) + 1 {
    563
    0
                    is_row_ahead(orderby_col, most_recent_ob_col, sort_options)
    564
                } else {
    565
    0
                    Ok(false)
    566
                }
    567
            }
    568
    0
            _ => Ok(false),
    569
        }
    570
    0
    }
    571
    572
    /// This utility function checks whether `current_cols` is ahead of the `old_cols`
    573
    /// in terms of `sort_options`.
    574
    0
    fn is_row_ahead(
    575
    0
        old_col: &ArrayRef,
    576
    0
        current_col: Option<&ArrayRef>,
    577
    0
        sort_options: &SortOptions,
    578
    0
    ) -> Result<bool> {
    579
    0
        let Some(current_col) = current_col else {
    580
    0
            return Ok(false);
    581
        };
    582
    0
        if old_col.is_empty() || current_col.is_empty() {
    583
    0
            return Ok(false);
    584
    0
        }
    585
    0
        let last_value = ScalarValue::try_from_array(old_col, old_col.len() - 1)?;
    586
    0
        let current_value = ScalarValue::try_from_array(current_col, 0)?;
    587
    0
        let cmp = compare_rows(&[current_value], &[last_value], &[*sort_options])?;
    588
    0
        Ok(cmp.is_gt())
    589
    0
    }
    590
    591
    /// Get order by expression results inside `order_by_columns`.
    592
    0
    pub(crate) fn get_orderby_values(order_by_columns: Vec<SortColumn>) -> Vec<ArrayRef> {
    593
    0
        order_by_columns.into_iter().map(|s| s.values).collect()
    594
    0
    }
    595
    596
    #[derive(Debug)]
    597
    pub enum WindowFn {
    598
        Builtin(Box<dyn PartitionEvaluator>),
    599
        Aggregate(Box<dyn Accumulator>),
    600
    }
    601
    602
    /// Key for IndexMap for each unique partition
    603
    ///
    604
    /// For instance, if window frame is `OVER(PARTITION BY a,b)`,
    605
    /// PartitionKey would consist of unique `[a,b]` pairs
    606
    pub type PartitionKey = Vec<ScalarValue>;
    607
    608
    #[derive(Debug)]
    609
    pub struct WindowState {
    610
        pub state: WindowAggState,
    611
        pub window_fn: WindowFn,
    612
    }
    613
    pub type PartitionWindowAggStates = IndexMap<PartitionKey, WindowState>;
    614
    615
    /// The IndexMap (i.e. an ordered HashMap) where record batches are separated for each partition.
    616
    pub type PartitionBatches = IndexMap<PartitionKey, PartitionBatchState>;
    617
    618
    #[cfg(test)]
    619
    mod tests {
    620
        use std::sync::Arc;
    621
    622
        use crate::window::window_expr::is_row_ahead;
    623
    624
        use arrow::array::{ArrayRef, Float64Array};
    625
        use arrow::compute::SortOptions;
    626
        use datafusion_common::Result;
    627
    628
        #[test]
    629
    0
        fn test_is_row_ahead() -> Result<()> {
    630
    0
            let old_values: ArrayRef =
    631
    0
                Arc::new(Float64Array::from(vec![5.0, 7.0, 8.0, 9., 10.]));
    632
    633
    0
            let new_values1: ArrayRef = Arc::new(Float64Array::from(vec![11.0]));
    634
    0
            let new_values2: ArrayRef = Arc::new(Float64Array::from(vec![10.0]));
    635
    636
    0
            assert!(is_row_ahead(
    637
    0
                &old_values,
    638
    0
                Some(&new_values1),
    639
    0
                &SortOptions {
    640
    0
                    descending: false,
    641
    0
                    nulls_first: false
    642
    0
                }
    643
    0
            )?);
    644
    0
            assert!(!is_row_ahead(
    645
    0
                &old_values,
    646
    0
                Some(&new_values2),
    647
    0
                &SortOptions {
    648
    0
                    descending: false,
    649
    0
                    nulls_first: false
    650
    0
                }
    651
    0
            )?);
    652
    653
    0
            Ok(())
    654
    0
        }
    655
    }