vortex-array 0.88.0

// SPDX-License-Identifier: Apache-2.0
// SPDX-FileCopyrightText: Copyright the Vortex contributors

use std::sync::Arc;

use itertools::Itertools;
use vortex_buffer::BitBuffer;
use vortex_buffer::Buffer;
use vortex_buffer::BufferAllocatorRef;
use vortex_buffer::BufferString;
use vortex_buffer::ByteBuffer;
use vortex_buffer::buffer;
use vortex_error::VortexExpect;
use vortex_error::VortexResult;

use crate::ArrayRef;
use crate::Canonical;
use crate::ExecutionCtx;
use crate::IntoArray;
use crate::array::ArrayView;
use crate::arrays::BoolArray;
use crate::arrays::ChunkedArray;
use crate::arrays::Constant;
use crate::arrays::ConstantArray;
use crate::arrays::DecimalArray;
use crate::arrays::ExtensionArray;
use crate::arrays::FixedSizeListArray;
use crate::arrays::ListViewArray;
use crate::arrays::MapArray;
use crate::arrays::NullArray;
use crate::arrays::PrimitiveArray;
use crate::arrays::StructArray;
use crate::arrays::UnionArray;
use crate::arrays::VarBinViewArray;
use crate::arrays::VariantArray;
use crate::arrays::varbinview::BinaryView;
use crate::builders::builder_with_capacity_in;
use crate::dtype::DType;
use crate::dtype::DecimalType;
use crate::dtype::Nullability;
use crate::match_each_decimal_value;
use crate::match_each_decimal_value_type;
use crate::match_each_native_ptype;
use crate::match_smallest_list_offset_type;
use crate::scalar::DecimalValue;
use crate::scalar::Scalar;
use crate::validity::Validity;

