## Coverage Report

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

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

    /Users/andrewlamb/Software/datafusion/datafusion/physical-expr/src/expressions/in_list.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
    //! Implementation of `InList` expressions: [`InListExpr`]
    19
    20
    use std::any::Any;
    21
    use std::fmt::Debug;
    22
    use std::hash::{Hash, Hasher};
    23
    use std::sync::Arc;
    24
    25
    use crate::physical_expr::physical_exprs_bag_equal;
    26
    use crate::PhysicalExpr;
    27
    28
    use arrow::array::*;
    29
    use arrow::buffer::{BooleanBuffer, NullBuffer};
    30
    use arrow::compute::kernels::boolean::{not, or_kleene};
    31
    use arrow::compute::{take, SortOptions};
    32
    use arrow::datatypes::*;
    33
    use arrow::util::bit_iterator::BitIndexIterator;
    34
    use datafusion_common::hash_utils::with_hashes;
    35
    use datafusion_common::{
    36
        assert_or_internal_err, exec_datafusion_err, exec_err, DFSchema, DataFusionError,
    37
        HashSet, Result, ScalarValue,
    38
    };
    39
    use datafusion_expr::{expr_vec_fmt, ColumnarValue};
    40
    41
    use ahash::RandomState;
    42
    use datafusion_common::HashMap;
    43
    use hashbrown::hash_map::RawEntryMut;
    44
    45
    /// Trait for InList static filters
    46
    trait StaticFilter {
    47
        fn null_count(&self) -> usize;
    48
    49
        /// Checks if values in `v` are contained in the filter
    50
        fn contains(&self, v: &dyn Array, negated: bool) -> Result<BooleanArray>;
    51
    }
    52
    53
    /// InList
    54
    pub struct InListExpr {
    55
        expr: Arc<dyn PhysicalExpr>,
    56
        list: Vec<Arc<dyn PhysicalExpr>>,
    57
        negated: bool,
    58
        static_filter: Option<Arc<dyn StaticFilter + Send + Sync>>,
    59
    }
    60
    61
    impl Debug for InListExpr {
    62
    0
        fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
    63
    0
            f.debug_struct("InListExpr")
    64
    0
                .field("expr", &self.expr)
    65
    0
                .field("list", &self.list)
    66
    0
                .field("negated", &self.negated)
    67
    0
                .finish()
    68
    0
        }
    69
    }
    70
    71
    /// Static filter for InList that stores the array and hash set for O(1) lookups
    72
    #[derive(Debug, Clone)]
    73
    struct ArrayStaticFilter {
    74
        in_array: ArrayRef,
    75
        state: RandomState,
    76
        /// Used to provide a lookup from value to in list index
    77
        ///
    78
        /// Note: usize::hash is not used, instead the raw entry
    79
        /// API is used to store entries w.r.t their value
    80
        map: HashMap<usize, (), ()>,
    81
    }
    82
    83
    impl StaticFilter for ArrayStaticFilter {
    84
    0
        fn null_count(&self) -> usize {
    85
    0
            self.in_array.null_count()
    86
    0
        }
    87
    88
        /// Checks if values in `v` are contained in the `in_array` using this hash set for lookup.
    89
    96
        fn contains(&self, v: &dyn Array, negated: bool) -> Result<BooleanArray> {
    90
            // Null type comparisons always return null (SQL three-valued logic)
    91
    96
            if v.data_type() == &DataType::Null
    92
    95
                || self.in_array.data_type() == &DataType::Null
    93
            {
    94
                // return Ok(BooleanArray::new(vec![None; v.len()]));
    95
    2
                let nulls = NullBuffer::new_null(v.len());
    96
    2
                return Ok(BooleanArray::new(
    97
    2
                    BooleanBuffer::new_unset(v.len()),
    98
    2
                    Some(nulls),
    99
    2
                ));
    100
    94
            }
    101
    102
    94
            downcast_dictionary_array! {
    103
                v => {
    104
    0
                    let values_contains = self.contains(v.values().as_ref(), negated)?;
    105
    0
                    let result = take(&values_contains, v.keys(), None)?;
    106
    0
                    return Ok(downcast_array(result.as_ref()))
    107
                }
    108
    86
                _ => {}
    109
            }
    110
    111
    86
            let needle_nulls = v.logical_nulls();
    112
    86
            let needle_nulls = needle_nulls.as_ref();
    113
    86
            let haystack_has_nulls = self.in_array.null_count() != 0;
    114
    115
    86
            with_hashes([v], &self.state, |hashes| {
    116
    86
                let cmp = make_comparator(v, &self.in_array, SortOptions::default())?0;
    117
    86
                Ok((0..v.len())
    118
    229
                    .map86(|i| {
    119
                        // SQL three-valued logic: null IN (...) is always null
    120
    229
                        if needle_nulls.is_some_and(|nulls| nulls178.is_null178(i178)) {
    121
    56
                            return None;
    122
    173
                        }
    123
    124
    173
                        let hash = hashes[i];
    125
    173
                        let contains = self
    126
    173
                            .map
    127
    173
                            .raw_entry()
    128
    173
                            .from_hash(hash, |idx| cmp86(i, *idx).is_eq86())
    129
    173
                            .is_some();
    130
    131
    88
                        match contains {
    132
    85
                            true => Some(!negated),
    133
    34
                            false if haystack_has_nulls => None,
    134
    54
                            false => Some(negated),
    135
                        }
    136
    229
                    })
    137
    86
                    .collect())
    138
    86
            })
    139
    96
        }
    140
    }
    141
    142
    158
    fn instantiate_static_filter(
    143
    158
        in_array: ArrayRef,
    144
    158
    ) -> Result<Arc<dyn StaticFilter + Send + Sync>> {
    145
    158
        match in_array.data_type() {
    146
            // Integer primitive types
    147
    4
            DataType::Int8 => Ok(Arc::new(Int8StaticFilter::try_new(&in_array)?0)),
    148
    4
            DataType::Int16 => Ok(Arc::new(Int16StaticFilter::try_new(&in_array)?0)),
    149
    24
            DataType::Int32 => Ok(Arc::new(Int32StaticFilter::try_new(&in_array)?0)),
    150
    17
            DataType::Int64 => Ok(Arc::new(Int64StaticFilter::try_new(&in_array)?0)),
    151
    4
            DataType::UInt8 => Ok(Arc::new(UInt8StaticFilter::try_new(&in_array)?0)),
    152
    4
            DataType::UInt16 => Ok(Arc::new(UInt16StaticFilter::try_new(&in_array)?0)),
    153
    4
            DataType::UInt32 => Ok(Arc::new(UInt32StaticFilter::try_new(&in_array)?0)),
    154
    4
            DataType::UInt64 => Ok(Arc::new(UInt64StaticFilter::try_new(&in_array)?0)),
    155
            _ => {
    156
                /* fall through to generic implementation for unsupported types (Float32/Float64, Struct, etc.) */
    157
    93
                Ok(Arc::new(ArrayStaticFilter::try_new(in_array)?0))
    158
            }
    159
        }
    160
    158
    }
    161
    162
    impl ArrayStaticFilter {
    163
        /// Computes a [`StaticFilter`] for the provided [`Array`] if there
    164
        /// are nulls present or there are more than the configured number of
    165
        /// elements.
    166
        ///
    167
        /// Note: This is split into a separate function as higher-rank trait bounds currently
    168
        /// cause type inference to misbehave
    169
    97
        fn try_new(in_array: ArrayRef) -> Result<ArrayStaticFilter> {
    170
            // Null type has no natural order - return empty hash set
    171
    97
            if in_array.data_type() == &DataType::Null {
    172
    2
                return Ok(ArrayStaticFilter {
    173
    2
                    in_array,
    174
    2
                    state: RandomState::new(),
    175
    2
                    map: HashMap::with_hasher(()),
    176
    2
                });
    177
    95
            }
    178
    179
    95
            let state = RandomState::new();
    180
    95
            let mut map: HashMap<usize, (), ()> = HashMap::with_hasher(());
    181
    182
    95
            with_hashes([&in_array], &state, |hashes| -> Result<()> {
    183
    95
                let cmp = make_comparator(&in_array, &in_array, SortOptions::default())?0;
    184
    185
    572
                let insert_value95 = |idx| {
    186
    572
                    let hash = hashes[idx];
    187
    572
                    if let RawEntryMut::Vacant(v571) = map
    188
    572
                        .raw_entry_mut()
    189
    572
                        .from_hash(hash, |x| cmp18(*x, idx).is_eq18())
    190
                    {
    191
    571
                        v.insert_with_hasher(hash, idx, (), |x| hashes[*x440]);
    192
    1
                    }
    193
    572
                };
    194
    195
    95
                match in_array.nulls() {
    196
    34
                    Some(nulls) => {
    197
    34
                        BitIndexIterator::new(nulls.validity(), nulls.offset(), nulls.len())
    198
    34
                            .for_each(insert_value)
    199
                    }
    200
    61
                    None => (0..in_array.len()).for_each(insert_value),
    201
                }
    202
    203
    95
                Ok(())
    204
    95
            })?0;
    205
    206
    95
            Ok(Self {
    207
    95
                in_array,
    208
    95
                state,
    209
    95
                map,
    210
    95
            })
    211
    97
        }
    212
    }
    213
    214
    // Macro to generate specialized StaticFilter implementations for primitive types
    215
    macro_rules! primitive_static_filter {
    216
        ($Name:ident, $ArrowType:ty) => {
    217
            struct $Name {
    218
                null_count: usize,
    219
                values: HashSet<<$ArrowType as ArrowPrimitiveType>::Native>,
    220
            }
    221
    222
            impl $Name {
    223
    65
                fn try_new(in_array: &ArrayRef) -> Result<Self> {
    224
    65
                    let in_array = in_array
    225
    65
                        .as_primitive_opt::<$ArrowType>()
    226
    65
                        .ok_or_else(|| exec_datafusion_err!0("Failed to downcast an array to a '{}' array"0, stringify!($ArrowType)))?0;
    227
    228
    65
                    let mut values = HashSet::with_capacity(in_array.len());
    229
    65
                    let null_count = in_array.null_count();
    230
    231
    119
                    for v in in_array65.iter65().flatten65() {
    232
    119
                        values.insert(v);
    233
    119
                    }
    234
    235
    65
                    Ok(Self { null_count, values })
    236
    65
                }
    237
            }
    238
    239
            impl StaticFilter for $Name {
    240
    2
                fn null_count(&self) -> usize {
    241
    2
                    self.null_count
    242
    2
                }
    243
    244
    57
                fn contains(&self, v: &dyn Array, negated: bool) -> Result<BooleanArray> {
    245
                    // Handle dictionary arrays by recursing on the values
    246
    57
                    downcast_dictionary_array! {
    247
                        v => {
    248
    0
                            let values_contains = self.contains(v.values().as_ref(), negated)?;
    249
    0
                            let result = take(&values_contains, v.keys(), None)?;
    250
    0
                            return Ok(downcast_array(result.as_ref()))
    251
                        }
    252
    57
                        _ => {}
    253
                    }
    254
    255
    57
                    let v = v
    256
    57
                        .as_primitive_opt::<$ArrowType>()
    257
    57
                        .ok_or_else(|| exec_datafusion_err!0("Failed to downcast an array to a '{}' array"0, stringify!($ArrowType)))?0;
    258
    259
    57
                    let haystack_has_nulls = self.null_count > 0;
    260
    261
    57
                    let result = match (v.null_count() > 0, haystack_has_nulls, negated) {
    262
                        (true, _, false) | (false, true, false) => {
    263
                            // Either needle or haystack has nulls, not negated
    264
    78
                            BooleanArray::from_iter30(v30.iter30().map30(|value| {
    265
    78
                                match value {
    266
                                    // SQL three-valued logic: null IN (...) is always null
    267
    27
                                    None => None,
    268
    51
                                    Some(v) => {
    269
    51
                                        if self.values.contains(&v) {
    270
    24
                                            Some(true)
    271
    27
                                        } else if haystack_has_nulls {
    272
                                            // value not in set, but set has nulls -> null
    273
    16
                                            None
    274
                                        } else {
    275
    11
                                            Some(false)
    276
                                        }
    277
                                    }
    278
                                }
    279
    78
                            }))
    280
                        }
    281
                        (true, _, true) | (false, true, true) => {
    282
                            // Either needle or haystack has nulls, negated
    283
    64
                            BooleanArray::from_iter22(v22.iter22().map22(|value| {
    284
    64
                                match value {
    285
                                    // SQL three-valued logic: null NOT IN (...) is always null
    286
    22
                                    None => None,
    287
    42
                                    Some(v) => {
    288
    42
                                        if self.values.contains(&v) {
    289
    19
                                            Some(false)
    290
    23
                                        } else if haystack_has_nulls {
    291
                                            // value not in set, but set has nulls -> null
    292
    13
                                            None
    293
                                        } else {
    294
    10
                                            Some(true)
    295
                                        }
    296
                                    }
    297
                                }
    298
    64
                            }))
    299
                        }
    300
                        (false, false, false) => {
    301
                            // no nulls anywhere, not negated
    302
    5
                            let values = v.values();
    303
    5
                            let mut builder = BooleanBufferBuilder::new(values.len());
    304
    7
                            for value in values5.iter5() {
    305
    7
                                builder.append(self.values.contains(value));
    306
    7
                            }
    307
    5
                            BooleanArray::new(builder.finish(), None)
    308
                        }
    309
                        (false, false, true) => {
    310
    0
                            let values = v.values();
    311
    0
                            let mut builder = BooleanBufferBuilder::new(values.len());
    312
    0
                            for value in values.iter() {
    313
    0
                                builder.append(!self.values.contains(value));
    314
    0
                            }
    315
    0
                            BooleanArray::new(builder.finish(), None)
    316
                        }
    317
                    };
    318
    57
                    Ok(result)
    319
    57
                }
    320
            }
    321
        };
    322
    }
    323
    324
    // Generate specialized filters for all integer primitive types
    325
    // Note: Float32 and Float64 are excluded because they don't implement Hash/Eq due to NaN
    326
    primitive_static_filter!(Int8StaticFilter, Int8Type);
    327
    primitive_static_filter!(Int16StaticFilter, Int16Type);
    328
    primitive_static_filter!(Int32StaticFilter, Int32Type);
    329
    primitive_static_filter!(Int64StaticFilter, Int64Type);
    330
    primitive_static_filter!(UInt8StaticFilter, UInt8Type);
    331
    primitive_static_filter!(UInt16StaticFilter, UInt16Type);
    332
    primitive_static_filter!(UInt32StaticFilter, UInt32Type);
    333
    primitive_static_filter!(UInt64StaticFilter, UInt64Type);
    334
    335
    /// Evaluates the list of expressions into an array, flattening any dictionaries
    336
    171
    fn evaluate_list(
    337
    171
        list: &[Arc<dyn PhysicalExpr>],
    338
    171
        batch: &RecordBatch,
    339
    171
    ) -> Result<ArrayRef> {
    340
    171
        let scalars159 = list
    341
    171
            .iter()
    342
    779
            .map171(|expr| {
    343
    779
                expr.evaluate(batch).and_then(|r| match r767 {
    344
                    ColumnarValue::Array(_) => {
    345
    12
                        exec_err!("InList expression must evaluate to a scalar")
    346
                    }
    347
                    // Flatten dictionary values
    348
    12
                    ColumnarValue::Scalar(ScalarValue::Dictionary(_, v)) => Ok(*v),
    349
    755
                    ColumnarValue::Scalar(s) => Ok(s),
    350
    779
                })
    351
    779
            })
    352
    171
            .collect::<Result<Vec<_>>>()?12;
    353
    354
    159
        ScalarValue::iter_to_array(scalars)
    355
    171
    }
    356
    357
    /// Try to evaluate a list of expressions as constants.
    358
    ///
    359
    /// Returns an ArrayRef if all expressions are constants (can be evaluated on an
    360
    /// empty RecordBatch), otherwise returns an error. This is used to detect when
    361
    /// a list contains only literals, casts of literals, or other constant expressions.
    362
    171
    fn try_evaluate_constant_list(
    363
    171
        list: &[Arc<dyn PhysicalExpr>],
    364
    171
        schema: &Schema,
    365
    171
    ) -> Result<ArrayRef> {
    366
    171
        let batch = RecordBatch::new_empty(Arc::new(schema.clone()));
    367
    171
        evaluate_list(list, &batch)
    368
    171
    }
    369
    370
    impl InListExpr {
    371
        /// Create a new InList expression
    372
    172
        fn new(
    373
    172
            expr: Arc<dyn PhysicalExpr>,
    374
    172
            list: Vec<Arc<dyn PhysicalExpr>>,
    375
    172
            negated: bool,
    376
    172
            static_filter: Option<Arc<dyn StaticFilter + Send + Sync>>,
    377
    172
        ) -> Self {
    378
    172
            Self {
    379
    172
                expr,
    380
    172
                list,
    381
    172
                negated,
    382
    172
                static_filter,
    383
    172
            }
    384
    172
        }
    385
    386
        /// Input expression
    387
    0
        pub fn expr(&self) -> &Arc<dyn PhysicalExpr> {
    388
    0
            &self.expr
    389
    0
        }
    390
    391
        /// List to search in
    392
    0
        pub fn list(&self) -> &[Arc<dyn PhysicalExpr>] {
    393
    0
            &self.list
    394
    0
        }
    395
    396
        /// Is this negated e.g. NOT IN LIST
    397
    0
        pub fn negated(&self) -> bool {
    398
    0
            self.negated
    399
    0
        }
    400
    401
        /// Create a new InList expression directly from an array, bypassing expression evaluation.
    402
        ///
    403
        /// This is more efficient than `in_list()` when you already have the list as an array,
    404
        /// as it avoids the conversion: `ArrayRef -> Vec<PhysicalExpr> -> ArrayRef -> StaticFilter`.
    405
        /// Instead it goes directly: `ArrayRef -> StaticFilter`.
    406
        ///
    407
        /// The `list` field will be empty when using this constructor, as the array is stored
    408
        /// directly in the static filter.
    409
        ///
    410
        /// This does not make the expression any more performant at runtime, but it does make it slightly
    411
        /// cheaper to build.
    412
    3
        pub fn try_new_from_array(
    413
    3
            expr: Arc<dyn PhysicalExpr>,
    414
    3
            array: ArrayRef,
    415
    3
            negated: bool,
    416
    3
        ) -> Result<Self> {
    417
    3
            let list = (0..array.len())
    418
    10
                .map3(|i| {
    419
    10
                    let scalar = ScalarValue::try_from_array(array.as_ref(), i)?0;
    420
    10
                    Ok(crate::expressions::lit(scalar) as Arc<dyn PhysicalExpr>)
    421
    10
                })
    422
    3
                .collect::<Result<Vec<_>>>()?0;
    423
    3
            Ok(Self::new(
    424
    3
                expr,
    425
    3
                list,
    426
    3
                negated,
    427
    3
                Some(instantiate_static_filter(array)?0),
    428
            ))
    429
    3
        }
    430
    }
    431
    impl std::fmt::Display for InListExpr {
    432
    5
        fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
    433
    5
            let list = expr_vec_fmt!(self.list);
    434
    435
    5
            if self.negated {
    436
    2
                if self.static_filter.is_some() {
    437
    2
                    write!(f, "{} NOT IN (SET) ([{list}])", self.expr)
    438
                } else {
    439
    0
                    write!(f, "{} NOT IN ([{list}])", self.expr)
    440
                }
    441
    3
            } else if self.static_filter.is_some() {
    442
    2
                write!(f, "{} IN (SET) ([{list}])", self.expr)
    443
            } else {
    444
    1
                write!(f, "{} IN ([{list}])", self.expr)
    445
            }
    446
    5
        }
    447
    }
    448
    449
    impl PhysicalExpr for InListExpr {
    450
        /// Return a reference to Any that can be used for downcasting
    451
    2
        fn as_any(&self) -> &dyn Any {
    452
    2
            self
    453
    2
        }
    454
    455
    0
        fn data_type(&self, _input_schema: &Schema) -> Result<DataType> {
    456
    0
            Ok(DataType::Boolean)
    457
    0
        }
    458
    459
    7
        fn nullable(&self, input_schema: &Schema) -> Result<bool> {
    460
    7
            if self.expr.nullable(input_schema)?0 {
    461
    3
                return Ok(true);
    462
    4
            }
    463
    464
    4
            if let Some(static_filter2) = &self.static_filter {
    465
    2
                Ok(static_filter.null_count() > 0)
    466
            } else {
    467
    4
                for expr3 in &self.list {
    468
    3
                    if expr.nullable(input_schema)?0 {
    469
    1
                        return Ok(true);
    470
    2
                    }
    471
                }
    472
    1
                Ok(false)
    473
            }
    474
    7
        }
    475
    476
    158
        fn evaluate(&self, batch: &RecordBatch) -> Result<ColumnarValue> {
    477
    158
            let num_rows = batch.num_rows();
    478
    158
            let value = self.expr.evaluate(batch)?0;
    479
    158
            let r154 = match &self.static_filter {
    480
    148
                Some(filter) => {
    481
    148
                    match value {
    482
    123
                        ColumnarValue::Array(array) => {
    483
    123
                            filter.contains(&array, self.negated)?0
    484
                        }
    485
    25
                        ColumnarValue::Scalar(scalar) => {
    486
    25
                            if scalar.is_null() {
    487
                                // SQL three-valued logic: null IN (...) is always null
    488
                                // The code below would handle this correctly but this is a faster path
    489
    4
                                let nulls = NullBuffer::new_null(num_rows);
    490
    4
                                return Ok(ColumnarValue::Array(Arc::new(
    491
    4
                                    BooleanArray::new(
    492
    4
                                        BooleanBuffer::new_unset(num_rows),
    493
    4
                                        Some(nulls),
    494
    4
                                    ),
    495
    4
                                )));
    496
    21
                            }
    497
                            // Use a 1 row array to avoid code duplication/branching
    498
                            // Since all we do is compute hash and lookup this should be efficient enough
    499
    21
                            let array = scalar.to_array()?0;
    500
    21
                            let result_array =
    501
    21
                                filter.contains(array.as_ref(), self.negated)?0;
    502
                            // Broadcast the single result to all rows
    503
                            // Must check is_null() to preserve NULL values (SQL three-valued logic)
    504
    21
                            if result_array.is_null(0) {
    505
    4
                                let nulls = NullBuffer::new_null(num_rows);
    506
    4
                                BooleanArray::new(
    507
    4
                                    BooleanBuffer::new_unset(num_rows),
    508
    4
                                    Some(nulls),
    509
                                )
    510
    17
                            } else if result_array.value(0) {
    511
    6
                                BooleanArray::new(BooleanBuffer::new_set(num_rows), None)
    512
                            } else {
    513
    11
                                BooleanArray::new(BooleanBuffer::new_unset(num_rows), None)
    514
                            }
    515
                        }
    516
                    }
    517
                }
    518
                None => {
    519
                    // No static filter: iterate through each expression, compare, and OR results
    520
    10
                    let value = value.into_array(num_rows)?0;
    521
    20
                    let found10 = self.list.iter()10.map10(|expr| expr.evaluate(batch)).try_fold10(
    522
    10
                        BooleanArray::new(BooleanBuffer::new_unset(num_rows), None),
    523
    20
                        |result, expr| -> Result<BooleanArray> {
    524
    20
                            let rhs = match expr?0 {
    525
    10
                                ColumnarValue::Array(array) => {
    526
    10
                                    let cmp = make_comparator(
    527
    10
                                        value.as_ref(),
    528
    10
                                        array.as_ref(),
    529
    10
                                        SortOptions::default(),
    530
    0
                                    )?;
    531
    10
                                    (0..num_rows)
    532
    26
                                        .map10(|i| {
    533
    26
                                            if value.is_null(i) || array25.is_null25(i25) {
    534
    4
                                                return None;
    535
    22
                                            }
    536
    22
                                            Some(cmp(i, i).is_eq())
    537
    26
                                        })
    538
    10
                                        .collect::<BooleanArray>()
    539
                                }
    540
    10
                                ColumnarValue::Scalar(scalar) => {
    541
                                    // Check if scalar is null once, before the loop
    542
    10
                                    if scalar.is_null() {
    543
                                        // If scalar is null, all comparisons return null
    544
    0
                                        BooleanArray::from(vec![None; num_rows])
    545
                                    } else {
    546
                                        // Convert scalar to 1-element array
    547
    10
                                        let array = scalar.to_array()?0;
    548
    10
                                        let cmp = make_comparator(
    549
    10
                                            value.as_ref(),
    550
    10
                                            array.as_ref(),
    551
    10
                                            SortOptions::default(),
    552
    0
                                        )?;
    553
                                        // Compare each row of value with the single scalar element
    554
    10
                                        (0..num_rows)
    555
    18
                                            .map10(|i| {
    556
    18
                                                if value.is_null(i) {
    557
    1
                                                    None
    558
                                                } else {
    559
    17
                                                    Some(cmp(i, 0).is_eq())
    560
                                                }
    561
    18
                                            })
    562
    10
                                            .collect::<BooleanArray>()
    563
                                    }
    564
                                }
    565
                            };
    566
    20
                            Ok(or_kleene(&result, &rhs)?0)
    567
    20
                        },
    568
    0
                    )?;
    569
    570
    10
                    if self.negated {
    571
    2
                        not(&found)?0
    572
                    } else {
    573
    8
                        found
    574
                    }
    575
                }
    576
            };
    577
    154
            Ok(ColumnarValue::Array(Arc::new(r)))
    578
    158
        }
    579
    580
    2
        fn children(&self) -> Vec<&Arc<dyn PhysicalExpr>> {
    581
    2
            let mut children = vec![&self.expr];
    582
    2
            children.extend(&self.list);
    583
    2
            children
    584
    2
        }
    585
    586
    1
        fn with_new_children(
    587
    1
            self: Arc<Self>,
    588
    1
            children: Vec<Arc<dyn PhysicalExpr>>,
    589
    1
        ) -> Result<Arc<dyn PhysicalExpr>> {
    590
            // assume the static_filter will not change during the rewrite process
    591
    1
            Ok(Arc::new(InListExpr::new(
    592
    1
                Arc::clone(&children[0]),
    593
    1
                children[1..].to_vec(),
    594
    1
                self.negated,
    595
    1
                self.static_filter.as_ref().map(Arc::clone),
    596
    1
            )))
    597
    1
        }
    598
    599
    4
        fn fmt_sql(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
    600
    4
            self.expr.fmt_sql(f)?0;
    601
    4
            if self.negated {
    602
    2
                write!(f, " NOT")?0;
    603
    2
            }
    604
    605
    4
            write!(f, " IN (")?0;
    606
    10
            for (i, expr) in self.list.iter()4.enumerate4() {
    607
    10
                if i > 0 {
    608
    6
                    write!(f, ", ")?0;
    609
    4
                }
    610
    10
                expr.fmt_sql(f)?0;
    611
            }
    612
    4
            write!(f, ")")
    613
    4
        }
    614
    }
    615
    616
    impl PartialEq for InListExpr {
    617
    1
        fn eq(&self, other: &Self) -> bool {
    618
    1
            self.expr.eq(&other.expr)
    619
    1
                && physical_exprs_bag_equal(&self.list, &other.list)
    620
    1
                && self.negated == other.negated
    621
    1
        }
    622
    }
    623
    624
    impl Eq for InListExpr {}
    625
    626
    impl Hash for InListExpr {
    627
    0
        fn hash<H: Hasher>(&self, state: &mut H) {
    628
    0
            self.expr.hash(state);
    629
    0
            self.negated.hash(state);
    630
            // Add `self.static_filter` when hash is available
    631
    0
            self.list.hash(state);
    632
    0
        }
    633
    }
    634
    635
    /// Creates a unary expression InList
    636
    166
    pub fn in_list(
    637
    166
        expr: Arc<dyn PhysicalExpr>,
    638
    166
        list: Vec<Arc<dyn PhysicalExpr>>,
    639
    166
        negated: &bool,
    640
    166
        schema: &Schema,
    641
    166
    ) -> Result<Arc<dyn PhysicalExpr>> {
    642
        // check the data type
    643
    166
        let expr_data_type = expr.data_type(schema)?0;
    644
    769
        for list_expr in list166.iter166() {
    645
    769
            let list_expr_data_type = list_expr.data_type(schema)?0;
    646
    769
            assert_or_internal_err!(
    647
    769
                DFSchema::datatype_is_logically_equal(&expr_data_type, &list_expr_data_type),
    648
    0
                "The data type inlist should be same, the value type is {expr_data_type}, one of list expr type is {list_expr_data_type}"
    649
            );
    650
        }
    651
    652
        // Try to create a static filter for constant expressions
    653
    166
        let static_filter = try_evaluate_constant_list(&list, schema)
    654
    166
            .and_then(instantiate_static_filter)
    655
    166
            .ok();
    656
    657
    166
        Ok(Arc::new(InListExpr::new(
    658
    166
            expr,
    659
    166
            list,
    660
    166
            *negated,
    661
    166
            static_filter,
    662
    166
        )))
    663
    166
    }
    664
    665
    #[cfg(test)]
    666
    mod tests {
    667
        use super::*;
    668
        use crate::expressions;
    669
        use crate::expressions::{col, lit, try_cast};
    670
        use arrow::buffer::NullBuffer;
    671
        use datafusion_common::plan_err;
    672
        use datafusion_expr::type_coercion::binary::comparison_coercion;
    673
        use datafusion_physical_expr_common::physical_expr::fmt_sql;
    674
        use insta::assert_snapshot;
    675
        use itertools::Itertools;
    676
    677
        type InListCastResult = (Arc<dyn PhysicalExpr>, Vec<Arc<dyn PhysicalExpr>>);
    678
    679
        // Try to do the type coercion for list physical expr.
    680
        // It's just used in the test
    681
    106
        fn in_list_cast(
    682
    106
            expr: Arc<dyn PhysicalExpr>,
    683
    106
            list: Vec<Arc<dyn PhysicalExpr>>,
    684
    106
            input_schema: &Schema,
    685
    106
        ) -> Result<InListCastResult> {
    686
    106
            let expr_type = &expr.data_type(input_schema)?0;
    687
    106
            let list_types: Vec<DataType> = list
    688
    106
                .iter()
    689
    648
                .map106(|list_expr| list_expr.data_type(input_schema).unwrap())
    690
    106
                .collect();
    691
    106
            let result_type = get_coerce_type(expr_type, &list_types);
    692
    106
            match result_type {
    693
    0
                None => plan_err!(
    694
    0
                    "Can not find compatible types to compare {expr_type} with [{}]",
    695
    0
                    list_types.iter().join(", ")
    696
                ),
    697
    106
                Some(data_type) => {
    698
                    // find the coerced type
    699
    106
                    let cast_expr = try_cast(expr, input_schema, data_type.clone())?0;
    700
    106
                    let cast_list_expr = list
    701
    106
                        .into_iter()
    702
    648
                        .map106(|list_expr| {
    703
    648
                            try_cast(list_expr, input_schema, data_type.clone()).unwrap()
    704
    648
                        })
    705
    106
                        .collect();
    706
    106
                    Ok((cast_expr, cast_list_expr))
    707
                }
    708
            }
    709
    106
        }
    710
    711
    5
        fn try_cast_static_filter_to_set(
    712
    5
            list: &[Arc<dyn PhysicalExpr>],
    713
    5
            schema: &Schema,
    714
    5
        ) -> Result<ArrayStaticFilter> {
    715
    5
            let array4 = try_evaluate_constant_list(list, schema)?1;
    716
    4
            ArrayStaticFilter::try_new(array)
    717
    5
        }
    718
    719
        // Attempts to coerce the types of `list_type` to be comparable with the
    720
        // `expr_type`
    721
    106
        fn get_coerce_type(expr_type: &DataType, list_type: &[DataType]) -> Option<DataType> {
    722
    106
            list_type
    723
    106
                .iter()
    724
    648
                .try_fold106(expr_type106.clone106(), |left_type, right_type| {
    725
    648
                    comparison_coercion(&left_type, right_type)
    726
    648
                })
    727
    106
        }
    728
    729
        /// Test helper macro that evaluates an IN LIST expression with automatic type casting.
    730
        ///
    731
        /// # Parameters
    732
        /// - `$BATCH`: The `RecordBatch` containing the input data to evaluate against
    733
        /// - `$LIST`: A `Vec<Arc<dyn PhysicalExpr>>` of literal expressions representing the IN list values
    734
        /// - `$NEGATED`: A `&bool` indicating whether this is a NOT IN operation (true) or IN operation (false)
    735
        /// - `$EXPECTED`: A `Vec<Option<bool>>` representing the expected boolean results for each row
    736
        /// - `$COL`: An `Arc<dyn PhysicalExpr>` representing the column expression to evaluate
    737
        /// - `$SCHEMA`: A `&Schema` reference for the input batch
    738
        ///
    739
        /// This macro first applies type casting to the column and list expressions to ensure
    740
        /// type compatibility, then delegates to `in_list_raw!` to perform the evaluation and assertion.
    741
        macro_rules! in_list {
    742
            ($BATCH:expr, $LIST:expr, $NEGATED:expr, $EXPECTED:expr, $COL:expr, $SCHEMA:expr) => {{
    743
                let (cast_expr, cast_list_exprs) = in_list_cast($COL, $LIST, $SCHEMA)?;
    744
                in_list_raw!(
    745
                    $BATCH,
    746
                    cast_list_exprs,
    747
                    $NEGATED,
    748
                    $EXPECTED,
    749
                    cast_expr,
    750
                    $SCHEMA
    751
                );
    752
            }};
    753
        }
    754
    755
        /// Test helper macro that evaluates an IN LIST expression without automatic type casting.
    756
        ///
    757
        /// # Parameters
    758
        /// - `$BATCH`: The `RecordBatch` containing the input data to evaluate against
    759
        /// - `$LIST`: A `Vec<Arc<dyn PhysicalExpr>>` of literal expressions representing the IN list values
    760
        /// - `$NEGATED`: A `&bool` indicating whether this is a NOT IN operation (true) or IN operation (false)
    761
        /// - `$EXPECTED`: A `Vec<Option<bool>>` representing the expected boolean results for each row
    762
        /// - `$COL`: An `Arc<dyn PhysicalExpr>` representing the column expression to evaluate
    763
        /// - `$SCHEMA`: A `&Schema` reference for the input batch
    764
        ///
    765
        /// This macro creates an IN LIST expression, evaluates it against the batch, converts the result
    766
        /// to a `BooleanArray`, and asserts that it matches the expected output. Use this when the column
    767
        /// and list expressions are already the correct types and don't require casting.
    768
        macro_rules! in_list_raw {
    769
            ($BATCH:expr, $LIST:expr, $NEGATED:expr, $EXPECTED:expr, $COL:expr, $SCHEMA:expr) => {{
    770
                let expr = in_list($COL, $LIST, $NEGATED, $SCHEMA).unwrap();
    771
                let result = expr
    772
                    .evaluate(&$BATCH)?
    773
                    .into_array($BATCH.num_rows())
    774
                    .expect("Failed to convert to array");
    775
                let result = as_boolean_array(&result);
    776
                let expected = &BooleanArray::from($EXPECTED);
    777
                assert_eq!(expected, result);
    778
            }};
    779
        }
    780
    781
        #[test]
    782
    1
        fn in_list_utf8() -> Result<()> {
    783
    1
            let schema = Schema::new(vec![Field::new("a", DataType::Utf8, true)]);
    784
    1
            let a = StringArray::from(vec![Some("a"), Some("d"), None]);
    785
    1
            let col_a = col("a", &schema)?0;
    786
    1
            let batch = RecordBatch::try_new(Arc::new(schema.clone()), vec![Arc::new(a)])?0;
    787
    788
            // expression: "a in ("a", "b")"
    789
    1
            let list = vec![lit("a"), lit("b")];
    790
    1
            in_list!(
    791
    1
                batch,
    792
    1
                list,
    793
    1
                &false,
    794
    1
                vec![Some(true), Some(false), None],
    795
    1
                Arc::clone(&col_a),
    796
    1
                &schema
    797
            );
    798
    799
            // expression: "a not in ("a", "b")"
    800
    1
            let list = vec![lit("a"), lit("b")];
    801
    1
            in_list!(
    802
    1
                batch,
    803
    1
                list,
    804
    1
                &true,
    805
    1
                vec![Some(false), Some(true), None],
    806
    1
                Arc::clone(&col_a),
    807
    1
                &schema
    808
            );
    809
    810
            // expression: "a in ("a", "b", null)"
    811
    1
            let list = vec![lit("a"), lit("b"), lit(ScalarValue::Utf8(None))];
    812
    1
            in_list!(
    813
    1
                batch,
    814
    1
                list,
    815
    1
                &false,
    816
    1
                vec![Some(true), None, None],
    817
    1
                Arc::clone(&col_a),
    818
    1
                &schema
    819
            );
    820
    821
            // expression: "a not in ("a", "b", null)"
    822
    1
            let list = vec![lit("a"), lit("b"), lit(ScalarValue::Utf8(None))];
    823
    1
            in_list!(
    824
    1
                batch,
    825
    1
                list,
    826
    1
                &true,
    827
    1
                vec![Some(false), None, None],
    828
    1
                Arc::clone(&col_a),
    829
    1
                &schema
    830
            );
    831
    832
    1
            Ok(())
    833
    1
        }
    834
    835
        #[test]
    836
    1
        fn in_list_binary() -> Result<()> {
    837
    1
            let schema = Schema::new(vec![Field::new("a", DataType::Binary, true)]);
    838
    1
            let a = BinaryArray::from(vec![
    839
    1
                Some([1, 2, 3].as_slice()),
    840
    1
                Some([1, 2, 2].as_slice()),
    841
    1
                None,
    842
            ]);
    843
    1
            let col_a = col("a", &schema)?0;
    844
    1
            let batch = RecordBatch::try_new(Arc::new(schema.clone()), vec![Arc::new(a)])?0;
    845
    846
            // expression: "a in ([1, 2, 3], [4, 5, 6])"
    847
    1
            let list = vec![lit([1, 2, 3].as_slice()), lit([4, 5, 6].as_slice())];
    848
    1
            in_list!(
    849
    1
                batch,
    850
    1
                list.clone(),
    851
    1
                &false,
    852
    1
                vec![Some(true), Some(false), None],
    853
    1
                Arc::clone(&col_a),
    854
    1
                &schema
    855
            );
    856
    857
            // expression: "a not in ([1, 2, 3], [4, 5, 6])"
    858
    1
            in_list!(
    859
    1
                batch,
    860
    1
                list,
    861
    1
                &true,
    862
    1
                vec![Some(false), Some(true), None],
    863
    1
                Arc::clone(&col_a),
    864
    1
                &schema
    865
            );
    866
    867
            // expression: "a in ([1, 2, 3], [4, 5, 6], null)"
    868
    1
            let list = vec![
    869
    1
                lit([1, 2, 3].as_slice()),
    870
    1
                lit([4, 5, 6].as_slice()),
    871
    1
                lit(ScalarValue::Binary(None)),
    872
            ];
    873
    1
            in_list!(
    874
    1
                batch,
    875
    1
                list.clone(),
    876
    1
                &false,
    877
    1
                vec![Some(true), None, None],
    878
    1
                Arc::clone(&col_a),
    879
    1
                &schema
    880
            );
    881
    882
            // expression: "a in ([1, 2, 3], [4, 5, 6], null)"
    883
    1
            in_list!(
    884
    1
                batch,
    885
    1
                list,
    886
    1
                &true,
    887
    1
                vec![Some(false), None, None],
    888
    1
                Arc::clone(&col_a),
    889
    1
                &schema
    890
            );
    891
    892
    1
            Ok(())
    893
    1
        }
    894
    895
        #[test]
    896
    1
        fn in_list_int64() -> Result<()> {
    897
    1
            let schema = Schema::new(vec![Field::new("a", DataType::Int64, true)]);
    898
    1
            let a = Int64Array::from(vec![Some(0), Some(2), None]);
    899
    1
            let col_a = col("a", &schema)?0;
    900
    1
            let batch = RecordBatch::try_new(Arc::new(schema.clone()), vec![Arc::new(a)])?0;
    901
    902
            // expression: "a in (0, 1)"
    903
    1
            let list = vec![lit(0i64), lit(1i64)];
    904
    1
            in_list!(
    905
    1
                batch,
    906
    1
                list,
    907
    1
                &false,
    908
    1
                vec![Some(true), Some(false), None],
    909
    1
                Arc::clone(&col_a),
    910
    1
                &schema
    911
            );
    912
    913
            // expression: "a not in (0, 1)"
    914
    1
            let list = vec![lit(0i64), lit(1i64)];
    915
    1
            in_list!(
    916
    1
                batch,
    917
    1
                list,
    918
    1
                &true,
    919
    1
                vec![Some(false), Some(true), None],
    920
    1
                Arc::clone(&col_a),
    921
    1
                &schema
    922
            );
    923
    924
            // expression: "a in (0, 1, NULL)"
    925
    1
            let list = vec![lit(0i64), lit(1i64), lit(ScalarValue::Null)];
    926
    1
            in_list!(
    927
    1
                batch,
    928
    1
                list,
    929
    1
                &false,
    930
    1
                vec![Some(true), None, None],
    931
    1
                Arc::clone(&col_a),
    932
    1
                &schema
    933
            );
    934
    935
            // expression: "a not in (0, 1, NULL)"
    936
    1
            let list = vec![lit(0i64), lit(1i64), lit(ScalarValue::Null)];
    937
    1
            in_list!(
    938
    1
                batch,
    939
    1
                list,
    940
    1
                &true,
    941
    1
                vec![Some(false), None, None],
    942
    1
                Arc::clone(&col_a),
    943
    1
                &schema
    944
            );
    945
    946
    1
            Ok(())
    947
    1
        }
    948
    949
        #[test]
    950
    1
        fn in_list_float64() -> Result<()> {
    951
    1
            let schema = Schema::new(vec![Field::new("a", DataType::Float64, true)]);
    952
    1
            let a = Float64Array::from(vec![
    953
    1
                Some(0.0),
    954
    1
                Some(0.2),
    955
    1
                None,
    956
    1
                Some(f64::NAN),
    957
    1
                Some(-f64::NAN),
    958
            ]);
    959
    1
            let col_a = col("a", &schema)?0;
    960
    1
            let batch = RecordBatch::try_new(Arc::new(schema.clone()), vec![Arc::new(a)])?0;
    961
    962
            // expression: "a in (0.0, 0.1)"
    963
    1
            let list = vec![lit(0.0f64), lit(0.1f64)];
    964
    1
            in_list!(
    965
    1
                batch,
    966
    1
                list,
    967
    1
                &false,
    968
    1
                vec![Some(true), Some(false), None, Some(false), Some(false)],
    969
    1
                Arc::clone(&col_a),
    970
    1
                &schema
    971
            );
    972
    973
            // expression: "a not in (0.0, 0.1)"
    974
    1
            let list = vec![lit(0.0f64), lit(0.1f64)];
    975
    1
            in_list!(
    976
    1
                batch,
    977
    1
                list,
    978
    1
                &true,
    979
    1
                vec![Some(false), Some(true), None, Some(true), Some(true)],
    980
    1
                Arc::clone(&col_a),
    981
    1
                &schema
    982
            );
    983
    984
            // expression: "a in (0.0, 0.1, NULL)"
    985
    1
            let list = vec![lit(0.0f64), lit(0.1f64), lit(ScalarValue::Null)];
    986
    1
            in_list!(
    987
    1
                batch,
    988
    1
                list,
    989
    1
                &false,
    990
    1
                vec![Some(true), None, None, None, None],
    991
    1
                Arc::clone(&col_a),
    992
    1
                &schema
    993
            );
    994
    995
            // expression: "a not in (0.0, 0.1, NULL)"
    996
    1
            let list = vec![lit(0.0f64), lit(0.1f64), lit(ScalarValue::Null)];
    997
    1
            in_list!(
    998
    1
                batch,
    999
    1
                list,
    1000
    1
                &true,
    1001
    1
                vec![Some(false), None, None, None, None],
    1002
    1
                Arc::clone(&col_a),
    1003
    1
                &schema
    1004
            );
    1005
    1006
            // expression: "a in (0.0, 0.1, NaN)"
    1007
    1
            let list = vec![lit(0.0f64), lit(0.1f64), lit(f64::NAN)];
    1008
    1
            in_list!(
    1009
    1
                batch,
    1010
    1
                list,
    1011
    1
                &false,
    1012
    1
                vec![Some(true), Some(false), None, Some(true), Some(false)],
    1013
    1
                Arc::clone(&col_a),
    1014
    1
                &schema
    1015
            );
    1016
    1017
            // expression: "a not in (0.0, 0.1, NaN)"
    1018
    1
            let list = vec![lit(0.0f64), lit(0.1f64), lit(f64::NAN)];
    1019
    1
            in_list!(
    1020
    1
                batch,
    1021
    1
                list,
    1022
    1
                &true,
    1023
    1
                vec![Some(false), Some(true), None, Some(false), Some(true)],
    1024
    1
                Arc::clone(&col_a),
    1025
    1
                &schema
    1026
            );
    1027
    1028
            // expression: "a in (0.0, 0.1, -NaN)"
    1029
    1
            let list = vec![lit(0.0f64), lit(0.1f64), lit(-f64::NAN)];
    1030
    1
            in_list!(
    1031
    1
                batch,
    1032
    1
                list,
    1033
    1
                &false,
    1034
    1
                vec![Some(true), Some(false), None, Some(false), Some(true)],
    1035
    1
                Arc::clone(&col_a),
    1036
    1
                &schema
    1037
            );
    1038
    1039
            // expression: "a not in (0.0, 0.1, -NaN)"
    1040
    1
            let list = vec![lit(0.0f64), lit(0.1f64), lit(-f64::NAN)];
    1041
    1
            in_list!(
    1042
    1
                batch,
    1043
    1
                list,
    1044
    1
                &true,
    1045
    1
                vec![Some(false), Some(true), None, Some(true), Some(false)],
    1046
    1
                Arc::clone(&col_a),
    1047
    1
                &schema
    1048
            );
    1049
    1050
    1
            Ok(())
    1051
    1
        }
    1052
    1053
        #[test]
    1054
    1
        fn in_list_bool() -> Result<()> {
    1055
    1
            let schema = Schema::new(vec![Field::new("a", DataType::Boolean, true)]);
    1056
    1
            let a = BooleanArray::from(vec![Some(true), None]);
    1057
    1
            let col_a = col("a", &schema)?0;
    1058
    1
            let batch = RecordBatch::try_new(Arc::new(schema.clone()), vec![Arc::new(a)])?0;
    1059
    1060
            // expression: "a in (true)"
    1061
    1
            let list = vec![lit(true)];
    1062
    1
            in_list!(
    1063
    1
                batch,
    1064
    1
                list,
    1065
    1
                &false,
    1066
    1
                vec![Some(true), None],
    1067
    1
                Arc::clone(&col_a),
    1068
    1
                &schema
    1069
            );
    1070
    1071
            // expression: "a not in (true)"
    1072
    1
            let list = vec![lit(true)];
    1073
    1
            in_list!(
    1074
    1
                batch,
    1075
    1
                list,
    1076
    1
                &true,
    1077
    1
                vec![Some(false), None],
    1078
    1
                Arc::clone(&col_a),
    1079
    1
                &schema
    1080
            );
    1081
    1082
            // expression: "a in (true, NULL)"
    1083
    1
            let list = vec![lit(true), lit(ScalarValue::Null)];
    1084
    1
            in_list!(
    1085
    1
                batch,
    1086
    1
                list,
    1087
    1
                &false,
    1088
    1
                vec![Some(true), None],
    1089
    1
                Arc::clone(&col_a),
    1090
    1
                &schema
    1091
            );
    1092
    1093
            // expression: "a not in (true, NULL)"
    1094
    1
            let list = vec![lit(true), lit(ScalarValue::Null)];
    1095
    1
            in_list!(
    1096
    1
                batch,
    1097
    1
                list,
    1098
    1
                &true,
    1099
    1
                vec![Some(false), None],
    1100
    1
                Arc::clone(&col_a),
    1101
    1
                &schema
    1102
            );
    1103
    1104
    1
            Ok(())
    1105
    1
        }
    1106
    1107
        #[test]
    1108
    1
        fn in_list_int8() -> Result<()> {
    1109
    1
            let schema = Schema::new(vec![Field::new("a", DataType::Int8, true)]);
    1110
    1
            let a = Int8Array::from(vec![Some(0), Some(2), None]);
    1111
    1
            let col_a = col("a", &schema)?0;
    1112
    1
            let batch = RecordBatch::try_new(Arc::new(schema.clone()), vec![Arc::new(a)])?0;
    1113
    1114
            // expression: "a in (0, 1)"
    1115
    1
            let list = vec![lit(0i8), lit(1i8)];
    1116
    1
            in_list!(
    1117
    1
                batch,
    1118
    1
                list,
    1119
    1
                &false,
    1120
    1
                vec![Some(true), Some(false), None],
    1121
    1
                Arc::clone(&col_a),
    1122
    1
                &schema
    1123
            );
    1124
    1125
            // expression: "a not in (0, 1)"
    1126
    1
            let list = vec![lit(0i8), lit(1i8)];
    1127
    1
            in_list!(
    1128
    1
                batch,
    1129
    1
                list,
    1130
    1
                &true,
    1131
    1
                vec![Some(false), Some(true), None],
    1132
    1
                Arc::clone(&col_a),
    1133
    1
                &schema
    1134
            );
    1135
    1136
            // expression: "a in (0, 1, NULL)"
    1137
    1
            let list = vec![lit(0i8), lit(1i8), lit(ScalarValue::Null)];
    1138
    1
            in_list!(
    1139
    1
                batch,
    1140
    1
                list,
    1141
    1
                &false,
    1142
    1
                vec![Some(true), None, None],
    1143
    1
                Arc::clone(&col_a),
    1144
    1
                &schema
    1145
            );
    1146
    1147
            // expression: "a not in (0, 1, NULL)"
    1148
    1
            let list = vec![lit(0i8), lit(1i8), lit(ScalarValue::Null)];
    1149
    1
            in_list!(
    1150
    1
                batch,
    1151
    1
                list,
    1152
    1
                &true,
    1153
    1
                vec![Some(false), None, None],
    1154
    1
                Arc::clone(&col_a),
    1155
    1
                &schema
    1156
            );
    1157
    1158
    1
            Ok(())
    1159
    1
        }
    1160
    1161
        #[test]
    1162
    1
        fn in_list_int16() -> Result<()> {
    1163
    1
            let schema = Schema::new(vec![Field::new("a", DataType::Int16, true)]);
    1164
    1
            let a = Int16Array::from(vec![Some(0), Some(2), None]);
    1165
    1
            let col_a = col("a", &schema)?0;
    1166
    1
            let batch = RecordBatch::try_new(Arc::new(schema.clone()), vec![Arc::new(a)])?0;
    1167
    1168
            // expression: "a in (0, 1)"
    1169
    1
            let list = vec![lit(0i16), lit(1i16)];
    1170
    1
            in_list!(
    1171
    1
                batch,
    1172
    1
                list,
    1173
    1
                &false,
    1174
    1
                vec![Some(true), Some(false), None],
    1175
    1
                Arc::clone(&col_a),
    1176
    1
                &schema
    1177
            );
    1178
    1179
            // expression: "a not in (0, 1)"
    1180
    1
            let list = vec![lit(0i16), lit(1i16)];
    1181
    1
            in_list!(
    1182
    1
                batch,
    1183
    1
                list,
    1184
    1
                &true,
    1185
    1
                vec![Some(false), Some(true), None],
    1186
    1
                Arc::clone(&col_a),
    1187
    1
                &schema
    1188
            );
    1189
    1190
            // expression: "a in (0, 1, NULL)"
    1191
    1
            let list = vec![lit(0i16), lit(1i16), lit(ScalarValue::Null)];
    1192
    1
            in_list!(
    1193
    1
                batch,
    1194
    1
                list,
    1195
    1
                &false,
    1196
    1
                vec![Some(true), None, None],
    1197
    1
                Arc::clone(&col_a),
    1198
    1
                &schema
    1199
            );
    1200
    1201
            // expression: "a not in (0, 1, NULL)"
    1202
    1
            let list = vec![lit(0i16), lit(1i16), lit(ScalarValue::Null)];
    1203
    1
            in_list!(
    1204
    1
                batch,
    1205
    1
                list,
    1206
    1
                &true,
    1207
    1
                vec![Some(false), None, None],
    1208
    1
                Arc::clone(&col_a),
    1209
    1
                &schema
    1210
            );
    1211
    1212
    1
            Ok(())
    1213
    1
        }
    1214
    1215
        #[test]
    1216
    1
        fn in_list_int32() -> Result<()> {
    1217
    1
            let schema = Schema::new(vec![Field::new("a", DataType::Int32, true)]);
    1218
    1
            let a = Int32Array::from(vec![Some(0), Some(2), None]);
    1219
    1
            let col_a = col("a", &schema)?0;
    1220
    1
            let batch = RecordBatch::try_new(Arc::new(schema.clone()), vec![Arc::new(a)])?0;
    1221
    1222
            // expression: "a in (0, 1)"
    1223
    1
            let list = vec![lit(0i32), lit(1i32)];
    1224
    1
            in_list!(
    1225
    1
                batch,
    1226
    1
                list,
    1227
    1
                &false,
    1228
    1
                vec![Some(true), Some(false), None],
    1229
    1
                Arc::clone(&col_a),
    1230
    1
                &schema
    1231
            );
    1232
    1233
            // expression: "a not in (0, 1)"
    1234
    1
            let list = vec![lit(0i32), lit(1i32)];
    1235
    1
            in_list!(
    1236
    1
                batch,
    1237
    1
                list,
    1238
    1
                &true,
    1239
    1
                vec![Some(false), Some(true), None],
    1240
    1
                Arc::clone(&col_a),
    1241
    1
                &schema
    1242
            );
    1243
    1244
            // expression: "a in (0, 1, NULL)"
    1245
    1
            let list = vec![lit(0i32), lit(1i32), lit(ScalarValue::Null)];
    1246
    1
            in_list!(
    1247
    1
                batch,
    1248
    1
                list,
    1249
    1
                &false,
    1250
    1
                vec![Some(true), None, None],
    1251
    1
                Arc::clone(&col_a),
    1252
    1
                &schema
    1253
            );
    1254
    1255
            // expression: "a not in (0, 1, NULL)"
    1256
    1
            let list = vec![lit(0i32), lit(1i32), lit(ScalarValue::Null)];
    1257
    1
            in_list!(
    1258
    1
                batch,
    1259
    1
                list,
    1260
    1
                &true,
    1261
    1
                vec![Some(false), None, None],
    1262
    1
                Arc::clone(&col_a),
    1263
    1
                &schema
    1264
            );
    1265
    1266
    1
            Ok(())
    1267
    1
        }
    1268
    1269
        #[test]
    1270
    1
        fn in_list_uint8() -> Result<()> {
    1271
    1
            let schema = Schema::new(vec![Field::new("a", DataType::UInt8, true)]);
    1272
    1
            let a = UInt8Array::from(vec![Some(0), Some(2), None]);
    1273
    1
            let col_a = col("a", &schema)?0;
    1274
    1
            let batch = RecordBatch::try_new(Arc::new(schema.clone()), vec![Arc::new(a)])?0;
    1275
    1276
            // expression: "a in (0, 1)"
    1277
    1
            let list = vec![lit(0u8), lit(1u8)];
    1278
    1
            in_list!(
    1279
    1
                batch,
    1280
    1
                list,
    1281
    1
                &false,
    1282
    1
                vec![Some(true), Some(false), None],
    1283
    1
                Arc::clone(&col_a),
    1284
    1
                &schema
    1285
            );
    1286
    1287
            // expression: "a not in (0, 1)"
    1288
    1
            let list = vec![lit(0u8), lit(1u8)];
    1289
    1
            in_list!(
    1290
    1
                batch,
    1291
    1
                list,
    1292
    1
                &true,
    1293
    1
                vec![Some(false), Some(true), None],
    1294
    1
                Arc::clone(&col_a),
    1295
    1
                &schema
    1296
            );
    1297
    1298
            // expression: "a in (0, 1, NULL)"
    1299
    1
            let list = vec![lit(0u8), lit(1u8), lit(ScalarValue::Null)];
    1300
    1
            in_list!(
    1301
    1
                batch,
    1302
    1
                list,
    1303
    1
                &false,
    1304
    1
                vec![Some(true), None, None],
    1305
    1
                Arc::clone(&col_a),
    1306
    1
                &schema
    1307
            );
    1308
    1309
            // expression: "a not in (0, 1, NULL)"
    1310
    1
            let list = vec![lit(0u8), lit(1u8), lit(ScalarValue::Null)];
    1311
    1
            in_list!(
    1312
    1
                batch,
    1313
    1
                list,
    1314
    1
                &true,
    1315
    1
                vec![Some(false), None, None],
    1316
    1
                Arc::clone(&col_a),
    1317
    1
                &schema
    1318
            );
    1319
    1320
    1
            Ok(())
    1321
    1
        }
    1322
    1323
        #[test]
    1324
    1
        fn in_list_uint16() -> Result<()> {
    1325
    1
            let schema = Schema::new(vec![Field::new("a", DataType::UInt16, true)]);
    1326
    1
            let a = UInt16Array::from(vec![Some(0), Some(2), None]);
    1327
    1
            let col_a = col("a", &schema)?0;
    1328
    1
            let batch = RecordBatch::try_new(Arc::new(schema.clone()), vec![Arc::new(a)])?0;
    1329
    1330
            // expression: "a in (0, 1)"
    1331
    1
            let list = vec![lit(0u16), lit(1u16)];
    1332
    1
            in_list!(
    1333
    1
                batch,
    1334
    1
                list,
    1335
    1
                &false,
    1336
    1
                vec![Some(true), Some(false), None],
    1337
    1
                Arc::clone(&col_a),
    1338
    1
                &schema
    1339
            );
    1340
    1341
            // expression: "a not in (0, 1)"
    1342
    1
            let list = vec![lit(0u16), lit(1u16)];
    1343
    1
            in_list!(
    1344
    1
                batch,
    1345
    1
                list,
    1346
    1
                &true,
    1347
    1
                vec![Some(false), Some(true), None],
    1348
    1
                Arc::clone(&col_a),
    1349
    1
                &schema
    1350
            );
    1351
    1352
            // expression: "a in (0, 1, NULL)"
    1353
    1
            let list = vec![lit(0u16), lit(1u16), lit(ScalarValue::Null)];
    1354
    1
            in_list!(
    1355
    1
                batch,
    1356
    1
                list,
    1357
    1
                &false,
    1358
    1
                vec![Some(true), None, None],
    1359
    1
                Arc::clone(&col_a),
    1360
    1
                &schema
    1361
            );
    1362
    1363
            // expression: "a not in (0, 1, NULL)"
    1364
    1
            let list = vec![lit(0u16), lit(1u16), lit(ScalarValue::Null)];
    1365
    1
            in_list!(
    1366
    1
                batch,
    1367
    1
                list,
    1368
    1
                &true,
    1369
    1
                vec![Some(false), None, None],
    1370
    1
                Arc::clone(&col_a),
    1371
    1
                &schema
    1372
            );
    1373
    1374
    1
            Ok(())
    1375
    1
        }
    1376
    1377
        #[test]
    1378
    1
        fn in_list_uint32() -> Result<()> {
    1379
    1
            let schema = Schema::new(vec![Field::new("a", DataType::UInt32, true)]);
    1380
    1
            let a = UInt32Array::from(vec![Some(0), Some(2), None]);
    1381
    1
            let col_a = col("a", &schema)?0;
    1382
    1
            let batch = RecordBatch::try_new(Arc::new(schema.clone()), vec![Arc::new(a)])?0;
    1383
    1384
            // expression: "a in (0, 1)"
    1385
    1
            let list = vec![lit(0u32), lit(1u32)];
    1386
    1
            in_list!(
    1387
    1
                batch,
    1388
    1
                list,
    1389
    1
                &false,
    1390
    1
                vec![Some(true), Some(false), None],
    1391
    1
                Arc::clone(&col_a),
    1392
    1
                &schema
    1393
            );
    1394
    1395
            // expression: "a not in (0, 1)"
    1396
    1
            let list = vec![lit(0u32), lit(1u32)];
    1397
    1
            in_list!(
    1398
    1
                batch,
    1399
    1
                list,
    1400
    1
                &true,
    1401
    1
                vec![Some(false), Some(true), None],
    1402
    1
                Arc::clone(&col_a),
    1403
    1
                &schema
    1404
            );
    1405
    1406
            // expression: "a in (0, 1, NULL)"
    1407
    1
            let list = vec![lit(0u32), lit(1u32), lit(ScalarValue::Null)];
    1408
    1
            in_list!(
    1409
    1
                batch,
    1410
    1
                list,
    1411
    1
                &false,
    1412
    1
                vec![Some(true), None, None],
    1413
    1
                Arc::clone(&col_a),
    1414
    1
                &schema
    1415
            );
    1416
    1417
            // expression: "a not in (0, 1, NULL)"
    1418
    1
            let list = vec![lit(0u32), lit(1u32), lit(ScalarValue::Null)];
    1419
    1
            in_list!(
    1420
    1
                batch,
    1421
    1
                list,
    1422
    1
                &true,
    1423
    1
                vec![Some(false), None, None],
    1424
    1
                Arc::clone(&col_a),
    1425
    1
                &schema
    1426
            );
    1427
    1428
    1
            Ok(())
    1429
    1
        }
    1430
    1431
        #[test]
    1432
    1
        fn in_list_uint64() -> Result<()> {
    1433
    1
            let schema = Schema::new(vec![Field::new("a", DataType::UInt64, true)]);
    1434
    1
            let a = UInt64Array::from(vec![Some(0), Some(2), None]);
    1435
    1
            let col_a = col("a", &schema)?0;
    1436
    1
            let batch = RecordBatch::try_new(Arc::new(schema.clone()), vec![Arc::new(a)])?0;
    1437
    1438
            // expression: "a in (0, 1)"
    1439
    1
            let list = vec![lit(0u64), lit(1u64)];
    1440
    1
            in_list!(
    1441
    1
                batch,
    1442
    1
                list,
    1443
    1
                &false,
    1444
    1
                vec![Some(true), Some(false), None],
    1445
    1
                Arc::clone(&col_a),
    1446
    1
                &schema
    1447
            );
    1448
    1449
            // expression: "a not in (0, 1)"
    1450
    1
            let list = vec![lit(0u64), lit(1u64)];
    1451
    1
            in_list!(
    1452
    1
                batch,
    1453
    1
                list,
    1454
    1
                &true,
    1455
    1
                vec![Some(false), Some(true), None],
    1456
    1
                Arc::clone(&col_a),
    1457
    1
                &schema
    1458
            );
    1459
    1460
            // expression: "a in (0, 1, NULL)"
    1461
    1
            let list = vec![lit(0u64), lit(1u64), lit(ScalarValue::Null)];
    1462
    1
            in_list!(
    1463
    1
                batch,
    1464
    1
                list,
    1465
    1
                &false,
    1466
    1
                vec![Some(true), None, None],
    1467
    1
                Arc::clone(&col_a),
    1468
    1
                &schema
    1469
            );
    1470
    1471
            // expression: "a not in (0, 1, NULL)"
    1472
    1
            let list = vec![lit(0u64), lit(1u64), lit(ScalarValue::Null)];
    1473
    1
            in_list!(
    1474
    1
                batch,
    1475
    1
                list,
    1476
    1
                &true,
    1477
    1
                vec![Some(false), None, None],
    1478
    1
                Arc::clone(&col_a),
    1479
    1
                &schema
    1480
            );
    1481
    1482
    1
            Ok(())
    1483
    1
        }
    1484
    1485
        #[test]
    1486
    1
        fn in_list_large_utf8() -> Result<()> {
    1487
    1
            let schema = Schema::new(vec![Field::new("a", DataType::LargeUtf8, true)]);
    1488
    1
            let a = LargeStringArray::from(vec![Some("a"), Some("d"), None]);
    1489
    1
            let col_a = col("a", &schema)?0;
    1490
    1
            let batch = RecordBatch::try_new(Arc::new(schema.clone()), vec![Arc::new(a)])?0;
    1491
    1492
            // expression: "a in ("a", "b")"
    1493
    1
            let list = vec![lit("a"), lit("b")];
    1494
    1
            in_list!(
    1495
    1
                batch,
    1496
    1
                list,
    1497
    1
                &false,
    1498
    1
                vec![Some(true), Some(false), None],
    1499
    1
                Arc::clone(&col_a),
    1500
    1
                &schema
    1501
            );
    1502
    1503
            // expression: "a not in ("a", "b")"
    1504
    1
            let list = vec![lit("a"), lit("b")];
    1505
    1
            in_list!(
    1506
    1
                batch,
    1507
    1
                list,
    1508
    1
                &true,
    1509
    1
                vec![Some(false), Some(true), None],
    1510
    1
                Arc::clone(&col_a),
    1511
    1
                &schema
    1512
            );
    1513
    1514
            // expression: "a in ("a", "b", null)"
    1515
    1
            let list = vec![lit("a"), lit("b"), lit(ScalarValue::LargeUtf8(None))];
    1516
    1
            in_list!(
    1517
    1
                batch,
    1518
    1
                list,
    1519
    1
                &false,
    1520
    1
                vec![Some(true), None, None],
    1521
    1
                Arc::clone(&col_a),
    1522
    1
                &schema
    1523
            );
    1524
    1525
            // expression: "a not in ("a", "b", null)"
    1526
    1
            let list = vec![lit("a"), lit("b"), lit(ScalarValue::LargeUtf8(None))];
    1527
    1
            in_list!(
    1528
    1
                batch,
    1529
    1
                list,
    1530
    1
                &true,
    1531
    1
                vec![Some(false), None, None],
    1532
    1
                Arc::clone(&col_a),
    1533
    1
                &schema
    1534
            );
    1535
    1536
    1
            Ok(())
    1537
    1
        }
    1538
    1539
        #[test]
    1540
    1
        fn in_list_utf8_view() -> Result<()> {
    1541
    1
            let schema = Schema::new(vec![Field::new("a", DataType::Utf8View, true)]);
    1542
    1
            let a = StringViewArray::from(vec![Some("a"), Some("d"), None]);
    1543
    1
            let col_a = col("a", &schema)?0;
    1544
    1
            let batch = RecordBatch::try_new(Arc::new(schema.clone()), vec![Arc::new(a)])?0;
    1545
    1546
            // expression: "a in ("a", "b")"
    1547
    1
            let list = vec![lit("a"), lit("b")];
    1548
    1
            in_list!(
    1549
    1
                batch,
    1550
    1
                list,
    1551
    1
                &false,
    1552
    1
                vec![Some(true), Some(false), None],
    1553
    1
                Arc::clone(&col_a),
    1554
    1
                &schema
    1555
            );
    1556
    1557
            // expression: "a not in ("a", "b")"
    1558
    1
            let list = vec![lit("a"), lit("b")];
    1559
    1
            in_list!(
    1560
    1
                batch,
    1561
    1
                list,
    1562
    1
                &true,
    1563
    1
                vec![Some(false), Some(true), None],
    1564
    1
                Arc::clone(&col_a),
    1565
    1
                &schema
    1566
            );
    1567
    1568
            // expression: "a in ("a", "b", null)"
    1569
    1
            let list = vec![lit("a"), lit("b"), lit(ScalarValue::Utf8View(None))];
    1570
    1
            in_list!(
    1571
    1
                batch,
    1572
    1
                list,
    1573
    1
                &false,
    1574
    1
                vec![Some(true), None, None],
    1575
    1
                Arc::clone(&col_a),
    1576
    1
                &schema
    1577
            );
    1578
    1579
            // expression: "a not in ("a", "b", null)"
    1580
    1
            let list = vec![lit("a"), lit("b"), lit(ScalarValue::Utf8View(None))];
    1581
    1
            in_list!(
    1582
    1
                batch,
    1583
    1
                list,
    1584
    1
                &true,
    1585
    1
                vec![Some(false), None, None],
    1586
    1
                Arc::clone(&col_a),
    1587
    1
                &schema
    1588
            );
    1589
    1590
    1
            Ok(())
    1591
    1
        }
    1592
    1593
        #[test]
    1594
    1
        fn in_list_large_binary() -> Result<()> {
    1595
    1
            let schema = Schema::new(vec![Field::new("a", DataType::LargeBinary, true)]);
    1596
    1
            let a = LargeBinaryArray::from(vec![
    1597
    1
                Some([1, 2, 3].as_slice()),
    1598
    1
                Some([1, 2, 2].as_slice()),
    1599
    1
                None,
    1600
            ]);
    1601
    1
            let col_a = col("a", &schema)?0;
    1602
    1
            let batch = RecordBatch::try_new(Arc::new(schema.clone()), vec![Arc::new(a)])?0;
    1603
    1604
            // expression: "a in ([1, 2, 3], [4, 5, 6])"
    1605
    1
            let list = vec![lit([1, 2, 3].as_slice()), lit([4, 5, 6].as_slice())];
    1606
    1
            in_list!(
    1607
    1
                batch,
    1608
    1
                list.clone(),
    1609
    1
                &false,
    1610
    1
                vec![Some(true), Some(false), None],
    1611
    1
                Arc::clone(&col_a),
    1612
    1
                &schema
    1613
            );
    1614
    1615
            // expression: "a not in ([1, 2, 3], [4, 5, 6])"
    1616
    1
            in_list!(
    1617
    1
                batch,
    1618
    1
                list,
    1619
    1
                &true,
    1620
    1
                vec![Some(false), Some(true), None],
    1621
    1
                Arc::clone(&col_a),
    1622
    1
                &schema
    1623
            );
    1624
    1625
            // expression: "a in ([1, 2, 3], [4, 5, 6], null)"
    1626
    1
            let list = vec![
    1627
    1
                lit([1, 2, 3].as_slice()),
    1628
    1
                lit([4, 5, 6].as_slice()),
    1629
    1
                lit(ScalarValue::LargeBinary(None)),
    1630
            ];
    1631
    1
            in_list!(
    1632
    1
                batch,
    1633
    1
                list.clone(),
    1634
    1
                &false,
    1635
    1
                vec![Some(true), None, None],
    1636
    1
                Arc::clone(&col_a),
    1637
    1
                &schema
    1638
            );
    1639
    1640
            // expression: "a not in ([1, 2, 3], [4, 5, 6], null)"
    1641
    1
            in_list!(
    1642
    1
                batch,
    1643
    1
                list,
    1644
    1
                &true,
    1645
    1
                vec![Some(false), None, None],
    1646
    1
                Arc::clone(&col_a),
    1647
    1
                &schema
    1648
            );
    1649
    1650
    1
            Ok(())
    1651
    1
        }
    1652
    1653
        #[test]
    1654
    1
        fn in_list_binary_view() -> Result<()> {
    1655
    1
            let schema = Schema::new(vec![Field::new("a", DataType::BinaryView, true)]);
    1656
    1
            let a = BinaryViewArray::from(vec![
    1657
    1
                Some([1, 2, 3].as_slice()),
    1658
    1
                Some([1, 2, 2].as_slice()),
    1659
    1
                None,
    1660
            ]);
    1661
    1
            let col_a = col("a", &schema)?0;
    1662
    1
            let batch = RecordBatch::try_new(Arc::new(schema.clone()), vec![Arc::new(a)])?0;
    1663
    1664
            // expression: "a in ([1, 2, 3], [4, 5, 6])"
    1665
    1
            let list = vec![lit([1, 2, 3].as_slice()), lit([4, 5, 6].as_slice())];
    1666
    1
            in_list!(
    1667
    1
                batch,
    1668
    1
                list.clone(),
    1669
    1
                &false,
    1670
    1
                vec![Some(true), Some(false), None],
    1671
    1
                Arc::clone(&col_a),
    1672
    1
                &schema
    1673
            );
    1674
    1675
            // expression: "a not in ([1, 2, 3], [4, 5, 6])"
    1676
    1
            in_list!(
    1677
    1
                batch,
    1678
    1
                list,
    1679
    1
                &true,
    1680
    1
                vec![Some(false), Some(true), None],
    1681
    1
                Arc::clone(&col_a),
    1682
    1
                &schema
    1683
            );
    1684
    1685
            // expression: "a in ([1, 2, 3], [4, 5, 6], null)"
    1686
    1
            let list = vec![
    1687
    1
                lit([1, 2, 3].as_slice()),
    1688
    1
                lit([4, 5, 6].as_slice()),
    1689
    1
                lit(ScalarValue::BinaryView(None)),
    1690
            ];
    1691
    1
            in_list!(
    1692
    1
                batch,
    1693
    1
                list.clone(),
    1694
    1
                &false,
    1695
    1
                vec![Some(true), None, None],
    1696
    1
                Arc::clone(&col_a),
    1697
    1
                &schema
    1698
            );
    1699
    1700
            // expression: "a not in ([1, 2, 3], [4, 5, 6], null)"
    1701
    1
            in_list!(
    1702
    1
                batch,
    1703
    1
                list,
    1704
    1
                &true,
    1705
    1
                vec![Some(false), None, None],
    1706
    1
                Arc::clone(&col_a),
    1707
    1
                &schema
    1708
            );
    1709
    1710
    1
            Ok(())
    1711
    1
        }
    1712
    1713
        #[test]
    1714
    1
        fn in_list_date64() -> Result<()> {
    1715
    1
            let schema = Schema::new(vec![Field::new("a", DataType::Date64, true)]);
    1716
    1
            let a = Date64Array::from(vec![Some(0), Some(2), None]);
    1717
    1
            let col_a = col("a", &schema)?0;
    1718
    1
            let batch = RecordBatch::try_new(Arc::new(schema.clone()), vec![Arc::new(a)])?0;
    1719
    1720
            // expression: "a in (0, 1)"
    1721
    1
            let list = vec![
    1722
    1
                lit(ScalarValue::Date64(Some(0))),
    1723
    1
                lit(ScalarValue::Date64(Some(1))),
    1724
            ];
    1725
    1
            in_list!(
    1726
    1
                batch,
    1727
    1
                list,
    1728
    1
                &false,
    1729
    1
                vec![Some(true), Some(false), None],
    1730
    1
                Arc::clone(&col_a),
    1731
    1
                &schema
    1732
            );
    1733
    1734
            // expression: "a not in (0, 1)"
    1735
    1
            let list = vec![
    1736
    1
                lit(ScalarValue::Date64(Some(0))),
    1737
    1
                lit(ScalarValue::Date64(Some(1))),
    1738
            ];
    1739
    1
            in_list!(
    1740
    1
                batch,
    1741
    1
                list,
    1742
    1
                &true,
    1743
    1
                vec![Some(false), Some(true), None],
    1744
    1
                Arc::clone(&col_a),
    1745
    1
                &schema
    1746
            );
    1747
    1748
            // expression: "a in (0, 1, NULL)"
    1749
    1
            let list = vec![
    1750
    1
                lit(ScalarValue::Date64(Some(0))),
    1751
    1
                lit(ScalarValue::Date64(Some(1))),
    1752
    1
                lit(ScalarValue::Null),
    1753
            ];
    1754
    1
            in_list!(
    1755
    1
                batch,
    1756
    1
                list,
    1757
    1
                &false,
    1758
    1
                vec![Some(true), None, None],
    1759
    1
                Arc::clone(&col_a),
    1760
    1
                &schema
    1761
            );
    1762
    1763
            // expression: "a not in (0, 1, NULL)"
    1764
    1
            let list = vec![
    1765
    1
                lit(ScalarValue::Date64(Some(0))),
    1766
    1
                lit(ScalarValue::Date64(Some(1))),
    1767
    1
                lit(ScalarValue::Null),
    1768
            ];
    1769
    1
            in_list!(
    1770
    1
                batch,
    1771
    1
                list,
    1772
    1
                &true,
    1773
    1
                vec![Some(false), None, None],
    1774
    1
                Arc::clone(&col_a),
    1775
    1
                &schema
    1776
            );
    1777
    1778
    1
            Ok(())
    1779
    1
        }
    1780
    1781
        #[test]
    1782
    1
        fn in_list_date32() -> Result<()> {
    1783
    1
            let schema = Schema::new(vec![Field::new("a", DataType::Date32, true)]);
    1784
    1
            let a = Date32Array::from(vec![Some(0), Some(2), None]);
    1785
    1
            let col_a = col("a", &schema)?0;
    1786
    1
            let batch = RecordBatch::try_new(Arc::new(schema.clone()), vec![Arc::new(a)])?0;
    1787
    1788
            // expression: "a in (0, 1)"
    1789
    1
            let list = vec![
    1790
    1
                lit(ScalarValue::Date32(Some(0))),
    1791
    1
                lit(ScalarValue::Date32(Some(1))),
    1792
            ];
    1793
    1
            in_list!(
    1794
    1
                batch,
    1795
    1
                list,
    1796
    1
                &false,
    1797
    1
                vec![Some(true), Some(false), None],
    1798
    1
                Arc::clone(&col_a),
    1799
    1
                &schema
    1800
            );
    1801
    1802
            // expression: "a not in (0, 1)"
    1803
    1
            let list = vec![
    1804
    1
                lit(ScalarValue::Date32(Some(0))),
    1805
    1
                lit(ScalarValue::Date32(Some(1))),
    1806
            ];
    1807
    1
            in_list!(
    1808
    1
                batch,
    1809
    1
                list,
    1810
    1
                &true,
    1811
    1
                vec![Some(false), Some(true), None],
    1812
    1
                Arc::clone(&col_a),
    1813
    1
                &schema
    1814
            );
    1815
    1816
            // expression: "a in (0, 1, NULL)"
    1817
    1
            let list = vec![
    1818
    1
                lit(ScalarValue::Date32(Some(0))),
    1819
    1
                lit(ScalarValue::Date32(Some(1))),
    1820
    1
                lit(ScalarValue::Null),
    1821
            ];
    1822
    1
            in_list!(
    1823
    1
                batch,
    1824
    1
                list,
    1825
    1
                &false,
    1826
    1
                vec![Some(true), None, None],
    1827
    1
                Arc::clone(&col_a),
    1828
    1
                &schema
    1829
            );
    1830
    1831
            // expression: "a not in (0, 1, NULL)"
    1832
    1
            let list = vec![
    1833
    1
                lit(ScalarValue::Date32(Some(0))),
    1834
    1
                lit(ScalarValue::Date32(Some(1))),
    1835
    1
                lit(ScalarValue::Null),
    1836
            ];
    1837
    1
            in_list!(
    1838
    1
                batch,
    1839
    1
                list,
    1840
    1
                &true,
    1841
    1
                vec![Some(false), None, None],
    1842
    1
                Arc::clone(&col_a),
    1843
    1
                &schema
    1844
            );
    1845
    1846
    1
            Ok(())
    1847
    1
        }
    1848
    1849
        #[test]
    1850
    1
        fn in_list_decimal() -> Result<()> {
    1851
            // Now, we can check the NULL type
    1852
    1
            let schema =
    1853
    1
                Schema::new(vec![Field::new("a", DataType::Decimal128(13, 4), true)]);
    1854
    1
            let array = vec![Some(100_0000_i128), None, Some(200_5000_i128)]
    1855
    1
                .into_iter()
    1856
    1
                .collect::<Decimal128Array>();
    1857
    1
            let array = array.with_precision_and_scale(13, 4).unwrap();
    1858
    1
            let col_a = col("a", &schema)?0;
    1859
    1
            let batch =
    1860
    1
                RecordBatch::try_new(Arc::new(schema.clone()), vec![Arc::new(array)])?0;
    1861
    1862
            // expression: "a in (100,200), the data type of list is INT32
    1863
    1
            let list = vec![lit(100i32), lit(200i32)];
    1864
    1
            in_list!(
    1865
    1
                batch,
    1866
    1
                list,
    1867
    1
                &false,
    1868
    1
                vec![Some(true), None, Some(false)],
    1869
    1
                Arc::clone(&col_a),
    1870
    1
                &schema
    1871
            );
    1872
            // expression: "a not in (100,200)
    1873
    1
            let list = vec![lit(100i32), lit(200i32)];
    1874
    1
            in_list!(
    1875
    1
                batch,
    1876
    1
                list,
    1877
    1
                &true,
    1878
    1
                vec![Some(false), None, Some(true)],
    1879
    1
                Arc::clone(&col_a),
    1880
    1
                &schema
    1881
            );
    1882
    1883
            // expression: "a in (200,NULL), the data type of list is INT32 AND NULL
    1884
    1
            let list = vec![lit(ScalarValue::Int32(Some(100))), lit(ScalarValue::Null)];
    1885
    1
            in_list!(
    1886
    1
                batch,
    1887
    1
                list.clone(),
    1888
    1
                &false,
    1889
    1
                vec![Some(true), None, None],
    1890
    1
                Arc::clone(&col_a),
    1891
    1
                &schema
    1892
            );
    1893
            // expression: "a not in (200,NULL), the data type of list is INT32 AND NULL
    1894
    1
            in_list!(
    1895
    1
                batch,
    1896
    1
                list,
    1897
    1
                &true,
    1898
    1
                vec![Some(false), None, None],
    1899
    1
                Arc::clone(&col_a),
    1900
    1
                &schema
    1901
            );
    1902
    1903
            // expression: "a in (200.5, 100), the data type of list is FLOAT32 and INT32
    1904
    1
            let list = vec![lit(200.50f32), lit(100i32)];
    1905
    1
            in_list!(
    1906
    1
                batch,
    1907
    1
                list,
    1908
    1
                &false,
    1909
    1
                vec![Some(true), None, Some(true)],
    1910
    1
                Arc::clone(&col_a),
    1911
    1
                &schema
    1912
            );
    1913
    1914
            // expression: "a not in (200.5, 100), the data type of list is FLOAT32 and INT32
    1915
    1
            let list = vec![lit(200.50f32), lit(101i32)];
    1916
    1
            in_list!(
    1917
    1
                batch,
    1918
    1
                list,
    1919
    1
                &true,
    1920
    1
                vec![Some(true), None, Some(false)],
    1921
    1
                Arc::clone(&col_a),
    1922
    1
                &schema
    1923
            );
    1924
    1925
            // test the optimization: set
    1926
            // expression: "a in (99..300), the data type of list is INT32
    1927
    1
            let list = (99i32..300).map(lit).collect::<Vec<_>>();
    1928
    1929
    1
            in_list!(
    1930
    1
                batch,
    1931
    1
                list.clone(),
    1932
    1
                &false,
    1933
    1
                vec![Some(true), None, Some(false)],
    1934
    1
                Arc::clone(&col_a),
    1935
    1
                &schema
    1936
            );
    1937
    1938
    1
            in_list!(
    1939
    1
                batch,
    1940
    1
                list,
    1941
    1
                &true,
    1942
    1
                vec![Some(false), None, Some(true)],
    1943
    1
                Arc::clone(&col_a),
    1944
    1
                &schema
    1945
            );
    1946
    1947
    1
            Ok(())
    1948
    1
        }
    1949
    1950
        #[test]
    1951
    1
        fn test_cast_static_filter_to_set() -> Result<()> {
    1952
            // random schema
    1953
    1
            let schema =
    1954
    1
                Schema::new(vec![Field::new("a", DataType::Decimal128(13, 4), true)]);
    1955
    1956
            // list of phy expr
    1957
    1
            let mut phy_exprs = vec![
    1958
    1
                lit(1i64),
    1959
    1
                expressions::cast(lit(2i32), &schema, DataType::Int64)?0,
    1960
    1
                try_cast(lit(3.13f32), &schema, DataType::Int64)?0,
    1961
            ];
    1962
    1
            let static_filter = try_cast_static_filter_to_set(&phy_exprs, &schema).unwrap();
    1963
    1964
    1
            let array = Int64Array::from(vec![1, 2, 3, 4]);
    1965
    1
            let r = static_filter.contains(&array, false).unwrap();
    1966
    1
            assert_eq!(r, BooleanArray::from(vec![true, true, true, false]));
    1967
    1968
    1
            try_cast_static_filter_to_set(&phy_exprs, &schema).unwrap();
    1969
            // cast(cast(lit())), but the cast to the same data type, one case will be ignored
    1970
    1
            phy_exprs.push(expressions::cast(
    1971
    1
                expressions::cast(lit(2i32), &schema, DataType::Int64)?0,
    1972
    1
                &schema,
    1973
    1
                DataType::Int64,
    1974
    0
            )?);
    1975
    1
            try_cast_static_filter_to_set(&phy_exprs, &schema).unwrap();
    1976
    1977
    1
            phy_exprs.clear();
    1978
    1979
            // case(cast(lit())), the cast to the diff data type
    1980
    1
            phy_exprs.push(expressions::cast(
    1981
    1
                expressions::cast(lit(2i32), &schema, DataType::Int64)?0,
    1982
    1
                &schema,
    1983
    1
                DataType::Int32,
    1984
    0
            )?);
    1985
    1
            try_cast_static_filter_to_set(&phy_exprs, &schema).unwrap();
    1986
    1987
            // column
    1988
    1
            phy_exprs.push(col("a", &schema)?0);
    1989
    1
            assert!(try_cast_static_filter_to_set(&phy_exprs, &schema).is_err());
    1990
    1991
    1
            Ok(())
    1992
    1
        }
    1993
    1994
        #[test]
    1995
    1
        fn in_list_timestamp() -> Result<()> {
    1996
    1
            let schema = Schema::new(vec![Field::new(
    1997
                "a",
    1998
    1
                DataType::Timestamp(TimeUnit::Microsecond, None),
    1999
                true,
    2000
            )]);
    2001
    1
            let a = TimestampMicrosecondArray::from(vec![
    2002
    1
                Some(1388588401000000000),
    2003
    1
                Some(1288588501000000000),
    2004
    1
                None,
    2005
            ]);
    2006
    1
            let col_a = col("a", &schema)?0;
    2007
    1
            let batch = RecordBatch::try_new(Arc::new(schema.clone()), vec![Arc::new(a)])?0;
    2008
    2009
    1
            let list = vec![
    2010
    1
                lit(ScalarValue::TimestampMicrosecond(
    2011
    1
                    Some(1388588401000000000),
    2012
    1
                    None,
    2013
    1
                )),
    2014
    1
                lit(ScalarValue::TimestampMicrosecond(
    2015
    1
                    Some(1388588401000000001),
    2016
    1
                    None,
    2017
    1
                )),
    2018
    1
                lit(ScalarValue::TimestampMicrosecond(
    2019
    1
                    Some(1388588401000000002),
    2020
    1
                    None,
    2021
    1
                )),
    2022
            ];
    2023
    2024
    1
            in_list!(
    2025
    1
                batch,
    2026
    1
                list.clone(),
    2027
    1
                &false,
    2028
    1
                vec![Some(true), Some(false), None],
    2029
    1
                Arc::clone(&col_a),
    2030
    1
                &schema
    2031
            );
    2032
    2033
    1
            in_list!(
    2034
    1
                batch,
    2035
    1
                list.clone(),
    2036
    1
                &true,
    2037
    1
                vec![Some(false), Some(true), None],
    2038
    1
                Arc::clone(&col_a),
    2039
    1
                &schema
    2040
            );
    2041
    1
            Ok(())
    2042
    1
        }
    2043
    2044
        #[test]
    2045
    1
        fn in_expr_with_multiple_element_in_list() -> Result<()> {
    2046
    1
            let schema = Schema::new(vec![
    2047
    1
                Field::new("a", DataType::Float64, true),
    2048
    1
                Field::new("b", DataType::Float64, true),
    2049
    1
                Field::new("c", DataType::Float64, true),
    2050
            ]);
    2051
    1
            let a = Float64Array::from(vec![
    2052
    1
                Some(0.0),
    2053
    1
                Some(1.0),
    2054
    1
                Some(2.0),
    2055
    1
                Some(f64::NAN),
    2056
    1
                Some(-f64::NAN),
    2057
            ]);
    2058
    1
            let b = Float64Array::from(vec![
    2059
    1
                Some(8.0),
    2060
    1
                Some(1.0),
    2061
    1
                Some(5.0),
    2062
    1
                Some(f64::NAN),
    2063
    1
                Some(3.0),
    2064
            ]);
    2065
    1
            let c = Float64Array::from(vec![
    2066
    1
                Some(6.0),
    2067
    1
                Some(7.0),
    2068
    1
                None,
    2069
    1
                Some(5.0),
    2070
    1
                Some(-f64::NAN),
    2071
            ]);
    2072
    1
            let col_a = col("a", &schema)?0;
    2073
    1
            let col_b = col("b", &schema)?0;
    2074
    1
            let col_c = col("c", &schema)?0;
    2075
    1
            let batch = RecordBatch::try_new(
    2076
    1
                Arc::new(schema.clone()),
    2077
    1
                vec![Arc::new(a), Arc::new(b), Arc::new(c)],
    2078
    0
            )?;
    2079
    2080
    1
            let list = vec![Arc::clone(&col_b), Arc::clone(&col_c)];
    2081
    1
            in_list!(
    2082
    1
                batch,
    2083
    1
                list.clone(),
    2084
    1
                &false,
    2085
    1
                vec![Some(false), Some(true), None, Some(true), Some(true)],
    2086
    1
                Arc::clone(&col_a),
    2087
    1
                &schema
    2088
            );
    2089
    2090
    1
            in_list!(
    2091
    1
                batch,
    2092
    1
                list,
    2093
    1
                &true,
    2094
    1
                vec![Some(true), Some(false), None, Some(false), Some(false)],
    2095
    1
                Arc::clone(&col_a),
    2096
    1
                &schema
    2097
            );
    2098
    2099
    1
            Ok(())
    2100
    1
        }
    2101
    2102
        macro_rules! test_nullable {
    2103
            ($COL:expr, $LIST:expr, $SCHEMA:expr, $EXPECTED:expr) => {{
    2104
                let (cast_expr, cast_list_exprs) = in_list_cast($COL, $LIST, $SCHEMA)?;
    2105
                let expr = in_list(cast_expr, cast_list_exprs, &false, $SCHEMA).unwrap();
    2106
                let result = expr.nullable($SCHEMA)?;
    2107
                assert_eq!($EXPECTED, result);
    2108
            }};
    2109
        }
    2110
    2111
        #[test]
    2112
    1
        fn in_list_nullable() -> Result<()> {
    2113
    1
            let schema = Schema::new(vec![
    2114
    1
                Field::new("c1_nullable", DataType::Int64, true),
    2115
    1
                Field::new("c2_non_nullable", DataType::Int64, false),
    2116
            ]);
    2117
    2118
    1
            let c1_nullable = col("c1_nullable", &schema)?0;
    2119
    1
            let c2_non_nullable = col("c2_non_nullable", &schema)?0;
    2120
    2121
            // static_filter has no nulls
    2122
    1
            let list = vec![lit(1_i64), lit(2_i64)];
    2123
    1
            test_nullable!(Arc::clone(&c1_nullable), list.clone(), &schema, true);
    2124
    1
            test_nullable!(Arc::clone(&c2_non_nullable), list.clone(), &schema, false);
    2125
    2126
            // static_filter has nulls
    2127
    1
            let list = vec![lit(1_i64), lit(2_i64), lit(ScalarValue::Null)];
    2128
    1
            test_nullable!(Arc::clone(&c1_nullable), list.clone(), &schema, true);
    2129
    1
            test_nullable!(Arc::clone(&c2_non_nullable), list.clone(), &schema, true);
    2130
    2131
    1
            let list = vec![Arc::clone(&c1_nullable)];
    2132
    1
            test_nullable!(Arc::clone(&c2_non_nullable), list.clone(), &schema, true);
    2133
    2134
    1
            let list = vec![Arc::clone(&c2_non_nullable)];
    2135
    1
            test_nullable!(Arc::clone(&c1_nullable), list.clone(), &schema, true);
    2136
    2137
    1
            let list = vec![Arc::clone(&c2_non_nullable), Arc::clone(&c2_non_nullable)];
    2138
    1
            test_nullable!(Arc::clone(&c2_non_nullable), list.clone(), &schema, false);
    2139
    2140
    1
            Ok(())
    2141
    1
        }
    2142
    2143
        #[test]
    2144
    1
        fn in_list_no_cols() -> Result<()> {
    2145
            // test logic when the in_list expression doesn't have any columns
    2146
    1
            let schema = Schema::new(vec![Field::new("a", DataType::Int32, true)]);
    2147
    1
            let a = Int32Array::from(vec![Some(1), Some(2), None]);
    2148
    1
            let batch = RecordBatch::try_new(Arc::new(schema.clone()), vec![Arc::new(a)])?0;
    2149
    2150
    1
            let list = vec![lit(ScalarValue::from(1i32)), lit(ScalarValue::from(6i32))];
    2151
    2152
            // 1 IN (1, 6)
    2153
    1
            let expr = lit(ScalarValue::Int32(Some(1)));
    2154
    1
            in_list!(
    2155
    1
                batch,
    2156
    1
                list.clone(),
    2157
    1
                &false,
    2158
                // should have three outputs, as the input batch has three rows
    2159
    1
                vec![Some(true), Some(true), Some(true)],
    2160
    1
                expr,
    2161
    1
                &schema
    2162
            );
    2163
    2164
            // 2 IN (1, 6)
    2165
    1
            let expr = lit(ScalarValue::Int32(Some(2)));
    2166
    1
            in_list!(
    2167
    1
                batch,
    2168
    1
                list.clone(),
    2169
    1
                &false,
    2170
                // should have three outputs, as the input batch has three rows
    2171
    1
                vec![Some(false), Some(false), Some(false)],
    2172
    1
                expr,
    2173
    1
                &schema
    2174
            );
    2175
    2176
            // NULL IN (1, 6)
    2177
    1
            let expr = lit(ScalarValue::Int32(None));
    2178
    1
            in_list!(
    2179
    1
                batch,
    2180
    1
                list.clone(),
    2181
    1
                &false,
    2182
                // should have three outputs, as the input batch has three rows
    2183
    1
                vec![None, None, None],
    2184
    1
                expr,
    2185
    1
                &schema
    2186
            );
    2187
    2188
    1
            Ok(())
    2189
    1
        }
    2190
    2191
        #[test]
    2192
    1
        fn in_list_utf8_with_dict_types() -> Result<()> {
    2193
    4
            fn dict_lit(key_type: DataType, value: &str) -> Arc<dyn PhysicalExpr> {
    2194
    4
                lit(ScalarValue::Dictionary(
    2195
    4
                    Box::new(key_type),
    2196
    4
                    Box::new(ScalarValue::new_utf8(value.to_string())),
    2197
    4
                ))
    2198
    4
            }
    2199
    2200
    1
            fn null_dict_lit(key_type: DataType) -> Arc<dyn PhysicalExpr> {
    2201
    1
                lit(ScalarValue::Dictionary(
    2202
    1
                    Box::new(key_type),
    2203
    1
                    Box::new(ScalarValue::Utf8(None)),
    2204
    1
                ))
    2205
    1
            }
    2206
    2207
    1
            let schema = Schema::new(vec![Field::new(
    2208
                "a",
    2209
    1
                DataType::Dictionary(Box::new(DataType::UInt16), Box::new(DataType::Utf8)),
    2210
                true,
    2211
            )]);
    2212
    1
            let a: UInt16DictionaryArray =
    2213
    1
                vec![Some("a"), Some("d"), None].into_iter().collect();
    2214
    1
            let col_a = col("a", &schema)?0;
    2215
    1
            let batch = RecordBatch::try_new(Arc::new(schema.clone()), vec![Arc::new(a)])?0;
    2216
    2217
            // expression: "a in ("a", "b")"
    2218
    1
            let lists = [
    2219
    1
                vec![lit("a"), lit("b")],
    2220
    1
                vec![
    2221
    1
                    dict_lit(DataType::Int8, "a"),
    2222
    1
                    dict_lit(DataType::UInt16, "b"),
    2223
    1
                ],
    2224
    1
            ];
    2225
    2
            for list in lists1.iter1() {
    2226
    2
                in_list_raw!(
    2227
    2
                    batch,
    2228
    2
                    list.clone(),
    2229
    2
                    &false,
    2230
    2
                    vec![Some(true), Some(false), None],
    2231
    2
                    Arc::clone(&col_a),
    2232
    2
                    &schema
    2233
                );
    2234
            }
    2235
    2236
            // expression: "a not in ("a", "b")"
    2237
    2
            for list in lists1.iter1() {
    2238
    2
                in_list_raw!(
    2239
    2
                    batch,
    2240
    2
                    list.clone(),
    2241
    2
                    &true,
    2242
    2
                    vec![Some(false), Some(true), None],
    2243
    2
                    Arc::clone(&col_a),
    2244
    2
                    &schema
    2245
                );
    2246
            }
    2247
    2248
            // expression: "a in ("a", "b", null)"
    2249
    1
            let lists = [
    2250
    1
                vec![lit("a"), lit("b"), lit(ScalarValue::Utf8(None))],
    2251
    1
                vec![
    2252
    1
                    dict_lit(DataType::Int8, "a"),
    2253
    1
                    dict_lit(DataType::UInt16, "b"),
    2254
    1
                    null_dict_lit(DataType::UInt16),
    2255
    1
                ],
    2256
    1
            ];
    2257
    2
            for list in lists1.iter1() {
    2258
    2
                in_list_raw!(
    2259
    2
                    batch,
    2260
    2
                    list.clone(),
    2261
    2
                    &false,
    2262
    2
                    vec![Some(true), None, None],
    2263
    2
                    Arc::clone(&col_a),
    2264
    2
                    &schema
    2265
                );
    2266
            }
    2267
    2268
            // expression: "a not in ("a", "b", null)"
    2269
    2
            for list in lists1.iter1() {
    2270
    2
                in_list_raw!(
    2271
    2
                    batch,
    2272
    2
                    list.clone(),
    2273
    2
                    &true,
    2274
    2
                    vec![Some(false), None, None],
    2275
    2
                    Arc::clone(&col_a),
    2276
    2
                    &schema
    2277
                );
    2278
            }
    2279
    2280
    1
            Ok(())
    2281
    1
        }
    2282
    2283
        #[test]
    2284
    1
        fn test_fmt_sql_1() -> Result<()> {
    2285
    1
            let schema = Schema::new(vec![Field::new("a", DataType::Utf8, true)]);
    2286
    1
            let col_a = col("a", &schema)?0;
    2287
    2288
            // Test: a IN ('a', 'b')
    2289
    1
            let list = vec![lit("a"), lit("b")];
    2290
    1
            let expr = in_list(Arc::clone(&col_a), list, &false, &schema)?0;
    2291
    1
            let sql_string = fmt_sql(expr.as_ref()).to_string();
    2292
    1
            let display_string = expr.to_string();
    2293
    1
            assert_snapshot!(sql_string, @"a IN (a, b)");
    2294
    1
            assert_snapshot!(display_string, @"a@0 IN (SET) ([a, b])");
    2295
    1
            Ok(())
    2296
    1
        }
    2297
    2298
        #[test]
    2299
    1
        fn test_fmt_sql_2() -> Result<()> {
    2300
    1
            let schema = Schema::new(vec![Field::new("a", DataType::Utf8, true)]);
    2301
    1
            let col_a = col("a", &schema)?0;
    2302
    2303
            // Test: a NOT IN ('a', 'b')
    2304
    1
            let list = vec![lit("a"), lit("b")];
    2305
    1
            let expr = in_list(Arc::clone(&col_a), list, &true, &schema)?0;
    2306
    1
            let sql_string = fmt_sql(expr.as_ref()).to_string();
    2307
    1
            let display_string = expr.to_string();
    2308
    2309
    1
            assert_snapshot!(sql_string, @"a NOT IN (a, b)");
    2310
    1
            assert_snapshot!(display_string, @"a@0 NOT IN (SET) ([a, b])");
    2311
    1
            Ok(())
    2312
    1
        }
    2313
    2314
        #[test]
    2315
    1
        fn test_fmt_sql_3() -> Result<()> {
    2316
    1
            let schema = Schema::new(vec![Field::new("a", DataType::Utf8, true)]);
    2317
    1
            let col_a = col("a", &schema)?0;
    2318
            // Test: a IN ('a', 'b', NULL)
    2319
    1
            let list = vec![lit("a"), lit("b"), lit(ScalarValue::Utf8(None))];
    2320
    1
            let expr = in_list(Arc::clone(&col_a), list, &false, &schema)?0;
    2321
    1
            let sql_string = fmt_sql(expr.as_ref()).to_string();
    2322
    1
            let display_string = expr.to_string();
    2323
    2324
    1
            assert_snapshot!(sql_string, @"a IN (a, b, NULL)");
    2325
    1
            assert_snapshot!(display_string, @"a@0 IN (SET) ([a, b, NULL])");
    2326
    1
            Ok(())
    2327
    1
        }
    2328
    2329
        #[test]
    2330
    1
        fn test_fmt_sql_4() -> Result<()> {
    2331
    1
            let schema = Schema::new(vec![Field::new("a", DataType::Utf8, true)]);
    2332
    1
            let col_a = col("a", &schema)?0;
    2333
            // Test: a NOT IN ('a', 'b', NULL)
    2334
    1
            let list = vec![lit("a"), lit("b"), lit(ScalarValue::Utf8(None))];
    2335
    1
            let expr = in_list(Arc::clone(&col_a), list, &true, &schema)?0;
    2336
    1
            let sql_string = fmt_sql(expr.as_ref()).to_string();
    2337
    1
            let display_string = expr.to_string();
    2338
    1
            assert_snapshot!(sql_string, @"a NOT IN (a, b, NULL)");
    2339
    1
            assert_snapshot!(display_string, @"a@0 NOT IN (SET) ([a, b, NULL])");
    2340
    1
            Ok(())
    2341
    1
        }
    2342
    2343
        #[test]
    2344
    1
        fn in_list_struct() -> Result<()> {
    2345
            // Create schema with a struct column
    2346
    1
            let struct_fields = Fields::from(vec![
    2347
    1
                Field::new("x", DataType::Int32, false),
    2348
    1
                Field::new("y", DataType::Utf8, false),
    2349
            ]);
    2350
    1
            let schema = Schema::new(vec![Field::new(
    2351
                "a",
    2352
    1
                DataType::Struct(struct_fields.clone()),
    2353
                true,
    2354
            )]);
    2355
    2356
            // Create test data: array of structs
    2357
    1
            let x_array = Arc::new(Int32Array::from(vec![1, 2, 3]));
    2358
    1
            let y_array = Arc::new(StringArray::from(vec!["a", "b", "c"]));
    2359
    1
            let struct_array =
    2360
    1
                StructArray::new(struct_fields.clone(), vec![x_array, y_array], None);
    2361
    2362
    1
            let col_a = col("a", &schema)?0;
    2363
    1
            let batch =
    2364
    1
                RecordBatch::try_new(Arc::new(schema.clone()), vec![Arc::new(struct_array)])?0;
    2365
    2366
            // Create literal structs for the IN list
    2367
            // Struct {x: 1, y: "a"}
    2368
    1
            let struct1 = ScalarValue::Struct(Arc::new(StructArray::new(
    2369
    1
                struct_fields.clone(),
    2370
    1
                vec![
    2371
    1
                    Arc::new(Int32Array::from(vec![1])),
    2372
    1
                    Arc::new(StringArray::from(vec!["a"])),
    2373
    1
                ],
    2374
    1
                None,
    2375
    1
            )));
    2376
    2377
            // Struct {x: 3, y: "c"}
    2378
    1
            let struct3 = ScalarValue::Struct(Arc::new(StructArray::new(
    2379
    1
                struct_fields.clone(),
    2380
    1
                vec![
    2381
    1
                    Arc::new(Int32Array::from(vec![3])),
    2382
    1
                    Arc::new(StringArray::from(vec!["c"])),
    2383
    1
                ],
    2384
    1
                None,
    2385
    1
            )));
    2386
    2387
            // Test: a IN ({1, "a"}, {3, "c"})
    2388
    1
            let list = vec![lit(struct1.clone()), lit(struct3.clone())];
    2389
    1
            in_list_raw!(
    2390
    1
                batch,
    2391
    1
                list.clone(),
    2392
    1
                &false,
    2393
    1
                vec![Some(true), Some(false), Some(true)],
    2394
    1
                Arc::clone(&col_a),
    2395
    1
                &schema
    2396
            );
    2397
    2398
            // Test: a NOT IN ({1, "a"}, {3, "c"})
    2399
    1
            in_list_raw!(
    2400
    1
                batch,
    2401
    1
                list,
    2402
    1
                &true,
    2403
    1
                vec![Some(false), Some(true), Some(false)],
    2404
    1
                Arc::clone(&col_a),
    2405
    1
                &schema
    2406
            );
    2407
    2408
    1
            Ok(())
    2409
    1
        }
    2410
    2411
        #[test]
    2412
    1
        fn in_list_struct_with_nulls() -> Result<()> {
    2413
            // Create schema with a struct column
    2414
    1
            let struct_fields = Fields::from(vec![
    2415
    1
                Field::new("x", DataType::Int32, false),
    2416
    1
                Field::new("y", DataType::Utf8, false),
    2417
            ]);
    2418
    1
            let schema = Schema::new(vec![Field::new(
    2419
                "a",
    2420
    1
                DataType::Struct(struct_fields.clone()),
    2421
                true,
    2422
            )]);
    2423
    2424
            // Create test data with a null struct
    2425
    1
            let x_array = Arc::new(Int32Array::from(vec![1, 2]));
    2426
    1
            let y_array = Arc::new(StringArray::from(vec!["a", "b"]));
    2427
    1
            let struct_array = StructArray::new(
    2428
    1
                struct_fields.clone(),
    2429
    1
                vec![x_array, y_array],
    2430
    1
                Some(NullBuffer::from(vec![true, false])),
    2431
            );
    2432
    2433
    1
            let col_a = col("a", &schema)?0;
    2434
    1
            let batch =
    2435
    1
                RecordBatch::try_new(Arc::new(schema.clone()), vec![Arc::new(struct_array)])?0;
    2436
    2437
            // Create literal struct for the IN list
    2438
    1
            let struct1 = ScalarValue::Struct(Arc::new(StructArray::new(
    2439
    1
                struct_fields.clone(),
    2440
    1
                vec![
    2441
    1
                    Arc::new(Int32Array::from(vec![1])),
    2442
    1
                    Arc::new(StringArray::from(vec!["a"])),
    2443
    1
                ],
    2444
    1
                None,
    2445
    1
            )));
    2446
    2447
            // Test: a IN ({1, "a"})
    2448
    1
            let list = vec![lit(struct1.clone())];
    2449
    1
            in_list_raw!(
    2450
    1
                batch,
    2451
    1
                list.clone(),
    2452
    1
                &false,
    2453
    1
                vec![Some(true), None],
    2454
    1
                Arc::clone(&col_a),
    2455
    1
                &schema
    2456
            );
    2457
    2458
            // Test: a NOT IN ({1, "a"})
    2459
    1
            in_list_raw!(
    2460
    1
                batch,
    2461
    1
                list,
    2462
    1
                &true,
    2463
    1
                vec![Some(false), None],
    2464
    1
                Arc::clone(&col_a),
    2465
    1
                &schema
    2466
            );
    2467
    2468
    1
            Ok(())
    2469
    1
        }
    2470
    2471
        #[test]
    2472
    1
        fn in_list_struct_with_null_in_list() -> Result<()> {
    2473
            // Create schema with a struct column
    2474
    1
            let struct_fields = Fields::from(vec![
    2475
    1
                Field::new("x", DataType::Int32, false),
    2476
    1
                Field::new("y", DataType::Utf8, false),
    2477
            ]);
    2478
    1
            let schema = Schema::new(vec![Field::new(
    2479
                "a",
    2480
    1
                DataType::Struct(struct_fields.clone()),
    2481
                true,
    2482
            )]);
    2483
    2484
            // Create test data
    2485
    1
            let x_array = Arc::new(Int32Array::from(vec![1, 2, 3]));
    2486
    1
            let y_array = Arc::new(StringArray::from(vec!["a", "b", "c"]));
    2487
    1
            let struct_array =
    2488
    1
                StructArray::new(struct_fields.clone(), vec![x_array, y_array], None);
    2489
    2490
    1
            let col_a = col("a", &schema)?0;
    2491
    1
            let batch =
    2492
    1
                RecordBatch::try_new(Arc::new(schema.clone()), vec![Arc::new(struct_array)])?0;
    2493
    2494
            // Create literal structs including a NULL
    2495
    1
            let struct1 = ScalarValue::Struct(Arc::new(StructArray::new(
    2496
    1
                struct_fields.clone(),
    2497
    1
                vec![
    2498
    1
                    Arc::new(Int32Array::from(vec![1])),
    2499
    1
                    Arc::new(StringArray::from(vec!["a"])),
    2500
    1
                ],
    2501
    1
                None,
    2502
    1
            )));
    2503
    2504
    1
            let null_struct = ScalarValue::Struct(Arc::new(StructArray::new_null(
    2505
    1
                struct_fields.clone(),
    2506
    1
                1,
    2507
    1
            )));
    2508
    2509
            // Test: a IN ({1, "a"}, NULL)
    2510
    1
            let list = vec![lit(struct1), lit(null_struct.clone())];
    2511
    1
            in_list_raw!(
    2512
    1
                batch,
    2513
    1
                list.clone(),
    2514
    1
                &false,
    2515
    1
                vec![Some(true), None, None],
    2516
    1
                Arc::clone(&col_a),
    2517
    1
                &schema
    2518
            );
    2519
    2520
            // Test: a NOT IN ({1, "a"}, NULL)
    2521
    1
            in_list_raw!(
    2522
    1
                batch,
    2523
    1
                list,
    2524
    1
                &true,
    2525
    1
                vec![Some(false), None, None],
    2526
    1
                Arc::clone(&col_a),
    2527
    1
                &schema
    2528
            );
    2529
    2530
    1
            Ok(())
    2531
    1
        }
    2532
    2533
        #[test]
    2534
    1
        fn in_list_nested_struct() -> Result<()> {
    2535
            // Create nested struct schema
    2536
    1
            let inner_struct_fields = Fields::from(vec![
    2537
    1
                Field::new("a", DataType::Int32, false),
    2538
    1
                Field::new("b", DataType::Utf8, false),
    2539
            ]);
    2540
    1
            let outer_struct_fields = Fields::from(vec![
    2541
    1
                Field::new(
    2542
                    "inner",
    2543
    1
                    DataType::Struct(inner_struct_fields.clone()),
    2544
                    false,
    2545
                ),
    2546
    1
                Field::new("c", DataType::Int32, false),
    2547
            ]);
    2548
    1
            let schema = Schema::new(vec![Field::new(
    2549
                "x",
    2550
    1
                DataType::Struct(outer_struct_fields.clone()),
    2551
                true,
    2552
            )]);
    2553
    2554
            // Create test data with nested structs
    2555
    1
            let inner1 = Arc::new(StructArray::new(
    2556
    1
                inner_struct_fields.clone(),
    2557
    1
                vec![
    2558
    1
                    Arc::new(Int32Array::from(vec![1, 2])),
    2559
    1
                    Arc::new(StringArray::from(vec!["x", "y"])),
    2560
                ],
    2561
    1
                None,
    2562
            ));
    2563
    1
            let c_array = Arc::new(Int32Array::from(vec![10, 20]));
    2564
    1
            let outer_array =
    2565
    1
                StructArray::new(outer_struct_fields.clone(), vec![inner1, c_array], None);
    2566
    2567
    1
            let col_x = col("x", &schema)?0;
    2568
    1
            let batch =
    2569
    1
                RecordBatch::try_new(Arc::new(schema.clone()), vec![Arc::new(outer_array)])?0;
    2570
    2571
            // Create a nested struct literal matching the first row
    2572
    1
            let inner_match = Arc::new(StructArray::new(
    2573
    1
                inner_struct_fields.clone(),
    2574
    1
                vec![
    2575
    1
                    Arc::new(Int32Array::from(vec![1])),
    2576
    1
                    Arc::new(StringArray::from(vec!["x"])),
    2577
                ],
    2578
    1
                None,
    2579
            ));
    2580
    1
            let outer_match = ScalarValue::Struct(Arc::new(StructArray::new(
    2581
    1
                outer_struct_fields.clone(),
    2582
    1
                vec![inner_match, Arc::new(Int32Array::from(vec![10]))],
    2583
    1
                None,
    2584
    1
            )));
    2585
    2586
            // Test: x IN ({{1, "x"}, 10})
    2587
    1
            let list = vec![lit(outer_match)];
    2588
    1
            in_list_raw!(
    2589
    1
                batch,
    2590
    1
                list.clone(),
    2591
    1
                &false,
    2592
    1
                vec![Some(true), Some(false)],
    2593
    1
                Arc::clone(&col_x),
    2594
    1
                &schema
    2595
            );
    2596
    2597
            // Test: x NOT IN ({{1, "x"}, 10})
    2598
    1
            in_list_raw!(
    2599
    1
                batch,
    2600
    1
                list,
    2601
    1
                &true,
    2602
    1
                vec![Some(false), Some(true)],
    2603
    1
                Arc::clone(&col_x),
    2604
    1
                &schema
    2605
            );
    2606
    2607
    1
            Ok(())
    2608
    1
        }
    2609
    2610
        #[test]
    2611
    1
        fn in_list_struct_with_exprs_not_array() -> Result<()> {
    2612
            // Test InList using expressions (not the array constructor) with structs
    2613
            // By using InListExpr::new directly, we bypass the array optimization
    2614
            // and use the Exprs variant, testing the expression evaluation path
    2615
    2616
            // Create schema with a struct column {x: Int32, y: Utf8}
    2617
    1
            let struct_fields = Fields::from(vec![
    2618
    1
                Field::new("x", DataType::Int32, false),
    2619
    1
                Field::new("y", DataType::Utf8, false),
    2620
            ]);
    2621
    1
            let schema = Schema::new(vec![Field::new(
    2622
                "a",
    2623
    1
                DataType::Struct(struct_fields.clone()),
    2624
                true,
    2625
            )]);
    2626
    2627
            // Create test data: array of structs [{1, "a"}, {2, "b"}, {3, "c"}]
    2628
    1
            let x_array = Arc::new(Int32Array::from(vec![1, 2, 3]));
    2629
    1
            let y_array = Arc::new(StringArray::from(vec!["a", "b", "c"]));
    2630
    1
            let struct_array =
    2631
    1
                StructArray::new(struct_fields.clone(), vec![x_array, y_array], None);
    2632
    2633
    1
            let col_a = col("a", &schema)?0;
    2634
    1
            let batch =
    2635
    1
                RecordBatch::try_new(Arc::new(schema.clone()), vec![Arc::new(struct_array)])?0;
    2636
    2637
            // Create struct literals with the SAME shape (so types are compatible)
    2638
            // Struct {x: 1, y: "a"}
    2639
    1
            let struct1 = ScalarValue::Struct(Arc::new(StructArray::new(
    2640
    1
                struct_fields.clone(),
    2641
    1
                vec![
    2642
    1
                    Arc::new(Int32Array::from(vec![1])),
    2643
    1
                    Arc::new(StringArray::from(vec!["a"])),
    2644
    1
                ],
    2645
    1
                None,
    2646
    1
            )));
    2647
    2648
            // Struct {x: 3, y: "c"}
    2649
    1
            let struct3 = ScalarValue::Struct(Arc::new(StructArray::new(
    2650
    1
                struct_fields.clone(),
    2651
    1
                vec![
    2652
    1
                    Arc::new(Int32Array::from(vec![3])),
    2653
    1
                    Arc::new(StringArray::from(vec!["c"])),
    2654
    1
                ],
    2655
    1
                None,
    2656
    1
            )));
    2657
    2658
            // Create list of struct expressions
    2659
    1
            let list = vec![lit(struct1), lit(struct3)];
    2660
    2661
            // Use InListExpr::new directly (not in_list()) to bypass array optimization
    2662
            // This creates an InList without a static filter
    2663
    1
            let expr = Arc::new(InListExpr::new(Arc::clone(&col_a), list, false, None));
    2664
    2665
            // Verify that the expression doesn't have a static filter
    2666
            // by checking the display string does NOT contain "(SET)"
    2667
    1
            let display_string = expr.to_string();
    2668
    1
            assert!(
    2669
    1
                !display_string.contains("(SET)"),
    2670
    0
                "Expected display string to NOT contain '(SET)' (should use Exprs variant), but got: {display_string}",
    2671
            );
    2672
    2673
            // Evaluate the expression
    2674
    1
            let result = expr.evaluate(&batch)?0.into_array(batch.num_rows())?0;
    2675
    1
            let result = as_boolean_array(&result);
    2676
    2677
            // Expected: first row {1, "a"} matches struct1,
    2678
            //           second row {2, "b"} doesn't match,
    2679
            //           third row {3, "c"} matches struct3
    2680
    1
            let expected = BooleanArray::from(vec![Some(true), Some(false), Some(true)]);
    2681
    1
            assert_eq!(result, &expected);
    2682
    2683
            // Test NOT IN as well
    2684
    1
            let expr_not = Arc::new(InListExpr::new(
    2685
    1
                Arc::clone(&col_a),
    2686
    1
                vec![
    2687
    1
                    lit(ScalarValue::Struct(Arc::new(StructArray::new(
    2688
    1
                        struct_fields.clone(),
    2689
    1
                        vec![
    2690
    1
                            Arc::new(Int32Array::from(vec![1])),
    2691
    1
                            Arc::new(StringArray::from(vec!["a"])),
    2692
    1
                        ],
    2693
    1
                        None,
    2694
    1
                    )))),
    2695
    1
                    lit(ScalarValue::Struct(Arc::new(StructArray::new(
    2696
    1
                        struct_fields.clone(),
    2697
    1
                        vec![
    2698
    1
                            Arc::new(Int32Array::from(vec![3])),
    2699
    1
                            Arc::new(StringArray::from(vec!["c"])),
    2700
    1
                        ],
    2701
    1
                        None,
    2702
    1
                    )))),
    2703
                ],
    2704
                true,
    2705
    1
                None,
    2706
            ));
    2707
    2708
    1
            let result_not = expr_not.evaluate(&batch)?0.into_array(batch.num_rows())?0;
    2709
    1
            let result_not = as_boolean_array(&result_not);
    2710
    2711
    1
            let expected_not = BooleanArray::from(vec![Some(false), Some(true), Some(false)]);
    2712
    1
            assert_eq!(result_not, &expected_not);
    2713
    2714
    1
            Ok(())
    2715
    1
        }
    2716
    2717
        #[test]
    2718
    1
        fn test_in_list_null_handling_comprehensive() -> Result<()> {
    2719
            // Comprehensive test demonstrating SQL three-valued logic for IN expressions
    2720
            // This test explicitly shows all possible outcomes: true, false, and null
    2721
    1
            let schema = Schema::new(vec![Field::new("a", DataType::Int64, true)]);
    2722
    2723
            // Test data: [1, 2, 3, null]
    2724
            // - 1 will match in both lists
    2725
            // - 2 will not match in either list
    2726
            // - 3 will not match in either list
    2727
            // - null is always null
    2728
    1
            let a = Int64Array::from(vec![Some(1), Some(2), Some(3), None]);
    2729
    1
            let col_a = col("a", &schema)?0;
    2730
    1
            let batch = RecordBatch::try_new(Arc::new(schema.clone()), vec![Arc::new(a)])?0;
    2731
    2732
            // Case 1: List WITHOUT null - demonstrates true/false/null outcomes
    2733
            // "a IN (1, 4)" - 1 matches, 2 and 3 don't match, null is null
    2734
    1
            let list = vec![lit(1i64), lit(4i64)];
    2735
    1
            in_list!(
    2736
    1
                batch,
    2737
    1
                list,
    2738
    1
                &false,
    2739
    1
                vec![
    2740
    1
                    Some(true),  // 1 is in the list → true
    2741
    1
                    Some(false), // 2 is not in the list → false
    2742
    1
                    Some(false), // 3 is not in the list → false
    2743
    1
                    None,        // null IN (...) → null (SQL three-valued logic)
    2744
                ],
    2745
    1
                Arc::clone(&col_a),
    2746
    1
                &schema
    2747
            );
    2748
    2749
            // Case 2: List WITH null - demonstrates null propagation for non-matches
    2750
            // "a IN (1, NULL)" - 1 matches (true), 2/3 don't match but list has null (null), null is null
    2751
    1
            let list = vec![lit(1i64), lit(ScalarValue::Int64(None))];
    2752
    1
            in_list!(
    2753
    1
                batch,
    2754
    1
                list,
    2755
    1
                &false,
    2756
    1
                vec![
    2757
    1
                    Some(true), // 1 is in the list → true (found match)
    2758
    1
                    None, // 2 is not in list, but list has NULL → null (might match NULL)
    2759
    1
                    None, // 3 is not in list, but list has NULL → null (might match NULL)
    2760
    1
                    None, // null IN (...) → null (SQL three-valued logic)
    2761
                ],
    2762
    1
                Arc::clone(&col_a),
    2763
    1
                &schema
    2764
            );
    2765
    2766
    1
            Ok(())
    2767
    1
        }
    2768
    2769
        #[test]
    2770
    1
        fn test_in_list_with_only_nulls() -> Result<()> {
    2771
            // Edge case: IN list contains ONLY null values
    2772
    1
            let schema = Schema::new(vec![Field::new("a", DataType::Int64, true)]);
    2773
    1
            let a = Int64Array::from(vec![Some(1), Some(2), None]);
    2774
    1
            let col_a = col("a", &schema)?0;
    2775
    1
            let batch = RecordBatch::try_new(Arc::new(schema.clone()), vec![Arc::new(a)])?0;
    2776
    2777
            // "a IN (NULL, NULL)" - list has only nulls
    2778
    1
            let list = vec![lit(ScalarValue::Int64(None)), lit(ScalarValue::Int64(None))];
    2779
    2780
            // All results should be NULL because:
    2781
            // - Non-null values (1, 2) can't match anything concrete, but list might contain matching value
    2782
            // - NULL value is always NULL in IN expressions
    2783
    1
            in_list!(
    2784
    1
                batch,
    2785
    1
                list.clone(),
    2786
    1
                &false,
    2787
    1
                vec![None, None, None],
    2788
    1
                Arc::clone(&col_a),
    2789
    1
                &schema
    2790
            );
    2791
    2792
            // "a NOT IN (NULL, NULL)" - list has only nulls
    2793
            // All results should still be NULL due to three-valued logic
    2794
    1
            in_list!(
    2795
    1
                batch,
    2796
    1
                list,
    2797
    1
                &true,
    2798
    1
                vec![None, None, None],
    2799
    1
                Arc::clone(&col_a),
    2800
    1
                &schema
    2801
            );
    2802
    2803
    1
            Ok(())
    2804
    1
        }
    2805
    2806
        #[test]
    2807
    1
        fn test_in_list_multiple_nulls_deduplication() -> Result<()> {
    2808
            // Test that multiple NULLs in the list are handled correctly
    2809
            // This verifies deduplication doesn't break null handling
    2810
    1
            let schema = Schema::new(vec![Field::new("a", DataType::Int64, true)]);
    2811
    1
            let col_a = col("a", &schema)?0;
    2812
    2813
            // Create array with multiple nulls: [1, 2, NULL, NULL, 3, NULL]
    2814
    1
            let array = Arc::new(Int64Array::from(vec![
    2815
    1
                Some(1),
    2816
    1
                Some(2),
    2817
    1
                None,
    2818
    1
                None,
    2819
    1
                Some(3),
    2820
    1
                None,
    2821
    1
            ])) as ArrayRef;
    2822
    2823
            // Create InListExpr from array
    2824
    1
            let expr = Arc::new(InListExpr::try_new_from_array(
    2825
    1
                Arc::clone(&col_a),
    2826
    1
                array,
    2827
                false,
    2828
    1
            )?0) as Arc<dyn PhysicalExpr>;
    2829
    2830
            // Create test data: [1, 2, 3, 4, null]
    2831
    1
            let a = Int64Array::from(vec![Some(1), Some(2), Some(3), Some(4), None]);
    2832
    1
            let batch = RecordBatch::try_new(Arc::new(schema.clone()), vec![Arc::new(a)])?0;
    2833
    2834
            // Evaluate the expression
    2835
    1
            let result = expr.evaluate(&batch)?0.into_array(batch.num_rows())?0;
    2836
    1
            let result = as_boolean_array(&result);
    2837
    2838
            // Expected behavior with multiple NULLs in list:
    2839
            // - Values in the list (1,2,3) → true
    2840
            // - Values not in the list (4) → NULL (because list contains NULL)
    2841
            // - NULL input → NULL
    2842
    1
            let expected = BooleanArray::from(vec![
    2843
    1
                Some(true), // 1 is in list
    2844
    1
                Some(true), // 2 is in list
    2845
    1
                Some(true), // 3 is in list
    2846
    1
                None,       // 4 not in list, but list has NULLs
    2847
    1
                None,       // NULL input
    2848
            ]);
    2849
    1
            assert_eq!(result, &expected);
    2850
    2851
    1
            Ok(())
    2852
    1
        }
    2853
    2854
        #[test]
    2855
    1
        fn test_not_in_null_handling_comprehensive() -> Result<()> {
    2856
            // Comprehensive test demonstrating SQL three-valued logic for NOT IN expressions
    2857
            // This test explicitly shows all possible outcomes for NOT IN: true, false, and null
    2858
    1
            let schema = Schema::new(vec![Field::new("a", DataType::Int64, true)]);
    2859
    2860
            // Test data: [1, 2, 3, null]
    2861
    1
            let a = Int64Array::from(vec![Some(1), Some(2), Some(3), None]);
    2862
    1
            let col_a = col("a", &schema)?0;
    2863
    1
            let batch = RecordBatch::try_new(Arc::new(schema.clone()), vec![Arc::new(a)])?0;
    2864
    2865
            // Case 1: List WITHOUT null - demonstrates true/false/null outcomes for NOT IN
    2866
            // "a NOT IN (1, 4)" - 1 matches (false), 2 and 3 don't match (true), null is null
    2867
    1
            let list = vec![lit(1i64), lit(4i64)];
    2868
    1
            in_list!(
    2869
    1
                batch,
    2870
    1
                list,
    2871
    1
                &true,
    2872
    1
                vec![
    2873
    1
                    Some(false), // 1 is in the list → NOT IN returns false
    2874
    1
                    Some(true),  // 2 is not in the list → NOT IN returns true
    2875
    1
                    Some(true),  // 3 is not in the list → NOT IN returns true
    2876
    1
                    None,        // null NOT IN (...) → null (SQL three-valued logic)
    2877
                ],
    2878
    1
                Arc::clone(&col_a),
    2879
    1
                &schema
    2880
            );
    2881
    2882
            // Case 2: List WITH null - demonstrates null propagation for NOT IN
    2883
            // "a NOT IN (1, NULL)" - 1 matches (false), 2/3 don't match but list has null (null), null is null
    2884
    1
            let list = vec![lit(1i64), lit(ScalarValue::Int64(None))];
    2885
    1
            in_list!(
    2886
    1
                batch,
    2887
    1
                list,
    2888
    1
                &true,
    2889
    1
                vec![
    2890
    1
                    Some(false), // 1 is in the list → NOT IN returns false
    2891
    1
                    None, // 2 is not in known values, but list has NULL → null (can't prove it's not in list)
    2892
    1
                    None, // 3 is not in known values, but list has NULL → null (can't prove it's not in list)
    2893
    1
                    None, // null NOT IN (...) → null (SQL three-valued logic)
    2894
                ],
    2895
    1
                Arc::clone(&col_a),
    2896
    1
                &schema
    2897
            );
    2898
    2899
    1
            Ok(())
    2900
    1
        }
    2901
    2902
        #[test]
    2903
    1
        fn test_in_list_null_type_column() -> Result<()> {
    2904
            // Test with a column that has DataType::Null (not just nullable values)
    2905
            // All values in a NullArray are null by definition
    2906
    1
            let schema = Schema::new(vec![Field::new("a", DataType::Null, true)]);
    2907
    1
            let a = NullArray::new(3);
    2908
    1
            let col_a = col("a", &schema)?0;
    2909
    1
            let batch = RecordBatch::try_new(Arc::new(schema.clone()), vec![Arc::new(a)])?0;
    2910
    2911
            // "null_column IN (1, 2)" - comparing Null type against Int64 list
    2912
            // Note: This tests type coercion behavior between Null and Int64
    2913
    1
            let list = vec![lit(1i64), lit(2i64)];
    2914
    2915
            // All results should be NULL because:
    2916
            // - Every value in the column is null (DataType::Null)
    2917
            // - null IN (anything) always returns null per SQL three-valued logic
    2918
    1
            in_list!(
    2919
    1
                batch,
    2920
    1
                list.clone(),
    2921
    1
                &false,
    2922
    1
                vec![None, None, None],
    2923
    1
                Arc::clone(&col_a),
    2924
    1
                &schema
    2925
            );
    2926
    2927
            // "null_column NOT IN (1, 2)"
    2928
            // Same behavior for NOT IN - null NOT IN (anything) is still null
    2929
    1
            in_list!(
    2930
    1
                batch,
    2931
    1
                list,
    2932
    1
                &true,
    2933
    1
                vec![None, None, None],
    2934
    1
                Arc::clone(&col_a),
    2935
    1
                &schema
    2936
            );
    2937
    2938
    1
            Ok(())
    2939
    1
        }
    2940
    2941
        #[test]
    2942
    1
        fn test_in_list_null_type_list() -> Result<()> {
    2943
            // Test with a list that has DataType::Null
    2944
    1
            let schema = Schema::new(vec![Field::new("a", DataType::Int64, true)]);
    2945
    1
            let a = Int64Array::from(vec![Some(1), Some(2), None]);
    2946
    1
            let col_a = col("a", &schema)?0;
    2947
    2948
            // Create a NullArray as the list
    2949
    1
            let null_array = Arc::new(NullArray::new(2)) as ArrayRef;
    2950
    2951
            // Try to create InListExpr with a NullArray list
    2952
            // This tests whether try_new_from_array can handle Null type arrays
    2953
    1
            let expr = Arc::new(InListExpr::try_new_from_array(
    2954
    1
                Arc::clone(&col_a),
    2955
    1
                null_array,
    2956
                false,
    2957
    1
            )?0) as Arc<dyn PhysicalExpr>;
    2958
    1
            let batch = RecordBatch::try_new(Arc::new(schema.clone()), vec![Arc::new(a)])?0;
    2959
    1
            let result = expr.evaluate(&batch)?0.into_array(batch.num_rows())?0;
    2960
    1
            let result = as_boolean_array(&result);
    2961
    2962
            // If it succeeds, all results should be NULL
    2963
            // because the list contains only null type values
    2964
    1
            let expected = BooleanArray::from(vec![None, None, None]);
    2965
    1
            assert_eq!(result, &expected);
    2966
    2967
    1
            Ok(())
    2968
    1
        }
    2969
    2970
        #[test]
    2971
    1
        fn test_in_list_null_type_both() -> Result<()> {
    2972
            // Test when both column and list are DataType::Null
    2973
    1
            let schema = Schema::new(vec![Field::new("a", DataType::Null, true)]);
    2974
    1
            let a = NullArray::new(3);
    2975
    1
            let col_a = col("a", &schema)?0;
    2976
    2977
            // Create a NullArray as the list
    2978
    1
            let null_array = Arc::new(NullArray::new(2)) as ArrayRef;
    2979
    2980
            // Try to create InListExpr with both Null types
    2981
    1
            let expr = Arc::new(InListExpr::try_new_from_array(
    2982
    1
                Arc::clone(&col_a),
    2983
    1
                null_array,
    2984
                false,
    2985
    1
            )?0) as Arc<dyn PhysicalExpr>;
    2986
    2987
    1
            let batch = RecordBatch::try_new(Arc::new(schema.clone()), vec![Arc::new(a)])?0;
    2988
    1
            let result = expr.evaluate(&batch)?0.into_array(batch.num_rows())?0;
    2989
    1
            let result = as_boolean_array(&result);
    2990
    2991
            // If successful, all results should be NULL
    2992
            // null IN [null, null] -> null
    2993
    1
            let expected = BooleanArray::from(vec![None, None, None]);
    2994
    1
            assert_eq!(result, &expected);
    2995
    2996
    1
            Ok(())
    2997
    1
        }
    2998
    2999
        #[test]
    3000
    1
        fn test_in_list_comprehensive_null_handling() -> Result<()> {
    3001
            // Comprehensive test for IN LIST operations with various NULL handling scenarios.
    3002
            // This test covers the key cases validated against DuckDB as the source of truth.
    3003
            //
    3004
            // Note: Some scalar literal tests (like NULL IN (1, 2)) are omitted as they
    3005
            // appear to expose an issue with static filter optimization. These are covered
    3006
            // by existing tests like in_list_no_cols().
    3007
    3008
    1
            let schema = Arc::new(Schema::new(vec![Field::new("b", DataType::Int32, true)]));
    3009
    1
            let col_b = col("b", &schema)?0;
    3010
    1
            let null_i32 = ScalarValue::Int32(None);
    3011
    3012
            // Helper to create a batch
    3013
    16
            let make_batch1 = |values: Vec<Option<i32>>| -> Result<RecordBatch> {
    3014
    16
                let array = Arc::new(Int32Array::from(values));
    3015
    16
                Ok(RecordBatch::try_new(Arc::clone(&schema), vec![array])?0)
    3016
    16
            };
    3017
    3018
            // Helper to run a test
    3019
    1
            let run_test = |batch: &RecordBatch,
    3020
                            expr: Arc<dyn PhysicalExpr>,
    3021
                            list: Vec<Arc<dyn PhysicalExpr>>,
    3022
                            expected: Vec<Option<bool>>|
    3023
    16
             -> Result<()> {
    3024
    16
                let in_expr = in_list(expr, list, &false, schema.as_ref())?0;
    3025
    16
                let result = in_expr.evaluate(batch)?0.into_array(batch.num_rows())?0;
    3026
    16
                let result = as_boolean_array(&result);
    3027
    16
                assert_eq!(result, &BooleanArray::from(expected));
    3028
    16
                Ok(())
    3029
    16
            };
    3030
    3031
            // ========================================================================
    3032
            // COLUMN TESTS - col(b) IN [1, 2]
    3033
            // ========================================================================
    3034
    3035
            // [1] IN (1, 2) => [TRUE]
    3036
    1
            let batch = make_batch(vec![Some(1)])?0;
    3037
    1
            run_test(
    3038
    1
                &batch,
    3039
    1
                Arc::clone(&col_b),
    3040
    1
                vec![lit(1i32), lit(2i32)],
    3041
    1
                vec![Some(true)],
    3042
    1
            )?0;
    3043
    3044
            // [1, 2] IN (1, 2) => [TRUE, TRUE]
    3045
    1
            let batch = make_batch(vec![Some(1), Some(2)])?0;
    3046
    1
            run_test(
    3047
    1
                &batch,
    3048
    1
                Arc::clone(&col_b),
    3049
    1
                vec![lit(1i32), lit(2i32)],
    3050
    1
                vec![Some(true), Some(true)],
    3051
    1
            )?0;
    3052
    3053
            // [3, 4] IN (1, 2) => [FALSE, FALSE]
    3054
    1
            let batch = make_batch(vec![Some(3), Some(4)])?0;
    3055
    1
            run_test(
    3056
    1
                &batch,
    3057
    1
                Arc::clone(&col_b),
    3058
    1
                vec![lit(1i32), lit(2i32)],
    3059
    1
                vec![Some(false), Some(false)],
    3060
    1
            )?0;
    3061
    3062
            // [1, NULL] IN (1, 2) => [TRUE, NULL]
    3063
    1
            let batch = make_batch(vec![Some(1), None])?0;
    3064
    1
            run_test(
    3065
    1
                &batch,
    3066
    1
                Arc::clone(&col_b),
    3067
    1
                vec![lit(1i32), lit(2i32)],
    3068
    1
                vec![Some(true), None],
    3069
    1
            )?0;
    3070
    3071
            // [3, NULL] IN (1, 2) => [FALSE, NULL] (no match, NULL is NULL)
    3072
    1
            let batch = make_batch(vec![Some(3), None])?0;
    3073
    1
            run_test(
    3074
    1
                &batch,
    3075
    1
                Arc::clone(&col_b),
    3076
    1
                vec![lit(1i32), lit(2i32)],
    3077
    1
                vec![Some(false), None],
    3078
    1
            )?0;
    3079
    3080
            // ========================================================================
    3081
            // COLUMN WITH NULL IN LIST - col(b) IN [NULL, 1]
    3082
            // ========================================================================
    3083
    3084
            // [1] IN (NULL, 1) => [TRUE] (found match)
    3085
    1
            let batch = make_batch(vec![Some(1)])?0;
    3086
    1
            run_test(
    3087
    1
                &batch,
    3088
    1
                Arc::clone(&col_b),
    3089
    1
                vec![lit(null_i32.clone()), lit(1i32)],
    3090
    1
                vec![Some(true)],
    3091
    1
            )?0;
    3092
    3093
            // [2] IN (NULL, 1) => [NULL] (no match, but list has NULL)
    3094
    1
            let batch = make_batch(vec![Some(2)])?0;
    3095
    1
            run_test(
    3096
    1
                &batch,
    3097
    1
                Arc::clone(&col_b),
    3098
    1
                vec![lit(null_i32.clone()), lit(1i32)],
    3099
    1
                vec![None],
    3100
    1
            )?0;
    3101
    3102
            // [NULL] IN (NULL, 1) => [NULL]
    3103
    1
            let batch = make_batch(vec![None])?0;
    3104
    1
            run_test(
    3105
    1
                &batch,
    3106
    1
                Arc::clone(&col_b),
    3107
    1
                vec![lit(null_i32.clone()), lit(1i32)],
    3108
    1
                vec![None],
    3109
    1
            )?0;
    3110
    3111
            // ========================================================================
    3112
            // COLUMN WITH ALL NULLS IN LIST - col(b) IN [NULL, NULL]
    3113
            // ========================================================================
    3114
    3115
            // [1] IN (NULL, NULL) => [NULL]
    3116
    1
            let batch = make_batch(vec![Some(1)])?0;
    3117
    1
            run_test(
    3118
    1
                &batch,
    3119
    1
                Arc::clone(&col_b),
    3120
    1
                vec![lit(null_i32.clone()), lit(null_i32.clone())],
    3121
    1
                vec![None],
    3122
    1
            )?0;
    3123
    3124
            // [NULL] IN (NULL, NULL) => [NULL]
    3125
    1
            let batch = make_batch(vec![None])?0;
    3126
    1
            run_test(
    3127
    1
                &batch,
    3128
    1
                Arc::clone(&col_b),
    3129
    1
                vec![lit(null_i32.clone()), lit(null_i32.clone())],
    3130
    1
                vec![None],
    3131
    1
            )?0;
    3132
    3133
            // ========================================================================
    3134
            // LITERAL IN LIST WITH COLUMN - lit(1) IN [2, col(b)]
    3135
            // ========================================================================
    3136
    3137
            // 1 IN (2, [1]) => [TRUE] (matches column value)
    3138
    1
            let batch = make_batch(vec![Some(1)])?0;
    3139
    1
            run_test(
    3140
    1
                &batch,
    3141
    1
                lit(1i32),
    3142
    1
                vec![lit(2i32), Arc::clone(&col_b)],
    3143
    1
                vec![Some(true)],
    3144
    1
            )?0;
    3145
    3146
            // 1 IN (2, [3]) => [FALSE] (no match)
    3147
    1
            let batch = make_batch(vec![Some(3)])?0;
    3148
    1
            run_test(
    3149
    1
                &batch,
    3150
    1
                lit(1i32),
    3151
    1
                vec![lit(2i32), Arc::clone(&col_b)],
    3152
    1
                vec![Some(false)],
    3153
    1
            )?0;
    3154
    3155
            // 1 IN (2, [NULL]) => [NULL] (no match, column is NULL)
    3156
    1
            let batch = make_batch(vec![None])?0;
    3157
    1
            run_test(
    3158
    1
                &batch,
    3159
    1
                lit(1i32),
    3160
    1
                vec![lit(2i32), Arc::clone(&col_b)],
    3161
    1
                vec![None],
    3162
    1
            )?0;
    3163
    3164
            // ========================================================================
    3165
            // COLUMN IN LIST CONTAINING ITSELF - col(b) IN [1, col(b)]
    3166
            // ========================================================================
    3167
    3168
            // [1] IN (1, [1]) => [TRUE] (always matches - either list literal or itself)
    3169
    1
            let batch = make_batch(vec![Some(1)])?0;
    3170
    1
            run_test(
    3171
    1
                &batch,
    3172
    1
                Arc::clone(&col_b),
    3173
    1
                vec![lit(1i32), Arc::clone(&col_b)],
    3174
    1
                vec![Some(true)],
    3175
    1
            )?0;
    3176
    3177
            // [2] IN (1, [2]) => [TRUE] (matches itself)
    3178
    1
            let batch = make_batch(vec![Some(2)])?0;
    3179
    1
            run_test(
    3180
    1
                &batch,
    3181
    1
                Arc::clone(&col_b),
    3182
    1
                vec![lit(1i32), Arc::clone(&col_b)],
    3183
    1
                vec![Some(true)],
    3184
    1
            )?0;
    3185
    3186
            // [NULL] IN (1, [NULL]) => [NULL] (NULL is never equal to anything)
    3187
    1
            let batch = make_batch(vec![None])?0;
    3188
    1
            run_test(
    3189
    1
                &batch,
    3190
    1
                Arc::clone(&col_b),
    3191
    1
                vec![lit(1i32), Arc::clone(&col_b)],
    3192
    1
                vec![None],
    3193
    1
            )?0;
    3194
    3195
    1
            Ok(())
    3196
    1
        }
    3197
    3198
        #[test]
    3199
    1
        fn test_in_list_scalar_literal_cases() -> Result<()> {
    3200
            // Test scalar literal cases (both NULL and non-NULL) to ensure SQL three-valued
    3201
            // logic is correctly implemented. This covers the important case where a scalar
    3202
            // value is tested against a list containing NULL.
    3203
    3204
    1
            let schema = Arc::new(Schema::new(vec![Field::new("b", DataType::Int32, true)]));
    3205
    1
            let null_i32 = ScalarValue::Int32(None);
    3206
    3207
            // Helper to create a batch
    3208
    1
            let make_batch = |values: Vec<Option<i32>>| -> Result<RecordBatch> {
    3209
    1
                let array = Arc::new(Int32Array::from(values));
    3210
    1
                Ok(RecordBatch::try_new(Arc::clone(&schema), vec![array])?0)
    3211
    1
            };
    3212
    3213
            // Helper to run a test
    3214
    1
            let run_test = |batch: &RecordBatch,
    3215
                            expr: Arc<dyn PhysicalExpr>,
    3216
                            list: Vec<Arc<dyn PhysicalExpr>>,
    3217
                            negated: bool,
    3218
                            expected: Vec<Option<bool>>|
    3219
    7
             -> Result<()> {
    3220
    7
                let in_expr = in_list(expr, list, &negated, schema.as_ref())?0;
    3221
    7
                let result = in_expr.evaluate(batch)?0.into_array(batch.num_rows())?0;
    3222
    7
                let result = as_boolean_array(&result);
    3223
    7
                let expected_array = BooleanArray::from(expected);
    3224
    7
                assert_eq!(
    3225
                    result,
    3226
    7
                    &expected_array,
    3227
    0
                    "Expected {:?}, got {:?}",
    3228
                    expected_array,
    3229
    0
                    result.iter().collect::<Vec<_>>()
    3230
                );
    3231
    7
                Ok(())
    3232
    7
            };
    3233
    3234
    1
            let batch = make_batch(vec![Some(1)])?0;
    3235
    3236
            // ========================================================================
    3237
            // NULL LITERAL TESTS
    3238
            // According to SQL semantics, NULL IN (any_list) should always return NULL
    3239
            // ========================================================================
    3240
    3241
            // NULL IN (1, 1) => NULL
    3242
    1
            run_test(
    3243
    1
                &batch,
    3244
    1
                lit(null_i32.clone()),
    3245
    1
                vec![lit(1i32), lit(1i32)],
    3246
    1
                false,
    3247
    1
                vec![None],
    3248
    1
            )?0;
    3249
    3250
            // NULL IN (NULL, 1) => NULL
    3251
    1
            run_test(
    3252
    1
                &batch,
    3253
    1
                lit(null_i32.clone()),
    3254
    1
                vec![lit(null_i32.clone()), lit(1i32)],
    3255
    1
                false,
    3256
    1
                vec![None],
    3257
    1
            )?0;
    3258
    3259
            // NULL IN (NULL, NULL) => NULL
    3260
    1
            run_test(
    3261
    1
                &batch,
    3262
    1
                lit(null_i32.clone()),
    3263
    1
                vec![lit(null_i32.clone()), lit(null_i32.clone())],
    3264
    1
                false,
    3265
    1
                vec![None],
    3266
    1
            )?0;
    3267
    3268
            // ========================================================================
    3269
            // NON-NULL SCALAR LITERALS WITH NULL IN LIST - Int32
    3270
            // When a scalar value is NOT in a list containing NULL, the result is NULL
    3271
            // When a scalar value IS in the list, the result is TRUE (NULL doesn't matter)
    3272
            // ========================================================================
    3273
    3274
            // 3 IN (0, 1, 2, NULL) => NULL (not in list, but list has NULL)
    3275
    1
            run_test(
    3276
    1
                &batch,
    3277
    1
                lit(3i32),
    3278
    1
                vec![lit(0i32), lit(1i32), lit(2i32), lit(null_i32.clone())],
    3279
    1
                false,
    3280
    1
                vec![None],
    3281
    1
            )?0;
    3282
    3283
            // 3 NOT IN (0, 1, 2, NULL) => NULL (not in list, but list has NULL)
    3284
    1
            run_test(
    3285
    1
                &batch,
    3286
    1
                lit(3i32),
    3287
    1
                vec![lit(0i32), lit(1i32), lit(2i32), lit(null_i32.clone())],
    3288
    1
                true,
    3289
    1
                vec![None],
    3290
    1
            )?0;
    3291
    3292
            // 1 IN (0, 1, 2, NULL) => TRUE (found match, NULL doesn't matter)
    3293
    1
            run_test(
    3294
    1
                &batch,
    3295
    1
                lit(1i32),
    3296
    1
                vec![lit(0i32), lit(1i32), lit(2i32), lit(null_i32.clone())],
    3297
    1
                false,
    3298
    1
                vec![Some(true)],
    3299
    1
            )?0;
    3300
    3301
            // 1 NOT IN (0, 1, 2, NULL) => FALSE (found match, NULL doesn't matter)
    3302
    1
            run_test(
    3303
    1
                &batch,
    3304
    1
                lit(1i32),
    3305
    1
                vec![lit(0i32), lit(1i32), lit(2i32), lit(null_i32.clone())],
    3306
    1
                true,
    3307
    1
                vec![Some(false)],
    3308
    1
            )?0;
    3309
    3310
            // ========================================================================
    3311
            // NON-NULL SCALAR LITERALS WITH NULL IN LIST - String
    3312
            // Same semantics as Int32 but with string type
    3313
            // ========================================================================
    3314
    3315
    1
            let schema_str =
    3316
    1
                Arc::new(Schema::new(vec![Field::new("s", DataType::Utf8, true)]));
    3317
    1
            let batch_str = RecordBatch::try_new(
    3318
    1
                Arc::clone(&schema_str),
    3319
    1
                vec![Arc::new(StringArray::from(vec![Some("dummy")]))],
    3320
    0
            )?;
    3321
    1
            let null_str = ScalarValue::Utf8(None);
    3322
    3323
    1
            let run_test_str = |expr: Arc<dyn PhysicalExpr>,
    3324
                                list: Vec<Arc<dyn PhysicalExpr>>,
    3325
                                negated: bool,
    3326
                                expected: Vec<Option<bool>>|
    3327
    4
             -> Result<()> {
    3328
    4
                let in_expr = in_list(expr, list, &negated, schema_str.as_ref())?0;
    3329
    4
                let result = in_expr
    3330
    4
                    .evaluate(&batch_str)?0
    3331
    4
                    .into_array(batch_str.num_rows())?0;
    3332
    4
                let result = as_boolean_array(&result);
    3333
    4
                let expected_array = BooleanArray::from(expected);
    3334
    4
                assert_eq!(
    3335
                    result,
    3336
    4
                    &expected_array,
    3337
    0
                    "Expected {:?}, got {:?}",
    3338
                    expected_array,
    3339
    0
                    result.iter().collect::<Vec<_>>()
    3340
                );
    3341
    4
                Ok(())
    3342
    4
            };
    3343
    3344
            // 'c' IN ('a', 'b', NULL) => NULL (not in list, but list has NULL)
    3345
    1
            run_test_str(
    3346
    1
                lit("c"),
    3347
    1
                vec![lit("a"), lit("b"), lit(null_str.clone())],
    3348
    1
                false,
    3349
    1
                vec![None],
    3350
    1
            )?0;
    3351
    3352
            // 'c' NOT IN ('a', 'b', NULL) => NULL (not in list, but list has NULL)
    3353
    1
            run_test_str(
    3354
    1
                lit("c"),
    3355
    1
                vec![lit("a"), lit("b"), lit(null_str.clone())],
    3356
    1
                true,
    3357
    1
                vec![None],
    3358
    1
            )?0;
    3359
    3360
            // 'a' IN ('a', 'b', NULL) => TRUE (found match, NULL doesn't matter)
    3361
    1
            run_test_str(
    3362
    1
                lit("a"),
    3363
    1
                vec![lit("a"), lit("b"), lit(null_str.clone())],
    3364
    1
                false,
    3365
    1
                vec![Some(true)],
    3366
    1
            )?0;
    3367
    3368
            // 'a' NOT IN ('a', 'b', NULL) => FALSE (found match, NULL doesn't matter)
    3369
    1
            run_test_str(
    3370
    1
                lit("a"),
    3371
    1
                vec![lit("a"), lit("b"), lit(null_str.clone())],
    3372
    1
                true,
    3373
    1
                vec![Some(false)],
    3374
    1
            )?0;
    3375
    3376
    1
            Ok(())
    3377
    1
        }
    3378
    3379
        #[test]
    3380
    1
        fn test_in_list_tuple_cases() -> Result<()> {
    3381
            // Test tuple/struct cases from the original request: (lit, lit) IN (lit, lit)
    3382
            // These test row-wise comparisons like (1, 2) IN ((1, 2), (3, 4))
    3383
    3384
    1
            let schema = Arc::new(Schema::new(vec![Field::new("b", DataType::Int32, true)]));
    3385
    3386
            // Helper to create struct scalars for tuple comparisons
    3387
    22
            let make_struct1 = |v1: Option<i32>, v2: Option<i32>| -> ScalarValue {
    3388
    22
                let fields = Fields::from(vec![
    3389
    22
                    Field::new("field_0", DataType::Int32, true),
    3390
    22
                    Field::new("field_1", DataType::Int32, true),
    3391
                ]);
    3392
    22
                ScalarValue::Struct(Arc::new(StructArray::new(
    3393
    22
                    fields,
    3394
    22
                    vec![
    3395
    22
                        Arc::new(Int32Array::from(vec![v1])),
    3396
    22
                        Arc::new(Int32Array::from(vec![v2])),
    3397
    22
                    ],
    3398
    22
                    None,
    3399
    22
                )))
    3400
    22
            };
    3401
    3402
            // Need a single row batch for scalar tests
    3403
    1
            let batch = RecordBatch::try_new(
    3404
    1
                Arc::clone(&schema),
    3405
    1
                vec![Arc::new(Int32Array::from(vec![Some(1)]))],
    3406
    0
            )?;
    3407
    3408
            // Helper to run tuple tests
    3409
    1
            let run_tuple_test = |lhs: ScalarValue,
    3410
                                  list: Vec<ScalarValue>,
    3411
                                  expected: Vec<Option<bool>>|
    3412
    11
             -> Result<()> {
    3413
    11
                let expr = in_list(
    3414
    11
                    lit(lhs),
    3415
    11
                    list.into_iter().map(lit).collect(),
    3416
    11
                    &false,
    3417
    11
                    schema.as_ref(),
    3418
    0
                )?;
    3419
    11
                let result = expr.evaluate(&batch)?0.into_array(batch.num_rows())?0;
    3420
    11
                let result = as_boolean_array(&result);
    3421
    11
                assert_eq!(result, &BooleanArray::from(expected));
    3422
    11
                Ok(())
    3423
    11
            };
    3424
    3425
            // (NULL, NULL) IN ((1, 2)) => FALSE (tuples don't match)
    3426
    1
            run_tuple_test(
    3427
    1
                make_struct(None, None),
    3428
    1
                vec![make_struct(Some(1), Some(2))],
    3429
    1
                vec![Some(false)],
    3430
    1
            )?0;
    3431
    3432
            // (NULL, NULL) IN ((NULL, 1)) => FALSE
    3433
    1
            run_tuple_test(
    3434
    1
                make_struct(None, None),
    3435
    1
                vec![make_struct(None, Some(1))],
    3436
    1
                vec![Some(false)],
    3437
    1
            )?0;
    3438
    3439
            // (NULL, NULL) IN ((NULL, NULL)) => TRUE (exact match including nulls)
    3440
    1
            run_tuple_test(
    3441
    1
                make_struct(None, None),
    3442
    1
                vec![make_struct(None, None)],
    3443
    1
                vec![Some(true)],
    3444
    1
            )?0;
    3445
    3446
            // (NULL, 1) IN ((1, 2)) => FALSE
    3447
    1
            run_tuple_test(
    3448
    1
                make_struct(None, Some(1)),
    3449
    1
                vec![make_struct(Some(1), Some(2))],
    3450
    1
                vec![Some(false)],
    3451
    1
            )?0;
    3452
    3453
            // (NULL, 1) IN ((NULL, 1)) => TRUE (exact match)
    3454
    1
            run_tuple_test(
    3455
    1
                make_struct(None, Some(1)),
    3456
    1
                vec![make_struct(None, Some(1))],
    3457
    1
                vec![Some(true)],
    3458
    1
            )?0;
    3459
    3460
            // (NULL, 1) IN ((NULL, NULL)) => FALSE
    3461
    1
            run_tuple_test(
    3462
    1
                make_struct(None, Some(1)),
    3463
    1
                vec![make_struct(None, None)],
    3464
    1
                vec![Some(false)],
    3465
    1
            )?0;
    3466
    3467
            // (1, 2) IN ((1, 2)) => TRUE
    3468
    1
            run_tuple_test(
    3469
    1
                make_struct(Some(1), Some(2)),
    3470
    1
                vec![make_struct(Some(1), Some(2))],
    3471
    1
                vec![Some(true)],
    3472
    1
            )?0;
    3473
    3474
            // (1, 3) IN ((1, 2)) => FALSE
    3475
    1
            run_tuple_test(
    3476
    1
                make_struct(Some(1), Some(3)),
    3477
    1
                vec![make_struct(Some(1), Some(2))],
    3478
    1
                vec![Some(false)],
    3479
    1
            )?0;
    3480
    3481
            // (4, 4) IN ((1, 2)) => FALSE
    3482
    1
            run_tuple_test(
    3483
    1
                make_struct(Some(4), Some(4)),
    3484
    1
                vec![make_struct(Some(1), Some(2))],
    3485
    1
                vec![Some(false)],
    3486
    1
            )?0;
    3487
    3488
            // (1, 1) IN ((NULL, 1)) => FALSE
    3489
    1
            run_tuple_test(
    3490
    1
                make_struct(Some(1), Some(1)),
    3491
    1
                vec![make_struct(None, Some(1))],
    3492
    1
                vec![Some(false)],
    3493
    1
            )?0;
    3494
    3495
            // (1, 1) IN ((NULL, NULL)) => FALSE
    3496
    1
            run_tuple_test(
    3497
    1
                make_struct(Some(1), Some(1)),
    3498
    1
                vec![make_struct(None, None)],
    3499
    1
                vec![Some(false)],
    3500
    1
            )?0;
    3501
    3502
    1
            Ok(())
    3503
    1
        }
    3504
    }