## 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-common/src/datum.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 arrow::array::BooleanArray;
    19
    use arrow::array::{make_comparator, ArrayRef, Datum};
    20
    use arrow::buffer::NullBuffer;
    21
    use arrow::compute::kernels::cmp::{
    22
        distinct, eq, gt, gt_eq, lt, lt_eq, neq, not_distinct,
    23
    };
    24
    use arrow::compute::{ilike, like, nilike, nlike, SortOptions};
    25
    use arrow::error::ArrowError;
    26
    use datafusion_common::DataFusionError;
    27
    use datafusion_common::{arrow_datafusion_err, assert_or_internal_err, internal_err};
    28
    use datafusion_common::{Result, ScalarValue};
    29
    use datafusion_expr_common::columnar_value::ColumnarValue;
    30
    use datafusion_expr_common::operator::Operator;
    31
    use std::sync::Arc;
    32
    33
    /// Applies a binary [`Datum`] kernel `f` to `lhs` and `rhs`
    34
    ///
    35
    /// This maps arrow-rs' [`Datum`] kernels to DataFusion's [`ColumnarValue`] abstraction
    36
    0
    pub fn apply(
    37
    0
        lhs: &ColumnarValue,
    38
    0
        rhs: &ColumnarValue,
    39
    0
        f: impl Fn(&dyn Datum, &dyn Datum) -> Result<ArrayRef, ArrowError>,
    40
    0
    ) -> Result<ColumnarValue> {
    41
    0
        match (&lhs, &rhs) {
    42
    0
            (ColumnarValue::Array(left), ColumnarValue::Array(right)) => {
    43
    0
                Ok(ColumnarValue::Array(f(&left.as_ref(), &right.as_ref())?))
    44
            }
    45
    0
            (ColumnarValue::Scalar(left), ColumnarValue::Array(right)) => Ok(
    46
    0
                ColumnarValue::Array(f(&left.to_scalar()?, &right.as_ref())?),
    47
            ),
    48
    0
            (ColumnarValue::Array(left), ColumnarValue::Scalar(right)) => Ok(
    49
    0
                ColumnarValue::Array(f(&left.as_ref(), &right.to_scalar()?)?),
    50
            ),
    51
    0
            (ColumnarValue::Scalar(left), ColumnarValue::Scalar(right)) => {
    52
    0
                let array = f(&left.to_scalar()?, &right.to_scalar()?)?;
    53
    0
                let scalar = ScalarValue::try_from_array(array.as_ref(), 0)?;
    54
    0
                Ok(ColumnarValue::Scalar(scalar))
    55
            }
    56
        }
    57
    0
    }
    58
    59
    /// Applies a binary [`Datum`] comparison operator `op` to `lhs` and `rhs`
    60
    0
    pub fn apply_cmp(
    61
    0
        op: Operator,
    62
    0
        lhs: &ColumnarValue,
    63
    0
        rhs: &ColumnarValue,
    64
    0
    ) -> Result<ColumnarValue> {
    65
    0
        if lhs.data_type().is_nested() {
    66
    0
            apply_cmp_for_nested(op, lhs, rhs)
    67
        } else {
    68
    0
            let f = match op {
    69
    0
                Operator::Eq => eq,
    70
    0
                Operator::NotEq => neq,
    71
    0
                Operator::Lt => lt,
    72
    0
                Operator::LtEq => lt_eq,
    73
    0
                Operator::Gt => gt,
    74
    0
                Operator::GtEq => gt_eq,
    75
    0
                Operator::IsDistinctFrom => distinct,
    76
    0
                Operator::IsNotDistinctFrom => not_distinct,
    77
    78
    0
                Operator::LikeMatch => like,
    79
    0
                Operator::ILikeMatch => ilike,
    80
    0
                Operator::NotLikeMatch => nlike,
    81
    0
                Operator::NotILikeMatch => nilike,
    82
    83
                _ => {
    84
    0
                    return internal_err!("Invalid compare operator: {}", op);
    85
                }
    86
            };
    87
    88
    0
            apply(lhs, rhs, |l, r| Ok(Arc::new(f(l, r)?)))
    89
        }
    90
    0
    }
    91
    92
    /// Applies a binary [`Datum`] comparison operator `op` to `lhs` and `rhs` for nested type like
    93
    /// List, FixedSizeList, LargeList, Struct, Union, Map, or a dictionary of a nested type
    94
    0
    pub fn apply_cmp_for_nested(
    95
    0
        op: Operator,
    96
    0
        lhs: &ColumnarValue,
    97
    0
        rhs: &ColumnarValue,
    98
    0
    ) -> Result<ColumnarValue> {
    99
    0
        let left_data_type = lhs.data_type();
    100
    0
        let right_data_type = rhs.data_type();
    101
    102
    0
        assert_or_internal_err!(
    103
    0
            matches!(
    104
    0
                op,
    105
                Operator::Eq
    106
                    | Operator::NotEq
    107
                    | Operator::Lt
    108
                    | Operator::Gt
    109
                    | Operator::LtEq
    110
                    | Operator::GtEq
    111
                    | Operator::IsDistinctFrom
    112
                    | Operator::IsNotDistinctFrom
    113
    0
            ) && left_data_type.equals_datatype(&right_data_type),
    114
    0
            "invalid operator or data type mismatch for nested data, op {op} left {left_data_type}, right {right_data_type}",
    115
        );
    116
    117
    0
        apply(lhs, rhs, |l, r| {
    118
    0
            Ok(Arc::new(compare_op_for_nested(op, l, r)?))
    119
    0
        })
    120
    0
    }
    121
    122
    /// Compare with eq with either nested or non-nested
    123
    0
    pub fn compare_with_eq(
    124
    0
        lhs: &dyn Datum,
    125
    0
        rhs: &dyn Datum,
    126
    0
        is_nested: bool,
    127
    0
    ) -> Result<BooleanArray> {
    128
    0
        if is_nested {
    129
    0
            compare_op_for_nested(Operator::Eq, lhs, rhs)
    130
        } else {
    131
    0
            eq(lhs, rhs).map_err(|e| arrow_datafusion_err!(e))
    132
        }
    133
    0
    }
    134
    135
    /// Compare on nested type List, Struct, and so on
    136
    0
    pub fn compare_op_for_nested(
    137
    0
        op: Operator,
    138
    0
        lhs: &dyn Datum,
    139
    0
        rhs: &dyn Datum,
    140
    0
    ) -> Result<BooleanArray> {
    141
    0
        let (l, is_l_scalar) = lhs.get();
    142
    0
        let (r, is_r_scalar) = rhs.get();
    143
    0
        let l_len = l.len();
    144
    0
        let r_len = r.len();
    145
    146
    0
        assert_or_internal_err!(l_len == r_len || is_l_scalar || is_r_scalar, "len mismatch");
    147
    148
    0
        let len = match is_l_scalar {
    149
    0
            true => r_len,
    150
    0
            false => l_len,
    151
        };
    152
    153
        // fast path, if compare with one null and operator is not 'distinct', then we can return null array directly
    154
    0
        if !matches!(op, Operator::IsDistinctFrom | Operator::IsNotDistinctFrom)
    155
    0
            && (is_l_scalar && l.null_count() == 1 || is_r_scalar && r.null_count() == 1)
    156
        {
    157
    0
            return Ok(BooleanArray::new_null(len));
    158
    0
        }
    159
    160
        // TODO: make SortOptions configurable
    161
        // we choose the default behaviour from arrow-rs which has null-first that follow spark's behaviour
    162
    0
        let cmp = make_comparator(l, r, SortOptions::default())?;
    163
    164
    0
        let cmp_with_op = |i, j| match op {
    165
    0
            Operator::Eq | Operator::IsNotDistinctFrom => cmp(i, j).is_eq(),
    166
    0
            Operator::Lt => cmp(i, j).is_lt(),
    167
    0
            Operator::Gt => cmp(i, j).is_gt(),
    168
    0
            Operator::LtEq => !cmp(i, j).is_gt(),
    169
    0
            Operator::GtEq => !cmp(i, j).is_lt(),
    170
    0
            Operator::NotEq | Operator::IsDistinctFrom => !cmp(i, j).is_eq(),
    171
    0
            _ => unreachable!("unexpected operator found"),
    172
    0
        };
    173
    174
    0
        let values = match (is_l_scalar, is_r_scalar) {
    175
    0
            (false, false) => (0..len).map(|i| cmp_with_op(i, i)).collect(),
    176
    0
            (true, false) => (0..len).map(|i| cmp_with_op(0, i)).collect(),
    177
    0
            (false, true) => (0..len).map(|i| cmp_with_op(i, 0)).collect(),
    178
    0
            (true, true) => std::iter::once(cmp_with_op(0, 0)).collect(),
    179
        };
    180
    181
        // Distinct understand how to compare with NULL
    182
        // i.e NULL is distinct from NULL -> false
    183
    0
        if matches!(op, Operator::IsDistinctFrom | Operator::IsNotDistinctFrom) {
    184
    0
            Ok(BooleanArray::new(values, None))
    185
        } else {
    186
            // If one of the side is NULL, we return NULL
    187
            // i.e. NULL eq NULL -> NULL
    188
            // For nested comparisons, we need to ensure the null buffer matches the result length
    189
    0
            let nulls = match (is_l_scalar, is_r_scalar) {
    190
    0
                (false, false) | (true, true) => NullBuffer::union(l.nulls(), r.nulls()),
    191
                (true, false) => {
    192
                    // When left is null-scalar and right is array, expand left nulls to match result length
    193
    0
                    match l.nulls().filter(|nulls| !nulls.is_valid(0)) {
    194
    0
                        Some(_) => Some(NullBuffer::new_null(len)), // Left scalar is null
    195
    0
                        None => r.nulls().cloned(),                 // Left scalar is non-null
    196
                    }
    197
                }
    198
                (false, true) => {
    199
                    // When right is null-scalar and left is array, expand right nulls to match result length
    200
    0
                    match r.nulls().filter(|nulls| !nulls.is_valid(0)) {
    201
    0
                        Some(_) => Some(NullBuffer::new_null(len)), // Right scalar is null
    202
    0
                        None => l.nulls().cloned(), // Right scalar is non-null
    203
                    }
    204
                }
    205
            };
    206
    0
            Ok(BooleanArray::new(values, nulls))
    207
        }
    208
    0
    }