/// Shared implementation for both `canonicalize` and `execute` methods.
pub(crate) fn constant_canonicalize(
    array: ArrayView<'_, Constant>,
    ctx: &mut ExecutionCtx,
) -> VortexResult<Canonical> {
    let scalar = array.scalar();

    let validity = match array.dtype().nullability() {
        Nullability::NonNullable => Validity::NonNullable,
        Nullability::Nullable => match scalar.is_null() {
            true => Validity::AllInvalid,
            false => Validity::AllValid,
        },
    };

    Ok(match array.dtype() {
        DType::Null => Canonical::Null(NullArray::new(array.len())),
        DType::Bool(..) => Canonical::Bool(BoolArray::new(
            if scalar.as_bool().value().unwrap_or_default() {
                BitBuffer::new_set(array.len())
            } else {
                BitBuffer::new_unset(array.len())
            },
            validity,
        )),
        DType::Primitive(ptype, ..) => {
            match_each_native_ptype!(ptype, |P| {
                Canonical::Primitive(PrimitiveArray::new(
                    if scalar.is_valid() {
                        Buffer::full(
                            P::try_from(scalar)
                                .vortex_expect("Couldn't unwrap scalar to primitive"),
                            array.len(),
                        )
                    } else {
                        Buffer::zeroed(array.len())
                    },
                    validity,
                ))
            })
        }
        DType::Decimal(decimal_type, ..) => {
            let size = DecimalType::smallest_decimal_value_type(decimal_type);
            let decimal = scalar.as_decimal();
            let Some(value) = decimal.decimal_value() else {
                let all_null = match_each_decimal_value_type!(size, |D| {
                    // SAFETY: All-null decimal arrays with zeroed buffers and matching validity.
                    unsafe {
                        DecimalArray::new_unchecked(
                            Buffer::<D>::zeroed(array.len()),
                            *decimal_type,
                            validity,
                        )
                    }
                });
                return Ok(Canonical::Decimal(all_null));
            };

            let decimal_array = match_each_decimal_value!(value, |value| {
                // SAFETY: Constant decimal values with correct type and validity.
                unsafe {
                    DecimalArray::new_unchecked(
                        Buffer::full(value, array.len()),
                        *decimal_type,
                        validity,
                    )
                }
            });
            Canonical::Decimal(decimal_array)
        }
        DType::Utf8(_) => {
            let value = scalar
                .as_utf8()
                .value()
                .cloned()
                .map(BufferString::into_inner);
            Canonical::VarBinView(constant_canonical_byte_view(
                value,
                array.dtype(),
                array.len(),
            ))
        }
        DType::Binary(_) => {
            let value = scalar.as_binary().value().cloned();
            Canonical::VarBinView(constant_canonical_byte_view(
                value,
                array.dtype(),
                array.len(),
            ))
        }
        DType::List(..) => Canonical::List(constant_canonical_list_array(
            scalar,
            array.len(),
            ctx.allocator(),
        )),
        DType::Map(map_dtype, nullability) => {
            let entries_scalar = Scalar::try_new(
                DType::List(Arc::new(map_dtype.entries_dtype()), *nullability),
                scalar.value().cloned(),
            )?;
            Canonical::Map(MapArray::try_new(
                map_dtype.clone(),
                constant_canonical_list_array(&entries_scalar, array.len(), ctx.allocator()),
            )?)
        }
        DType::FixedSizeList(element_dtype, list_size, _) => {
            let value = scalar.as_list();

            Canonical::FixedSizeList(constant_canonical_fixed_size_list_array(
                value.elements(),
                element_dtype,
                *list_size,
                value.dtype().nullability(),
                array.len(),
                ctx.allocator(),
            ))
        }
        DType::Struct(struct_dtype, _) => {
            let value = scalar.as_struct();
            let fields: Vec<_> = match value.fields_iter() {
                Some(fields) => fields
                    .into_iter()
                    .map(|s| ConstantArray::new(s, array.len()).into_array())
                    .collect(),
                None => {
                    assert!(matches!(validity, Validity::AllInvalid));
                    // The struct is entirely null, so fields just need placeholder values with the
                    // correct dtype. We use `default_value` which returns a zero for non-nullable
                    // dtypes and null for nullable dtypes, preserving each field's nullability.
                    struct_dtype
                        .fields()
                        .map(|dt| {
                            let scalar = Scalar::default_value(&dt);
                            ConstantArray::new(scalar, array.len()).into_array()
                        })
                        .collect()
                }
            };
            // SAFETY: Fields are constructed from the same struct scalar, all have same
            // length, dtypes match by construction.
            Canonical::Struct(unsafe {
                StructArray::new_unchecked(fields, struct_dtype.clone(), array.len(), validity)
            })
        }
        DType::Union(..) => Canonical::Union(UnionArray::constant(scalar, array.len())?),
        DType::Variant(_) => Canonical::Variant(VariantArray::try_new(
            array.array().clone().into_array(),
            None,
        )?),
        DType::Extension(ext_dtype) => {
            let s = scalar.as_extension();

            let storage_scalar = s.to_storage_scalar();

            // NB: We need to execute the constant array to be canonical because there is a
            // reduction rule that turns `Extension(Constant(..))` into `Constant(Extension(..))`,
            // and if we don't do this we create an infinite cycle.
            // See `ExtensionConstantRule` for more details.
            let storage_self = ConstantArray::new(storage_scalar, array.len())
                .into_array()
                .execute::<Canonical>(ctx)?
                .into_array();

            Canonical::Extension(ExtensionArray::new(ext_dtype.clone(), storage_self))
        }
    })
}

/// Builds the canonical view array for a constant string or binary run.
fn constant_canonical_byte_view(
    scalar_bytes: Option<ByteBuffer>,
    dtype: &DType,
    len: usize,
) -> VarBinViewArray {
    match scalar_bytes {
        None => {
            let views = buffer![BinaryView::empty_view(); len];

            // SAFETY: for all-null the views and buffers are just zeroed, never accessed.
            unsafe {
                VarBinViewArray::new_unchecked(
                    views,
                    Default::default(),
                    dtype.clone(),
                    Validity::AllInvalid,
                )
            }
        }
        Some(scalar_bytes) => {
            // Create a view to hold the scalar bytes.
            let view = BinaryView::make_view(scalar_bytes.as_slice(), 0, 0);
            // A value short enough to inline lives entirely in its view, so only a longer one
            // needs a data buffer. Adopt the scalar's own rather than copying it.
            let mut buffers = Vec::new();
            if scalar_bytes.len() > BinaryView::MAX_INLINED_SIZE {
                buffers.push(scalar_bytes);
            }

            // Clone our constant view `len` times.
            let views = buffer![view; len];

            // SAFETY: all the views are identical and point to a constant value.
            unsafe {
                VarBinViewArray::new_unchecked(
                    views,
                    Arc::from(buffers),
                    dtype.clone(),
                    Validity::from(dtype.nullability()),
                )
            }
        }
    }
}

/// Creates a [`ListViewArray`] with constant values.
///
/// We basically just project the list scalar value into list view components. If the caller wants
/// a fully decompressed and non-overlapping array, they can rebuild the array.
fn constant_canonical_list_array(
    scalar: &Scalar,
    len: usize,
    allocator: &BufferAllocatorRef,
) -> ListViewArray {
    let list = scalar.as_list();

    // Since "canonicalize" only applies to the top level array, we can simply have 1 scalar in our
    // child `elements` and have all list views point to that scalar.
    let elements = if let Some(elements) = list.elements() {
        // Extract the list elements out of the scalar into a new array.
        let mut builder = builder_with_capacity_in(
            list.dtype()
                .as_list_element_opt()
                .vortex_expect("list scalar somehow did not have a list DType"),
            list.len(),
            allocator,
        );
        for scalar in &elements {
            builder
                .append_scalar(scalar)
                .vortex_expect("list element scalar was invalid");
        }
        builder.finish()
    } else {
        // Otherwise all values are null, and we don't need to store anything in our `elements`.
        Canonical::empty(list.element_dtype()).into_array()
    };

    let validity = if scalar.dtype().is_nullable() {
        if list.is_null() {
            Validity::AllInvalid
        } else {
            Validity::AllValid
        }
    } else {
        debug_assert!(!list.is_null());
        Validity::NonNullable
    };

    // Every row has the same offset and size, so use the narrowest width that fits the list.
    let (offsets, sizes) = match_smallest_list_offset_type!(list.len(), |O| {
        let size = O::try_from(list.len()).vortex_expect("list length fits the chosen offset type");
        (
            ConstantArray::new::<O>(O::default(), len).into_array(),
            ConstantArray::new::<O>(size, len).into_array(),
        )
    });

    debug_assert!(!offsets.dtype().is_nullable());
    debug_assert!(!sizes.dtype().is_nullable());

    // SAFETY: All views point to the same range [0, list.len()) in the elements array.
    // The elements array contains `len` copies of the same value, offsets are all 0,
    // and sizes are all equal to the list length. The validity matches the scalar's nullability.
    unsafe { ListViewArray::new_unchecked(elements, offsets, sizes, validity) }
}

/// Creates a [`FixedSizeListArray`] whose every row holds the same list.
fn constant_canonical_fixed_size_list_array(
    values: Option<Vec<Scalar>>,
    element_dtype: &DType,
    list_size: u32,
    list_nullability: Nullability,
    len: usize,
    allocator: &BufferAllocatorRef,
) -> FixedSizeListArray {
    let elements_len = list_size as usize * len;

    let (elements, validity) = match values {
        // A null list's elements are all placeholders, so one constant array covers the run.
        None => (
            ConstantArray::new(Scalar::default_value(element_dtype), elements_len).into_array(),
            Validity::AllInvalid,
        ),
        Some(values) => (
            tile_fixed_size_list_elements(&values, element_dtype, len, elements_len, allocator),
            Validity::from(list_nullability),
        ),
    };

    // SAFETY: `elements` holds exactly `list_size * len` values, and the validity carries no
    // length of its own.
    unsafe { FixedSizeListArray::new_unchecked(elements, list_size, validity, len) }
}

/// Tiles one row's `values` across a run of `len` rows, storing them once.
///
/// A fixed-size list holds its elements back to back, so the rows cannot share one range the way a
/// list view's can. Uniform elements stay a [`ConstantArray`] covering the whole run; otherwise the
/// row materializes once and the run chunks that single copy.
fn tile_fixed_size_list_elements(
    values: &[Scalar],
    element_dtype: &DType,
    len: usize,
    elements_len: usize,
    allocator: &BufferAllocatorRef,
) -> ArrayRef {
    // An empty run, or a degenerate `list_size == 0`.
    if elements_len == 0 {
        return Canonical::empty(element_dtype).into_array();
    }

    match values.iter().all_equal_value() {
        Ok(uniform) => ConstantArray::new(uniform.clone(), elements_len).into_array(),
        Err(_) => {
            let mut elements_builder =
                builder_with_capacity_in(element_dtype, values.len(), allocator);
            for value in values {
                elements_builder
                    .append_scalar(value)
                    .vortex_expect("fixed-size-list element scalar was invalid");
            }
            let tile = elements_builder.finish();

            if len == 1 {
                return tile;
            }

            // SAFETY: every chunk is `tile` itself, so they share its dtype and none is empty.
            unsafe {
                ChunkedArray::new_unchecked(
                    std::iter::repeat_n(tile, len).collect::<Vec<_>>(),
                    element_dtype.clone(),
                )
            }
            .into_array()
        }
    }
}

#[cfg(test)]
mod tests {
    use std::sync::Arc;
    use std::sync::LazyLock;

    use enum_iterator::all;
    use itertools::Itertools;
    use rstest::rstest;
    use vortex_error::VortexExpect;
    use vortex_error::VortexResult;
    use vortex_session::VortexSession;

    use crate::Canonical;
    use crate::IntoArray;
    use crate::VortexSessionExecute;
    use crate::arrays::Chunked;
    use crate::arrays::Constant;
    use crate::arrays::ConstantArray;
    use crate::arrays::FixedSizeListArray;
    use crate::arrays::ListViewArray;
    use crate::arrays::NullArray;
    use crate::arrays::PrimitiveArray;
    use crate::arrays::StructArray;
    use crate::arrays::VarBinArray;
    use crate::arrays::VarBinViewArray;
    use crate::arrays::chunked::ChunkedArrayExt;
    use crate::arrays::fixed_size_list::FixedSizeListArrayExt;
    use crate::arrays::fixed_size_list::FixedSizeListArraySlotsExt;
    use crate::arrays::listview::ListViewArraySlotsExt;
    use crate::arrays::listview::ListViewRebuildMode;
    use crate::arrays::struct_::StructArrayExt;
    use crate::assert_arrays_eq;
    use crate::dtype::DType;
    use crate::dtype::Nullability;
    use crate::dtype::PType;
    use crate::dtype::half::f16;
    use crate::expr::stats::Stat;
    use crate::expr::stats::StatsProvider;
    use crate::scalar::Scalar;
    use crate::validity::Validity;

    /// A shared session for these constant-array tests, used to create execution contexts.
    static SESSION: LazyLock<VortexSession> = LazyLock::new(crate::array_session);

    #[test]
    fn test_canonicalize_null() {
        let mut ctx = SESSION.create_execution_ctx();
        let const_null = ConstantArray::new(Scalar::null(DType::Null), 42);
        let actual = const_null
            .as_array()
            .clone()
            .execute::<NullArray>(&mut ctx)
            .unwrap();
        assert_eq!(actual.len(), 42);
        assert_eq!(
            actual.execute_scalar(33, &mut ctx).unwrap(),
            Scalar::null(DType::Null)
        );
    }

    #[test]
    fn test_canonicalize_const_str() {
        let mut ctx = SESSION.create_execution_ctx();
        let const_array = ConstantArray::new("four".to_string(), 4);

        let expected = VarBinArray::from(vec!["four", "four", "four", "four"]);
        assert_arrays_eq!(const_array, expected, &mut ctx);
    }

    #[test]
    fn test_canonicalize_propagates_stats() -> VortexResult<()> {
        let mut ctx = SESSION.create_execution_ctx();
        let scalar = Scalar::bool(true, Nullability::NonNullable);
        let const_array = ConstantArray::new(scalar, 4).into_array();
        let stats = const_array
            .statistics()
            .compute_all(&all::<Stat>().collect_vec(), &mut ctx)?;
        let canonical = const_array.execute::<Canonical>(&mut ctx)?.into_array();
        let canonical_stats = canonical.statistics();

        let stats_ref = stats.as_typed_ref(canonical.dtype());

        for stat in all::<Stat>() {
            if stat.dtype(canonical.dtype()).is_none() {
                continue;
            }
            assert_eq!(
                canonical_stats.get(stat),
                stats_ref.get(stat),
                "stat mismatch {stat}"
            );
        }
        Ok(())
    }

    #[test]
    fn test_canonicalize_scalar_values() {
        let mut ctx = SESSION.create_execution_ctx();
        let f16_value = f16::from_f32(5.722046e-6);
        let f16_scalar = Scalar::primitive(f16_value, Nullability::NonNullable);

        // Create a ConstantArray with the f16 scalar
        let const_array = ConstantArray::new(f16_scalar.clone(), 1).into_array();
        let canonical_const = const_array.execute::<PrimitiveArray>(&mut ctx).unwrap();

        // Verify the scalar value is preserved through canonicalization
        assert_eq!(
            canonical_const.execute_scalar(0, &mut ctx).unwrap(),
            f16_scalar
        );
    }

    #[test]
    fn test_canonicalize_lists() -> VortexResult<()> {
        let mut ctx = SESSION.create_execution_ctx();
        let list_scalar = Scalar::list(
            Arc::new(DType::Primitive(PType::U64, Nullability::NonNullable)),
            vec![1u64.into(), 2u64.into()],
            Nullability::NonNullable,
        );
        let const_array = ConstantArray::new(list_scalar, 2).into_array();
        let canonical_const = const_array.execute::<ListViewArray>(&mut ctx)?;
        let list_array =
            canonical_const.rebuild(ListViewRebuildMode::MakeZeroCopyToList, &mut ctx)?;
        assert_arrays_eq!(
            list_array
                .elements()
                .clone()
                .execute::<PrimitiveArray>(&mut ctx)?,
            PrimitiveArray::from_iter([1u64, 2, 1, 2]),
            &mut ctx
        );
        assert_arrays_eq!(
            list_array
                .offsets()
                .clone()
                .execute::<PrimitiveArray>(&mut ctx)?,
            PrimitiveArray::from_iter([0u32, 2]),
            &mut ctx
        );
        assert_arrays_eq!(
            list_array
                .sizes()
                .clone()
                .execute::<PrimitiveArray>(&mut ctx)?,
            PrimitiveArray::from_iter([2u32, 2]),
            &mut ctx
        );
        Ok(())
    }

    #[test]
    fn test_canonicalize_empty_list() {
        let mut ctx = SESSION.create_execution_ctx();
        let list_scalar = Scalar::list(
            Arc::new(DType::Primitive(PType::U64, Nullability::NonNullable)),
            vec![],
            Nullability::NonNullable,
        );
        let const_array = ConstantArray::new(list_scalar, 2).into_array();
        let canonical_const = const_array.execute::<ListViewArray>(&mut ctx).unwrap();
        let elements_prim = canonical_const
            .elements()
            .clone()
            .execute::<PrimitiveArray>(&mut ctx)
            .unwrap();
        assert!(elements_prim.is_empty());
        assert_arrays_eq!(
            canonical_const
                .offsets()
                .clone()
                .execute::<PrimitiveArray>(&mut ctx)
                .unwrap(),
            PrimitiveArray::from_iter([0u32, 0]),
            &mut ctx
        );
        assert_arrays_eq!(
            canonical_const
                .sizes()
                .clone()
                .execute::<PrimitiveArray>(&mut ctx)
                .unwrap(),
            PrimitiveArray::from_iter([0u32, 0]),
            &mut ctx
        );
    }

    #[test]
    fn test_canonicalize_null_list() {
        let mut ctx = SESSION.create_execution_ctx();
        let list_scalar = Scalar::null(DType::List(
            Arc::new(DType::Primitive(PType::U64, Nullability::NonNullable)),
            Nullability::Nullable,
        ));
        let const_array = ConstantArray::new(list_scalar, 2).into_array();
        let canonical_const = const_array.execute::<ListViewArray>(&mut ctx).unwrap();
        let elements_prim = canonical_const
            .elements()
            .clone()
            .execute::<PrimitiveArray>(&mut ctx)
            .unwrap();
        assert!(elements_prim.is_empty());
        assert_arrays_eq!(
            canonical_const
                .offsets()
                .clone()
                .execute::<PrimitiveArray>(&mut ctx)
                .unwrap(),
            PrimitiveArray::from_iter([0u32, 0]),
            &mut ctx
        );
        assert_arrays_eq!(
            canonical_const
                .sizes()
                .clone()
                .execute::<PrimitiveArray>(&mut ctx)
                .unwrap(),
            PrimitiveArray::from_iter([0u32, 0]),
            &mut ctx
        );
    }

    #[test]
    fn test_canonicalize_nullable_struct() {
        let mut ctx = SESSION.create_execution_ctx();
        let array = ConstantArray::new(
            Scalar::null(DType::struct_(
                [(
                    "non_null_field",
                    DType::Primitive(PType::I8, Nullability::NonNullable),
                )],
                Nullability::Nullable,
            )),
            3,
        );

        let struct_array = array
            .as_array()
            .clone()
            .execute::<StructArray>(&mut ctx)
            .unwrap();
        assert_eq!(struct_array.len(), 3);
        assert_eq!(struct_array.valid_count(&mut ctx).unwrap(), 0);

        let field = struct_array
            .unmasked_field_by_name("non_null_field")
            .unwrap();

        assert_eq!(
            field.dtype(),
            &DType::Primitive(PType::I8, Nullability::NonNullable)
        );
    }

    #[test]
    fn test_canonicalize_fixed_size_list_non_null() {
        let mut ctx = SESSION.create_execution_ctx();
        // Test with a non-null fixed-size list constant.
        let fsl_scalar = Scalar::fixed_size_list(
            Arc::new(DType::Primitive(PType::I32, Nullability::NonNullable)),
            vec![
                Scalar::primitive(10i32, Nullability::NonNullable),
                Scalar::primitive(20i32, Nullability::NonNullable),
                Scalar::primitive(30i32, Nullability::NonNullable),
            ],
            Nullability::NonNullable,
        );

        let const_array = ConstantArray::new(fsl_scalar, 4).into_array();
        let canonical = const_array.execute::<FixedSizeListArray>(&mut ctx).unwrap();

        assert_eq!(canonical.len(), 4);
        assert_eq!(canonical.list_size(), 3);
        assert!(matches!(canonical.validity(), Ok(Validity::NonNullable)));

        // Check that each list is [10, 20, 30].
        for i in 0..4 {
            let list = canonical.fixed_size_list_elements_at(i).unwrap();
            let list_primitive = list.execute::<PrimitiveArray>(&mut ctx).unwrap();
            assert_arrays_eq!(
                list_primitive,
                PrimitiveArray::from_iter([10i32, 20, 30]),
                &mut ctx
            );
        }
    }

    #[test]
    fn test_canonicalize_fixed_size_list_nullable() {
        let mut ctx = SESSION.create_execution_ctx();
        // Test with a nullable but non-null fixed-size list constant.
        let fsl_scalar = Scalar::fixed_size_list(
            Arc::new(DType::Primitive(PType::F64, Nullability::NonNullable)),
            vec![
                Scalar::primitive(1.5f64, Nullability::NonNullable),
                Scalar::primitive(2.5f64, Nullability::NonNullable),
            ],
            Nullability::Nullable,
        );

        let const_array = ConstantArray::new(fsl_scalar, 3).into_array();
        let canonical = const_array.execute::<FixedSizeListArray>(&mut ctx).unwrap();

        assert_eq!(canonical.len(), 3);
        assert_eq!(canonical.list_size(), 2);
        assert!(matches!(canonical.validity(), Ok(Validity::AllValid)));

        // Check elements.
        let elements = canonical
            .elements()
            .clone()
            .execute::<PrimitiveArray>(&mut ctx)
            .unwrap();
        assert_arrays_eq!(
            elements,
            PrimitiveArray::from_iter([1.5f64, 2.5, 1.5, 2.5, 1.5, 2.5]),
            &mut ctx
        );
    }

    #[test]
    fn test_canonicalize_fixed_size_list_null() {
        let mut ctx = SESSION.create_execution_ctx();
        // Test with a null fixed-size list constant.
        let fsl_scalar = Scalar::null(DType::FixedSizeList(
            Arc::new(DType::Primitive(PType::U64, Nullability::NonNullable)),
            4,
            Nullability::Nullable,
        ));

        let const_array = ConstantArray::new(fsl_scalar, 5).into_array();
        let canonical = const_array.execute::<FixedSizeListArray>(&mut ctx).unwrap();

        assert_eq!(canonical.len(), 5);
        assert_eq!(canonical.list_size(), 4);
        assert!(matches!(canonical.validity(), Ok(Validity::AllInvalid)));

        // Elements should be defaults (zeros).
        let elements = canonical
            .elements()
            .clone()
            .execute::<PrimitiveArray>(&mut ctx)
            .unwrap();
        assert_eq!(elements.len(), 20); // 5 lists * 4 elements each
        assert!(elements.as_slice::<u64>().iter().all(|&x| x == 0));
    }

    #[test]
    fn test_canonicalize_fixed_size_list_empty() {
        let mut ctx = SESSION.create_execution_ctx();
        // Test with size-0 lists (edge case).
        let fsl_scalar = Scalar::fixed_size_list(
            Arc::new(DType::Primitive(PType::I8, Nullability::NonNullable)),
            vec![],
            Nullability::NonNullable,
        );

        let const_array = ConstantArray::new(fsl_scalar, 10).into_array();
        let canonical = const_array.execute::<FixedSizeListArray>(&mut ctx).unwrap();

        assert_eq!(canonical.len(), 10);
        assert_eq!(canonical.list_size(), 0);
        assert!(matches!(canonical.validity(), Ok(Validity::NonNullable)));

        // Elements array should be empty.
        assert!(canonical.elements().is_empty());
    }

    #[test]
    fn test_canonicalize_fixed_size_list_nested() {
        let mut ctx = SESSION.create_execution_ctx();
        // Test with nested data types (list of strings).
        let fsl_scalar = Scalar::fixed_size_list(
            Arc::new(DType::Utf8(Nullability::NonNullable)),
            vec![Scalar::from("hello"), Scalar::from("world")],
            Nullability::NonNullable,
        );

        let const_array = ConstantArray::new(fsl_scalar, 2).into_array();
        let canonical = const_array.execute::<FixedSizeListArray>(&mut ctx).unwrap();

        assert_eq!(canonical.len(), 2);
        assert_eq!(canonical.list_size(), 2);

        // Check elements are repeated correctly.
        let elements = canonical
            .elements()
            .clone()
            .execute::<VarBinViewArray>(&mut ctx)
            .unwrap();
        assert_eq!(
            elements.execute_scalar(0, &mut ctx).unwrap(),
            "hello".into()
        );
        assert_eq!(
            elements.execute_scalar(1, &mut ctx).unwrap(),
            "world".into()
        );
        assert_eq!(
            elements.execute_scalar(2, &mut ctx).unwrap(),
            "hello".into()
        );
        assert_eq!(
            elements.execute_scalar(3, &mut ctx).unwrap(),
            "world".into()
        );
    }

    #[test]
    fn test_canonicalize_fixed_size_list_single_element() {
        let mut ctx = SESSION.create_execution_ctx();
        // Test with a single-element list.
        let fsl_scalar = Scalar::fixed_size_list(
            Arc::new(DType::Primitive(PType::I16, Nullability::NonNullable)),
            vec![Scalar::primitive(42i16, Nullability::NonNullable)],
            Nullability::NonNullable,
        );

        let const_array = ConstantArray::new(fsl_scalar, 1).into_array();
        let canonical = const_array.execute::<FixedSizeListArray>(&mut ctx).unwrap();

        assert_eq!(canonical.len(), 1);
        assert_eq!(canonical.list_size(), 1);

        let elements = canonical
            .elements()
            .clone()
            .execute::<PrimitiveArray>(&mut ctx)
            .unwrap();
        assert_arrays_eq!(elements, PrimitiveArray::from_iter([42i16]), &mut ctx);
    }

    #[test]
    fn test_canonicalize_fixed_size_list_with_null_elements() {
        let mut ctx = SESSION.create_execution_ctx();
        // Test FSL with nullable element type where some elements are null.
        let fsl_scalar = Scalar::fixed_size_list(
            Arc::new(DType::Primitive(PType::I32, Nullability::Nullable)),
            vec![
                Scalar::primitive(100i32, Nullability::Nullable),
                Scalar::null(DType::Primitive(PType::I32, Nullability::Nullable)),
                Scalar::primitive(200i32, Nullability::Nullable),
            ],
            Nullability::NonNullable,
        );

        let const_array = ConstantArray::new(fsl_scalar, 3).into_array();
        let canonical = const_array.execute::<FixedSizeListArray>(&mut ctx).unwrap();

        assert_eq!(canonical.len(), 3);
        assert_eq!(canonical.list_size(), 3);
        assert!(matches!(canonical.validity(), Ok(Validity::NonNullable)));

        // Check elements including nulls.
        let elements = canonical
            .elements()
            .clone()
            .execute::<PrimitiveArray>(&mut ctx)
            .unwrap();
        assert_eq!(
            elements.execute_scalar(0, &mut ctx).unwrap(),
            Scalar::from(100i32)
        );
        assert_eq!(
            elements.execute_scalar(1, &mut ctx).unwrap(),
            Scalar::null(DType::Primitive(PType::I32, Nullability::Nullable))
        );
        assert_eq!(
            elements.execute_scalar(2, &mut ctx).unwrap(),
            Scalar::from(200i32)
        );

        // Check element validity.
        let element_validity = elements
            .validity()
            .vortex_expect("constant canonical element validity should be derivable");
        assert!(element_validity.execute_is_valid(0, &mut ctx).unwrap());
        assert!(!element_validity.execute_is_valid(1, &mut ctx).unwrap());
        assert!(element_validity.execute_is_valid(2, &mut ctx).unwrap());

        // Pattern should repeat.
        assert!(element_validity.execute_is_valid(3, &mut ctx).unwrap());
        assert!(!element_validity.execute_is_valid(4, &mut ctx).unwrap());
        assert!(element_validity.execute_is_valid(5, &mut ctx).unwrap());
    }

    #[test]
    fn test_canonicalize_fixed_size_list_large() {
        let mut ctx = SESSION.create_execution_ctx();
        // Test with a large constant array.
        let fsl_scalar = Scalar::fixed_size_list(
            Arc::new(DType::Primitive(PType::U8, Nullability::NonNullable)),
            vec![
                Scalar::primitive(1u8, Nullability::NonNullable),
                Scalar::primitive(2u8, Nullability::NonNullable),
                Scalar::primitive(3u8, Nullability::NonNullable),
                Scalar::primitive(4u8, Nullability::NonNullable),
                Scalar::primitive(5u8, Nullability::NonNullable),
            ],
            Nullability::NonNullable,
        );

        let const_array = ConstantArray::new(fsl_scalar, 1000).into_array();
        let canonical = const_array.execute::<FixedSizeListArray>(&mut ctx).unwrap();

        assert_eq!(canonical.len(), 1000);
        assert_eq!(canonical.list_size(), 5);

        let elements = canonical
            .elements()
            .clone()
            .execute::<PrimitiveArray>(&mut ctx)
            .unwrap();
        assert_eq!(elements.len(), 5000);

        // Check pattern repeats correctly.
        for i in 0..1000 {
            let base = i * 5;
            assert_eq!(
                elements.execute_scalar(base, &mut ctx).unwrap(),
                Scalar::from(1u8)
            );
            assert_eq!(
                elements.execute_scalar(base + 1, &mut ctx).unwrap(),
                Scalar::from(2u8)
            );
            assert_eq!(
                elements.execute_scalar(base + 2, &mut ctx).unwrap(),
                Scalar::from(3u8)
            );
            assert_eq!(
                elements.execute_scalar(base + 3, &mut ctx).unwrap(),
                Scalar::from(4u8)
            );
            assert_eq!(
                elements.execute_scalar(base + 4, &mut ctx).unwrap(),
                Scalar::from(5u8)
            );
        }
    }

    /// Uniform elements should stay one constant array, not materialize per row.
    #[test]
    fn test_canonicalize_fixed_size_list_uniform_elements_stay_constant() {
        let mut ctx = SESSION.create_execution_ctx();
        let fsl_scalar = Scalar::fixed_size_list(
            Arc::new(DType::Primitive(PType::I32, Nullability::NonNullable)),
            vec![Scalar::primitive(7i32, Nullability::NonNullable); 3],
            Nullability::NonNullable,
        );

        let canonical = ConstantArray::new(fsl_scalar, 10_000)
            .into_array()
            .execute::<FixedSizeListArray>(&mut ctx)
            .unwrap();

        assert_eq!(canonical.len(), 10_000);
        assert_eq!(canonical.elements().len(), 30_000);
        assert!(
            canonical.elements().as_opt::<Constant>().is_some(),
            "uniform elements should stay constant-encoded rather than materialize per row",
        );
        for index in [0, 5_000, 9_999] {
            assert_arrays_eq!(
                canonical.fixed_size_list_elements_at(index).unwrap(),
                PrimitiveArray::from_iter([7i32, 7, 7]),
                &mut ctx
            );
        }
    }

    /// Mixed elements should materialize once, with every row chunking that one copy.
    #[test]
    fn test_canonicalize_fixed_size_list_tiles_one_copy_of_mixed_elements() {
        let mut ctx = SESSION.create_execution_ctx();
        const LEN: usize = 1_000;
        let fsl_scalar = Scalar::fixed_size_list(
            Arc::new(DType::Primitive(PType::I32, Nullability::NonNullable)),
            vec![
                Scalar::primitive(1i32, Nullability::NonNullable),
                Scalar::primitive(2i32, Nullability::NonNullable),
            ],
            Nullability::NonNullable,
        );

        let canonical = ConstantArray::new(fsl_scalar, LEN)
            .into_array()
            .execute::<FixedSizeListArray>(&mut ctx)
            .unwrap();

        assert_eq!(canonical.elements().len(), 2 * LEN);
        assert_eq!(
            canonical.elements().as_::<Chunked>().nchunks(),
            LEN,
            "every row should reference the same tile rather than hold its own copy",
        );
        for index in [0, 1, LEN - 1] {
            assert_arrays_eq!(
                canonical.fixed_size_list_elements_at(index).unwrap(),
                PrimitiveArray::from_iter([1i32, 2]),
                &mut ctx
            );
        }
    }

    /// Only a value too long to inline should cost a data buffer.
    #[rstest]
    #[case::inlined("exactly12chr", 0)]
    #[case::referenced("thirteen chrs", 1)]
    fn test_canonicalize_string_stores_the_value_once(
        #[case] value: &str,
        #[case] data_buffers: usize,
    ) {
        let mut ctx = SESSION.create_execution_ctx();
        let canonical = ConstantArray::new(value.to_string(), 100)
            .into_array()
            .execute::<VarBinViewArray>(&mut ctx)
            .unwrap();

        assert_eq!(canonical.data_buffers().len(), data_buffers);
        assert_arrays_eq!(
            canonical,
            VarBinViewArray::from_iter_str(std::iter::repeat_n(value, 100)),
            &mut ctx
        );
    }
}