parquet_variant_compute/

variant_get.rs

// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.
use arrow::{
    array::{
        self, Array, ArrayRef, GenericListArray, GenericListViewArray, ListLikeArray, StructArray,
        UInt64Array, make_array,
    },
    buffer::NullBuffer,
    compute::{CastOptions, take},
    datatypes::Field,
    error::Result,
};
use arrow_schema::{ArrowError, DataType, FieldRef};
use parquet_variant::{VariantPath, VariantPathElement};

use crate::ShreddingState;
use crate::variant_to_arrow::make_variant_to_arrow_row_builder;
use crate::{VariantArray, VariantType, unshred_variant};

use arrow::array::AsArray;
use std::sync::Arc;

pub(crate) enum ShreddedPathStep {
    /// Path step succeeded, return the new shredding state
    Success(ShreddingState),
    /// The path element is not present in the `typed_value` column and there is no `value` column,
    /// so we know it does not exist. It, and all paths under it, are all-NULL.
    Missing,
    /// The path element is not present in the `typed_value` column and must be retrieved from the `value`
    /// column instead. The caller should be prepared to handle any value, including the requested
    /// type, an arbitrary "wrong" type, or `Variant::Null`.
    NotShredded,
}

/// Build the next shredding state by taking one list-like element (at `index`) per input row.
///
fn take_list_like_index_as_shredding_state<L: ListLikeArray + 'static>(
    typed_value: &dyn Array,
    index: usize,
) -> Result<Option<ShreddingState>> {
    let list_array = typed_value.as_any().downcast_ref::<L>().ok_or_else(|| {
        ArrowError::ComputeError(format!(
            "Expected array type '{}' while handling list-like path step, got '{}'",
            std::any::type_name::<L>(),
            typed_value.data_type()
        ))
    })?;

    let values = list_array.values();

    let Some(struct_array) = values.as_struct_opt() else {
        return Ok(None);
    };
    let shredding_state = ShreddingState::try_from(struct_array)?;

    let value_array = shredding_state.value_column();
    let typed_array = shredding_state.typed_value_column();

    // If list elements have neither typed nor fallback value, this path step is missing.
    if value_array.is_none() && typed_array.is_none() {
        return Ok(None);
    }

    let mut take_indices = Vec::with_capacity(list_array.len());
    for row in 0..list_array.len() {
        let row_range = list_array.element_range(row);
        let take_index = (index < row_range.len()).then(|| (row_range.start + index) as u64);
        take_indices.push(take_index);
    }

    let index_array = UInt64Array::from(take_indices);

    // Gather both typed and fallback values at the requested element index.
    let taken_value = value_array
        .map(|value| take(value, &index_array, None))
        .transpose()?;
    let taken_typed = typed_array
        .map(|typed| take(typed, &index_array, None))
        .transpose()?;

    Ok(Some(ShreddingState::new(taken_value, taken_typed)))
}

/// Given a shredded variant field -- a `(value?, typed_value?)` pair -- try to take one path step
/// deeper. For a `VariantPathElement::Field`, if there is no `typed_value` at this level, if
/// `typed_value` is not a struct, or if the requested field name does not exist, traversal returns
/// a missing-path step (`Missing` or `NotShredded` depending on whether `value` exists).
///
/// Safe-cast behavior (`cast_options.safe = true`):
/// - Type mismatch during path traversal (for example field access on non-struct, index access on
///   non-list) returns [`ShreddedPathStep::Missing`] or [`ShreddedPathStep::NotShredded`], allowing
///   the caller to continue with null/fallback semantics.
/// - List index out-of-bounds produces nulls for the corresponding rows.
///
/// Unsafe-cast behavior (`cast_options.safe = false`):
/// - Field access on non-struct returns [`ArrowError::CastError`].
/// - List index path steps follow JSONPath semantics and return missing/null for non-list or
///   out-of-bounds rows.
pub(crate) fn follow_shredded_path_element(
    shredding_state: &ShreddingState,
    path_element: &VariantPathElement<'_>,
    _cast_options: &CastOptions,
) -> Result<ShreddedPathStep> {
    // If the requested path element is not present in `typed_value`, and `value` is missing, then
    // we know it does not exist; it, and all paths under it, are all-NULL.
    let missing_path_step = || match shredding_state.value_column() {
        Some(_) => ShreddedPathStep::NotShredded,
        None => ShreddedPathStep::Missing,
    };

    let Some(typed_value) = shredding_state.typed_value_column() else {
        return Ok(missing_path_step());
    };

    match path_element {
        VariantPathElement::Field { name } => {
            // Try to step into the requested field name of a struct.
            // First, try to downcast to StructArray
            let Some(struct_array) = typed_value.as_struct_opt() else {
                // Object field path step follows JSONPath semantics and returns missing path step (NotShredded/Missing) on non-struct path
                return Ok(missing_path_step());
            };

            // Now try to find the column - missing column in a present struct is just missing data
            let Some(field) = struct_array.column_by_name(name) else {
                // Missing column in a present struct is just missing, not wrong - return Ok
                return Ok(missing_path_step());
            };

            let struct_array = field.as_struct_opt().ok_or_else(|| {
                // TODO: Should we blow up? Or just end the traversal and let the normal
                // variant pathing code sort out the mess that it must anyway be
                // prepared to handle?
                ArrowError::InvalidArgumentError(format!(
                    "Expected Struct array while following path, got {}",
                    field.data_type(),
                ))
            })?;

            let state = ShreddingState::try_from(struct_array)?;
            Ok(ShreddedPathStep::Success(state))
        }
        VariantPathElement::Index { index } => {
            let state = match typed_value.data_type() {
                DataType::List(_) => take_list_like_index_as_shredding_state::<
                    GenericListArray<i32>,
                >(typed_value.as_ref(), *index)?,
                DataType::LargeList(_) => take_list_like_index_as_shredding_state::<
                    GenericListArray<i64>,
                >(typed_value.as_ref(), *index)?,
                DataType::ListView(_) => take_list_like_index_as_shredding_state::<
                    GenericListViewArray<i32>,
                >(typed_value.as_ref(), *index)?,
                DataType::LargeListView(_) => take_list_like_index_as_shredding_state::<
                    GenericListViewArray<i64>,
                >(typed_value.as_ref(), *index)?,
                _ => {
                    // JSONPath semantics: indexing a non-list yields no match.
                    return Ok(missing_path_step());
                }
            };

            match state {
                Some(state) => Ok(ShreddedPathStep::Success(state)),
                None => Ok(missing_path_step()),
            }
        }
    }
}

/// Follows the given path as far as possible through shredded variant fields. If the path ends on a
/// shredded field, return it directly. Otherwise, use a row shredder to follow the rest of the path
/// and extract the requested value on a per-row basis.
fn shredded_get_path(
    input: &VariantArray,
    path: &[VariantPathElement<'_>],
    as_field: Option<&Field>,
    cast_options: &CastOptions,
) -> Result<ArrayRef> {
    // Helper that creates a new VariantArray from the given nested value and typed_value columns,
    // properly accounting for accumulated nulls from path traversal
    let make_target_variant =
        |value: Option<ArrayRef>,
         typed_value: Option<ArrayRef>,
         accumulated_nulls: Option<NullBuffer>| {
            let metadata = input.metadata_column().clone();
            VariantArray::from_parts(metadata, value, typed_value, accumulated_nulls)
        };

    // Helper that extracts the value at `path` and casts it to the requested type, or returns it as
    // an unshredded binary variant when `Variant` output is requested.
    let shred_basic_variant =
        |target: VariantArray, path: VariantPath<'_>, as_field: Option<&Field>| {
            // A `VariantType` extension on `as_field` requests `Variant` output: return an
            // unshredded binary variant instead of casting to a concrete Arrow type.
            let requested_variant =
                as_field.is_some_and(Field::has_valid_extension_type::<VariantType>);

            // A `typed_value` in that field requests shredded output -- a `VariantArray` with
            // `typed_value` columns. We produce only unshredded variant output. Shredded output is
            // tracked in https://github.com/apache/arrow-rs/issues/8153. Reject such a request
            // instead of silently dropping the shredding it asked for.
            if requested_variant && requested_field_is_shredded(as_field) {
                return Err(ArrowError::NotYetImplemented(
                    "variant_get with shredded `Variant` output is not yet supported".to_string(),
                ));
            }

            // Collapse any shredding back to binary. Only the `NotShredded` step below passes a
            // non-empty `path`, and there `target` is already a plain `value` column (no
            // `typed_value`) -- so `unshred_variant` hits its clone fast-path, with nothing deeper
            // to shred. The builder then walks any remaining path per-row, emitting variant output
            // because `as_type` is `None`.
            let target = if requested_variant {
                unshred_variant(&target)?
            } else {
                target
            };

            // Path exhausted, variant requested: return the target directly.
            if requested_variant && path.is_empty() {
                return Ok(ArrayRef::from(target));
            }

            let as_type = if requested_variant {
                None
            } else {
                as_field.map(|f| f.data_type())
            };
            let mut builder = make_variant_to_arrow_row_builder(
                target.metadata_column(),
                path,
                as_type,
                cast_options,
                target.len(),
            )?;
            for i in 0..target.len() {
                if target.is_null(i) {
                    builder.append_null()?;
                } else if !cast_options.safe {
                    let value = target.try_value(i)?;
                    builder.append_value(value)?;
                } else {
                    let _ = match target.try_value(i) {
                        Ok(v) => builder.append_value(v)?,
                        Err(_) => {
                            builder.append_null()?;
                            false // add this to make match arms have the same return type
                        }
                    };
                }
            }
            builder.finish()
        };

    // Peel away the prefix of path elements that traverses the shredded parts of this variant
    // column. Shredding will traverse the rest of the path on a per-row basis.
    let mut shredding_state = input.shredding_state().clone();
    let mut accumulated_nulls = input.inner().nulls().cloned();
    let mut path_index = 0;
    for path_element in path {
        match follow_shredded_path_element(&shredding_state, path_element, cast_options)? {
            ShreddedPathStep::Success(state) => {
                // Union nulls from the typed_value we just accessed
                if let Some(typed_value) = shredding_state.typed_value_column() {
                    accumulated_nulls =
                        NullBuffer::union(accumulated_nulls.as_ref(), typed_value.nulls());
                }
                shredding_state = state;
                path_index += 1;
                continue;
            }
            ShreddedPathStep::Missing => {
                let num_rows = input.len();
                if as_field.is_some_and(Field::has_valid_extension_type::<VariantType>) {
                    let all_nulls = Some(arrow::buffer::NullBuffer::from(vec![false; num_rows]));
                    // Propagating metadata is not necessary for an all-NULL array, but is cheaper than constructing
                    // a new empty metadata array. (n * 3 bytes vs Arc bump)
                    let metadata = input.metadata_column().clone();
                    let arr = VariantArray::from_parts(metadata, None, None, all_nulls);
                    return Ok(ArrayRef::from(arr));
                }
                let arr = match as_field.map(|f| f.data_type()) {
                    Some(data_type) => array::new_null_array(data_type, num_rows),
                    None => Arc::new(array::NullArray::new(num_rows)) as _,
                };
                return Ok(arr);
            }
            ShreddedPathStep::NotShredded => {
                let target = make_target_variant(
                    shredding_state.value_column().cloned(),
                    None,
                    accumulated_nulls,
                );
                return shred_basic_variant(target, path[path_index..].into(), as_field);
            }
        };
    }

    // Path exhausted! Create a new `VariantArray` for the location we landed on.
    let target = make_target_variant(
        shredding_state.value_column().cloned(),
        shredding_state.typed_value_column().cloned(),
        accumulated_nulls,
    );

    // If our caller did not request any specific type, we can just return whatever we landed on.
    let Some(as_field) = as_field else {
        return Ok(ArrayRef::from(target));
    };

    // Try to return the typed value directly when we have a perfect shredding match.
    if let Some(shredded) = try_perfect_shredding(&target, as_field) {
        return Ok(shredded);
    }

    // Structs are special.
    //
    // For fully unshredded targets (`typed_value` absent), delegate to the row builder so we
    // preserve struct-level cast semantics:
    // - safe mode: non-object rows become NULL structs
    // - strict mode: non-object rows raise a cast error
    //
    // For shredded/partially-shredded targets (`typed_value` present), recurse into each field
    // separately to take advantage of deeper shredding in child fields.
    if !as_field.has_valid_extension_type::<VariantType>() {
        if let DataType::Struct(fields) = as_field.data_type() {
            if target.typed_value_column().is_none() {
                return shred_basic_variant(target, VariantPath::default(), Some(as_field));
            }

            let children = fields
                .iter()
                .map(|field| {
                    let path = &[VariantPathElement::from(field.name().as_str())];
                    shredded_get_path(&target, path, Some(field), cast_options)
                })
                .collect::<Result<Vec<_>>>()?;

            return Ok(Arc::new(StructArray::try_new(
                fields.clone(),
                children,
                target.nulls().cloned(),
            )?));
        }
    }

    // Not a struct, so directly shred the variant as the requested type
    shred_basic_variant(target, VariantPath::default(), Some(as_field))
}

/// Returns true if `as_field` requests *shredded* `Variant` output.
///
/// Its struct carries a `typed_value` field naming the type to shred to.
/// A plain variant request has only `metadata` and `value`.
fn requested_field_is_shredded(as_field: Option<&Field>) -> bool {
    as_field.is_some_and(|f| match f.data_type() {
        DataType::Struct(fields) => fields.iter().any(|field| field.name() == "typed_value"),
        _ => false,
    })
}

fn try_perfect_shredding(variant_array: &VariantArray, as_field: &Field) -> Option<ArrayRef> {
    // Try to return the typed value directly when we have a perfect shredding match.
    if matches!(as_field.data_type(), DataType::Struct(_)) {
        return None;
    }
    let typed_value = variant_array.typed_value_column()?;

    if typed_value.data_type() == as_field.data_type()
        && variant_array
            .value_column()
            .is_none_or(|v| v.null_count() == v.len())
    {
        // Here we need to gate against the case where the `typed_value` is null but data is in the `value` column.
        // 1. If the `value` column is null, or
        // 2. If every row in the `value` column is null

        // This is a perfect shredding, where the value is entirely shredded out,
        // so we can just return the typed value after merging the accumulated nulls.
        let parent_nulls = variant_array.nulls();

        // If we have no nulls OR the shredded array is `Null`, which doesn't support external nulls.
        let target_array = if parent_nulls.is_none() || typed_value.data_type().is_null() {
            typed_value.clone()
        } else {
            let merged_nulls = NullBuffer::union(parent_nulls, typed_value.nulls());
            let data = typed_value
                .to_data()
                .into_builder()
                .nulls(merged_nulls)
                .build()
                .ok()?;
            make_array(data)
        };

        return Some(target_array);
    }

    None
}

/// Returns an array with the specified path extracted from the variant values.
///
/// The return array type depends on the `as_type` field of the options parameter
/// 1. `as_type: None`: a VariantArray is returned. The values in this new VariantArray will point
///    to the specified path.
/// 2. `as_type: Some(<specific field>)`: an array of the specified type is returned.
///
/// TODO: How would a caller request a struct or list type where the fields/elements can be any
/// variant? Caller can pass None as the requested type to fetch a specific path, but it would
/// quickly become annoying (and inefficient) to call `variant_get` for each leaf value in a struct or
/// list and then try to assemble the results.
pub fn variant_get(input: &ArrayRef, options: GetOptions) -> Result<ArrayRef> {
    let variant_array = VariantArray::try_new(input)?;

    let GetOptions {
        as_type,
        path,
        cast_options,
    } = options;

    shredded_get_path(&variant_array, &path, as_type.as_deref(), &cast_options)
}

/// Controls the action of the variant_get kernel.
#[derive(Debug, Clone, Default)]
pub struct GetOptions<'a> {
    /// What path to extract
    pub path: VariantPath<'a>,
    /// if `as_type` is None, the returned array will itself be a VariantArray.
    ///
    /// if `as_type` is `Some(type)` the field is returned as the specified type.
    pub as_type: Option<FieldRef>,
    /// Controls the casting behavior (e.g. error vs substituting null on cast error).
    pub cast_options: CastOptions<'a>,
}

impl<'a> GetOptions<'a> {
    /// Construct default options to get the specified path as a variant.
    pub fn new() -> Self {
        Default::default()
    }

    /// Construct options to get the specified path as a variant.
    pub fn new_with_path(path: VariantPath<'a>) -> Self {
        Self {
            path,
            as_type: None,
            cast_options: Default::default(),
        }
    }

    /// Specify the type to return.
    pub fn with_as_type(mut self, as_type: Option<FieldRef>) -> Self {
        self.as_type = as_type;
        self
    }

    /// Specify the cast options to use when casting to the specified type.
    pub fn with_cast_options(mut self, cast_options: CastOptions<'a>) -> Self {
        self.cast_options = cast_options;
        self
    }
}

#[cfg(test)]
mod test {
    use std::str::FromStr;
    use std::sync::Arc;

    use super::{GetOptions, requested_field_is_shredded, variant_get};
    use crate::variant_array::{ShreddedVariantFieldArray, StructArrayBuilder};
    use crate::{
        ShreddedSchemaBuilder, VariantArray, VariantArrayBuilder, cast_to_variant, json_to_variant,
        shred_variant,
    };
    use arrow::array::{
        Array, ArrayRef, AsArray, BinaryArray, BinaryViewArray, BooleanArray, Date32Array,
        Date64Array, Decimal32Array, Decimal64Array, Decimal128Array, Decimal256Array,
        FixedSizeListArray, Float32Array, Float64Array, Int8Array, Int16Array, Int32Array,
        Int64Array, LargeBinaryArray, LargeListArray, LargeListViewArray, LargeStringArray,
        ListArray, ListViewArray, NullArray, NullBuilder, StringArray, StringViewArray,
        StructArray, Time32MillisecondArray, Time32SecondArray, Time64MicrosecondArray,
        Time64NanosecondArray,
    };
    use arrow::buffer::{NullBuffer, OffsetBuffer, ScalarBuffer};
    use arrow::compute::{CastOptions, cast};
    use arrow::datatypes::DataType::{Int16, Int32, Int64};
    use arrow::datatypes::i256;
    use arrow::util::display::FormatOptions;
    use arrow_schema::ArrowError;
    use arrow_schema::DataType::{Boolean, Float32, Float64, Int8};
    use arrow_schema::{DataType, Field, FieldRef, Fields, IntervalUnit, TimeUnit};
    use chrono::DateTime;
    use parquet_variant::{
        EMPTY_VARIANT_METADATA_BYTES, Variant, VariantDecimal4, VariantDecimal8, VariantDecimal16,
        VariantDecimalType, VariantPath,
    };

    fn single_variant_get_test(input_json: &str, path: VariantPath, expected_json: &str) {
        // Create input array from JSON string
        let input_array_ref: ArrayRef = Arc::new(StringArray::from(vec![Some(input_json)]));
        let input_variant_array_ref = ArrayRef::from(json_to_variant(&input_array_ref).unwrap());

        let result =
            variant_get(&input_variant_array_ref, GetOptions::new_with_path(path)).unwrap();

        // Create expected array from JSON string
        let expected_array_ref: ArrayRef = Arc::new(StringArray::from(vec![Some(expected_json)]));
        let expected_variant_array = json_to_variant(&expected_array_ref).unwrap();

        let result_array = VariantArray::try_new(&result).unwrap();
        assert_eq!(
            result_array.len(),
            1,
            "Expected result array to have length 1"
        );
        assert!(
            result_array.nulls().is_none(),
            "Expected no nulls in result array"
        );
        let result_variant = result_array.value(0);
        let expected_variant = expected_variant_array.value(0);
        assert_eq!(
            result_variant, expected_variant,
            "Result variant does not match expected variant"
        );
    }

    #[test]
    fn get_primitive_variant_field() {
        single_variant_get_test(
            r#"{"some_field": 1234}"#,
            VariantPath::try_from("some_field").unwrap(),
            "1234",
        );
    }

    #[test]
    fn get_primitive_variant_list_index() {
        single_variant_get_test("[1234, 5678]", VariantPath::from(0), "1234");
    }

    #[test]
    fn get_primitive_variant_inside_object_of_object() {
        single_variant_get_test(
            r#"{"top_level_field": {"inner_field": 1234}}"#,
            VariantPath::try_from("top_level_field")
                .unwrap()
                .join("inner_field"),
            "1234",
        );
    }

    #[test]
    fn get_primitive_variant_inside_list_of_object() {
        single_variant_get_test(
            r#"[{"some_field": 1234}]"#,
            VariantPath::from(0).join("some_field"),
            "1234",
        );
    }

    #[test]
    fn get_primitive_variant_inside_object_of_list() {
        single_variant_get_test(
            r#"{"some_field": [1234]}"#,
            VariantPath::try_from("some_field[0]").unwrap(),
            "1234",
        );
    }

    #[test]
    fn get_complex_variant() {
        single_variant_get_test(
            r#"{"top_level_field": {"inner_field": 1234}}"#,
            VariantPath::try_from("top_level_field").unwrap(),
            r#"{"inner_field": 1234}"#,
        );
    }

    /// Partial Shredding: extract a value as a VariantArray
    macro_rules! numeric_partially_shredded_test {
        ($primitive_type:ty, $data_fn:ident) => {
            let array = $data_fn();
            let options = GetOptions::new();
            let result = variant_get(&array, options).unwrap();

            // expect the result is a VariantArray
            let result = VariantArray::try_new(&result).unwrap();
            assert_eq!(result.len(), 4);

            // Expect the values are the same as the original values
            assert_eq!(
                result.value(0),
                Variant::from(<$primitive_type>::try_from(34u8).unwrap())
            );
            assert!(!result.is_valid(1));
            assert_eq!(result.value(2), Variant::from("n/a"));
            assert_eq!(
                result.value(3),
                Variant::from(<$primitive_type>::try_from(100u8).unwrap())
            );
        };
    }

    /// Build a mixed input [typed, null, fallback, typed] and let shred_variant
    /// generate the shredded fixture for the requested type.
    macro_rules! partially_shredded_variant_array_gen {
        ($func_name:ident,  $typed_value_array_gen: expr) => {
            partially_shredded_variant_array_gen!(
                $func_name,
                $typed_value_array_gen,
                Variant::from("n/a")
            );
        };
        ($func_name:ident,  $typed_value_array_gen: expr, $fallback_variant:expr) => {
            fn $func_name() -> ArrayRef {
                let typed_value: ArrayRef = Arc::new($typed_value_array_gen());
                let typed_as_variant = cast_to_variant(typed_value.as_ref())
                    .expect("should cast typed array to variant");
                let mut input_builder = VariantArrayBuilder::new(typed_as_variant.len());
                input_builder.append_variant(typed_as_variant.value(0));
                input_builder.append_null();
                input_builder.append_variant($fallback_variant);
                input_builder.append_variant(typed_as_variant.value(3));

                let variant_array = shred_variant(&input_builder.build(), typed_value.data_type())
                    .expect("should shred variant array");
                ArrayRef::from(variant_array)
            }
        };
    }

    // Fixture definitions grouped with the partially-shredded tests.
    macro_rules! numeric_partially_shredded_variant_array_fn {
        ($func:ident, $array_type:ident, $primitive_type:ty) => {
            partially_shredded_variant_array_gen!($func, || $array_type::from(vec![
                Some(<$primitive_type>::try_from(34u8).unwrap()),
                None,
                None,
                Some(<$primitive_type>::try_from(100u8).unwrap()),
            ]));
        };
    }

    numeric_partially_shredded_variant_array_fn!(
        partially_shredded_int8_variant_array,
        Int8Array,
        i8
    );
    numeric_partially_shredded_variant_array_fn!(
        partially_shredded_int16_variant_array,
        Int16Array,
        i16
    );
    numeric_partially_shredded_variant_array_fn!(
        partially_shredded_int32_variant_array,
        Int32Array,
        i32
    );
    numeric_partially_shredded_variant_array_fn!(
        partially_shredded_int64_variant_array,
        Int64Array,
        i64
    );
    numeric_partially_shredded_variant_array_fn!(
        partially_shredded_float32_variant_array,
        Float32Array,
        f32
    );
    numeric_partially_shredded_variant_array_fn!(
        partially_shredded_float64_variant_array,
        Float64Array,
        f64
    );

    partially_shredded_variant_array_gen!(partially_shredded_bool_variant_array, || {
        arrow::array::BooleanArray::from(vec![Some(true), None, None, Some(false)])
    });

    partially_shredded_variant_array_gen!(
        partially_shredded_utf8_variant_array,
        || { StringArray::from(vec![Some("hello"), None, None, Some("world")]) },
        Variant::from(42i32)
    );

    partially_shredded_variant_array_gen!(partially_shredded_date32_variant_array, || {
        Date32Array::from(vec![
            Some(20348), // 2025-09-17
            None,
            None,
            Some(20340), // 2025-09-09
        ])
    });

    #[test]
    fn get_variant_partially_shredded_int8_as_variant() {
        numeric_partially_shredded_test!(i8, partially_shredded_int8_variant_array);
    }

    #[test]
    fn get_variant_partially_shredded_int16_as_variant() {
        numeric_partially_shredded_test!(i16, partially_shredded_int16_variant_array);
    }

    #[test]
    fn get_variant_partially_shredded_int32_as_variant() {
        numeric_partially_shredded_test!(i32, partially_shredded_int32_variant_array);
    }

    #[test]
    fn get_variant_partially_shredded_int64_as_variant() {
        numeric_partially_shredded_test!(i64, partially_shredded_int64_variant_array);
    }

    #[test]
    fn get_variant_partially_shredded_float32_as_variant() {
        numeric_partially_shredded_test!(f32, partially_shredded_float32_variant_array);
    }

    #[test]
    fn get_variant_partially_shredded_float64_as_variant() {
        numeric_partially_shredded_test!(f64, partially_shredded_float64_variant_array);
    }

    #[test]
    fn get_variant_partially_shredded_bool_as_variant() {
        let array = partially_shredded_bool_variant_array();
        let options = GetOptions::new();
        let result = variant_get(&array, options).unwrap();

        // expect the result is a VariantArray
        let result = VariantArray::try_new(&result).unwrap();
        assert_eq!(result.len(), 4);

        // Expect the values are the same as the original values
        assert_eq!(result.value(0), Variant::from(true));
        assert!(!result.is_valid(1));
        assert_eq!(result.value(2), Variant::from("n/a"));
        assert_eq!(result.value(3), Variant::from(false));
    }

    #[test]
    fn get_variant_partially_shredded_utf8_as_variant() {
        let array = partially_shredded_utf8_variant_array();
        let options = GetOptions::new();
        let result = variant_get(&array, options).unwrap();

        // expect the result is a VariantArray
        let result = VariantArray::try_new(&result).unwrap();
        assert_eq!(result.len(), 4);

        // Expect the values are the same as the original values
        assert_eq!(result.value(0), Variant::from("hello"));
        assert!(!result.is_valid(1));
        assert_eq!(result.value(2), Variant::from(42i32));
        assert_eq!(result.value(3), Variant::from("world"));
    }

    partially_shredded_variant_array_gen!(partially_shredded_binary_view_variant_array, || {
        BinaryViewArray::from(vec![
            Some(&[1u8, 2u8, 3u8][..]), // row 0 is shredded
            None,                       // row 1 is null
            None,                       // row 2 is a string
            Some(&[4u8, 5u8, 6u8][..]), // row 3 is shredded
        ])
    });

    #[test]
    fn get_variant_partially_shredded_date32_as_variant() {
        let array = partially_shredded_date32_variant_array();
        let options = GetOptions::new();
        let result = variant_get(&array, options).unwrap();

        // expect the result is a VariantArray
        let result = VariantArray::try_new(&result).unwrap();
        assert_eq!(result.len(), 4);

        // Expect the values are the same as the original values
        use chrono::NaiveDate;
        let date1 = NaiveDate::from_ymd_opt(2025, 9, 17).unwrap();
        let date2 = NaiveDate::from_ymd_opt(2025, 9, 9).unwrap();
        assert_eq!(result.value(0), Variant::from(date1));
        assert!(!result.is_valid(1));
        assert_eq!(result.value(2), Variant::from("n/a"));
        assert_eq!(result.value(3), Variant::from(date2));
    }

    #[test]
    fn get_variant_partially_shredded_binary_view_as_variant() {
        let array = partially_shredded_binary_view_variant_array();
        let options = GetOptions::new();
        let result = variant_get(&array, options).unwrap();

        // expect the result is a VariantArray
        let result = VariantArray::try_new(&result).unwrap();
        assert_eq!(result.len(), 4);

        // Expect the values are the same as the original values
        assert_eq!(result.value(0), Variant::from(&[1u8, 2u8, 3u8][..]));
        assert!(!result.is_valid(1));
        assert_eq!(result.value(2), Variant::from("n/a"));
        assert_eq!(result.value(3), Variant::from(&[4u8, 5u8, 6u8][..]));
    }

    // Timestamp partially-shredded tests grouped with the other partially-shredded cases.
    macro_rules! assert_variant_get_as_variant_array_with_default_option {
        ($variant_array: expr, $array_expected: expr) => {{
            let options = GetOptions::new();
            let array = $variant_array;
            let result = variant_get(&array, options).unwrap();
            let result = VariantArray::try_new(&result).unwrap();

            assert_eq!(result.len(), $array_expected.len());

            for (idx, item) in $array_expected.into_iter().enumerate() {
                match item {
                    Some(item) => assert_eq!(result.value(idx), item),
                    None => assert!(result.is_null(idx)),
                }
            }
        }};
    }

    partially_shredded_variant_array_gen!(
        partially_shredded_timestamp_micro_ntz_variant_array,
        || {
            arrow::array::TimestampMicrosecondArray::from(vec![
                Some(-456000),
                None,
                None,
                Some(1758602096000000),
            ])
        }
    );

    #[test]
    fn get_variant_partial_shredded_timestamp_micro_ntz_as_variant() {
        let array = partially_shredded_timestamp_micro_ntz_variant_array();
        assert_variant_get_as_variant_array_with_default_option!(
            array,
            vec![
                Some(Variant::from(
                    DateTime::from_timestamp_micros(-456000i64)
                        .unwrap()
                        .naive_utc(),
                )),
                None,
                Some(Variant::from("n/a")),
                Some(Variant::from(
                    DateTime::parse_from_rfc3339("2025-09-23T12:34:56+08:00")
                        .unwrap()
                        .naive_utc(),
                )),
            ]
        )
    }

    partially_shredded_variant_array_gen!(partially_shredded_timestamp_micro_variant_array, || {
        arrow::array::TimestampMicrosecondArray::from(vec![
            Some(-456000),
            None,
            None,
            Some(1758602096000000),
        ])
        .with_timezone("+00:00")
    });

    #[test]
    fn get_variant_partial_shredded_timestamp_micro_as_variant() {
        let array = partially_shredded_timestamp_micro_variant_array();
        assert_variant_get_as_variant_array_with_default_option!(
            array,
            vec![
                Some(Variant::from(
                    DateTime::from_timestamp_micros(-456000i64)
                        .unwrap()
                        .to_utc(),
                )),
                None,
                Some(Variant::from("n/a")),
                Some(Variant::from(
                    DateTime::parse_from_rfc3339("2025-09-23T12:34:56+08:00")
                        .unwrap()
                        .to_utc(),
                )),
            ]
        )
    }

    partially_shredded_variant_array_gen!(
        partially_shredded_timestamp_nano_ntz_variant_array,
        || {
            arrow::array::TimestampNanosecondArray::from(vec![
                Some(-4999999561),
                None,
                None,
                Some(1758602096000000000),
            ])
        }
    );

    #[test]
    fn get_variant_partial_shredded_timestamp_nano_ntz_as_variant() {
        let array = partially_shredded_timestamp_nano_ntz_variant_array();
        assert_variant_get_as_variant_array_with_default_option!(
            array,
            vec![
                Some(Variant::from(
                    DateTime::from_timestamp(-5, 439).unwrap().naive_utc()
                )),
                None,
                Some(Variant::from("n/a")),
                Some(Variant::from(
                    DateTime::parse_from_rfc3339("2025-09-23T12:34:56+08:00")
                        .unwrap()
                        .naive_utc()
                )),
            ]
        )
    }

    partially_shredded_variant_array_gen!(partially_shredded_timestamp_nano_variant_array, || {
        arrow::array::TimestampNanosecondArray::from(vec![
            Some(-4999999561),
            None,
            None,
            Some(1758602096000000000),
        ])
        .with_timezone("+00:00")
    });

    #[test]
    fn get_variant_partial_shredded_timestamp_nano_as_variant() {
        let array = partially_shredded_timestamp_nano_variant_array();
        assert_variant_get_as_variant_array_with_default_option!(
            array,
            vec![
                Some(Variant::from(
                    DateTime::from_timestamp(-5, 439).unwrap().to_utc()
                )),
                None,
                Some(Variant::from("n/a")),
                Some(Variant::from(
                    DateTime::parse_from_rfc3339("2025-09-23T12:34:56+08:00")
                        .unwrap()
                        .to_utc()
                )),
            ]
        )
    }

    /// Shredding: extract a value as an Int32Array
    #[test]
    fn get_variant_shredded_int32_as_int32_safe_cast() {
        // Extract the typed value as Int32Array
        let array = partially_shredded_int32_variant_array();
        // specify we want the typed value as Int32
        let field = Field::new("typed_value", DataType::Int32, true);
        let options = GetOptions::new().with_as_type(Some(FieldRef::from(field)));
        let result = variant_get(&array, options).unwrap();
        let expected: ArrayRef = Arc::new(Int32Array::from(vec![
            Some(34),
            None,
            None, // "n/a" is not an Int32 so converted to null
            Some(100),
        ]));
        assert_eq!(&result, &expected)
    }

    /// Shredding: extract a value as an Int32Array, unsafe cast (should error on "n/a")
    #[test]
    fn get_variant_shredded_int32_as_int32_unsafe_cast() {
        // Extract the typed value as Int32Array
        let array = partially_shredded_int32_variant_array();
        let field = Field::new("typed_value", DataType::Int32, true);
        let cast_options = CastOptions {
            safe: false, // unsafe cast
            ..Default::default()
        };
        let options = GetOptions::new()
            .with_as_type(Some(FieldRef::from(field)))
            .with_cast_options(cast_options);

        let err = variant_get(&array, options).unwrap_err();
        // TODO make this error message nicer (not Debug format)
        assert_eq!(
            err.to_string(),
            "Cast error: Failed to extract primitive of type Int32 from variant ShortString(ShortString(\"n/a\")) at path VariantPath([])"
        );
    }

    /// Perfect Shredding: extract the typed value as a VariantArray
    macro_rules! numeric_perfectly_shredded_test {
        ($primitive_type:ty, $data_fn:ident) => {
            let array = $data_fn();
            let options = GetOptions::new();
            let result = variant_get(&array, options).unwrap();

            // expect the result is a VariantArray
            let result = VariantArray::try_new(&result).unwrap();
            assert_eq!(result.len(), 3);

            // Expect the values are the same as the original values
            assert_eq!(
                result.value(0),
                Variant::from(<$primitive_type>::try_from(1u8).unwrap())
            );
            assert_eq!(
                result.value(1),
                Variant::from(<$primitive_type>::try_from(2u8).unwrap())
            );
            assert_eq!(
                result.value(2),
                Variant::from(<$primitive_type>::try_from(3u8).unwrap())
            );
        };
    }

    #[test]
    fn get_variant_perfectly_shredded_int8_as_variant() {
        numeric_perfectly_shredded_test!(i8, perfectly_shredded_int8_variant_array);
    }

    #[test]
    fn get_variant_perfectly_shredded_int16_as_variant() {
        numeric_perfectly_shredded_test!(i16, perfectly_shredded_int16_variant_array);
    }

    #[test]
    fn get_variant_perfectly_shredded_int32_as_variant() {
        numeric_perfectly_shredded_test!(i32, perfectly_shredded_int32_variant_array);
    }

    #[test]
    fn get_variant_perfectly_shredded_int64_as_variant() {
        numeric_perfectly_shredded_test!(i64, perfectly_shredded_int64_variant_array);
    }

    #[test]
    fn get_variant_perfectly_shredded_float32_as_variant() {
        numeric_perfectly_shredded_test!(f32, perfectly_shredded_float32_variant_array);
    }

    #[test]
    fn get_variant_perfectly_shredded_float64_as_variant() {
        numeric_perfectly_shredded_test!(f64, perfectly_shredded_float64_variant_array);
    }

    /// AllNull: extract a value as a VariantArray
    #[test]
    fn get_variant_all_null_as_variant() {
        let array = all_null_variant_array();
        let options = GetOptions::new();
        let result = variant_get(&array, options).unwrap();

        // expect the result is a VariantArray
        let result = VariantArray::try_new(&result).unwrap();
        assert_eq!(result.len(), 3);

        // All values should be null
        assert!(!result.is_valid(0));
        assert!(!result.is_valid(1));
        assert!(!result.is_valid(2));
    }

    /// AllNull: extract a value as an Int32Array
    #[test]
    fn get_variant_all_null_as_int32() {
        let array = all_null_variant_array();
        // specify we want the typed value as Int32
        let field = Field::new("typed_value", DataType::Int32, true);
        let options = GetOptions::new().with_as_type(Some(FieldRef::from(field)));
        let result = variant_get(&array, options).unwrap();

        let expected: ArrayRef = Arc::new(Int32Array::from(vec![
            Option::<i32>::None,
            Option::<i32>::None,
            Option::<i32>::None,
        ]));
        assert_eq!(&result, &expected)
    }

    macro_rules! perfectly_shredded_to_arrow_primitive_test {
        ($name:ident, $primitive_type:expr, $perfectly_shredded_array_gen_fun:ident, $expected_array:expr) => {
            #[test]
            fn $name() {
                let array = $perfectly_shredded_array_gen_fun();
                let field = Field::new("typed_value", $primitive_type, true);
                let options = GetOptions::new().with_as_type(Some(FieldRef::from(field)));
                let result = variant_get(&array, options).unwrap();
                let expected_array: ArrayRef = Arc::new($expected_array);
                assert_eq!(&result, &expected_array);
            }
        };
    }

    perfectly_shredded_to_arrow_primitive_test!(
        get_variant_perfectly_shredded_int18_as_int8,
        Int8,
        perfectly_shredded_int8_variant_array,
        Int8Array::from(vec![Some(1), Some(2), Some(3)])
    );

    perfectly_shredded_to_arrow_primitive_test!(
        get_variant_perfectly_shredded_int16_as_int16,
        Int16,
        perfectly_shredded_int16_variant_array,
        Int16Array::from(vec![Some(1), Some(2), Some(3)])
    );

    perfectly_shredded_to_arrow_primitive_test!(
        get_variant_perfectly_shredded_int32_as_int32,
        Int32,
        perfectly_shredded_int32_variant_array,
        Int32Array::from(vec![Some(1), Some(2), Some(3)])
    );

    perfectly_shredded_to_arrow_primitive_test!(
        get_variant_perfectly_shredded_int64_as_int64,
        Int64,
        perfectly_shredded_int64_variant_array,
        Int64Array::from(vec![Some(1), Some(2), Some(3)])
    );

    perfectly_shredded_to_arrow_primitive_test!(
        get_variant_perfectly_shredded_float32_as_float32,
        Float32,
        perfectly_shredded_float32_variant_array,
        Float32Array::from(vec![Some(1.0), Some(2.0), Some(3.0)])
    );

    perfectly_shredded_to_arrow_primitive_test!(
        get_variant_perfectly_shredded_float64_as_float64,
        Float64,
        perfectly_shredded_float64_variant_array,
        Float64Array::from(vec![Some(1.0), Some(2.0), Some(3.0)])
    );

    perfectly_shredded_to_arrow_primitive_test!(
        get_variant_perfectly_shredded_boolean_as_boolean,
        Boolean,
        perfectly_shredded_bool_variant_array,
        BooleanArray::from(vec![Some(true), Some(false), Some(true)])
    );

    perfectly_shredded_to_arrow_primitive_test!(
        get_variant_perfectly_shredded_utf8_as_utf8,
        DataType::Utf8,
        perfectly_shredded_utf8_variant_array,
        StringArray::from(vec![Some("foo"), Some("bar"), Some("baz")])
    );

    perfectly_shredded_to_arrow_primitive_test!(
        get_variant_perfectly_shredded_large_utf8_as_utf8,
        DataType::Utf8,
        perfectly_shredded_large_utf8_variant_array,
        StringArray::from(vec![Some("foo"), Some("bar"), Some("baz")])
    );

    perfectly_shredded_to_arrow_primitive_test!(
        get_variant_perfectly_shredded_utf8_view_as_utf8,
        DataType::Utf8,
        perfectly_shredded_utf8_view_variant_array,
        StringArray::from(vec![Some("foo"), Some("bar"), Some("baz")])
    );

    macro_rules! perfectly_shredded_variant_array_fn {
        ($func:ident, $typed_value_gen:expr) => {
            fn $func() -> ArrayRef {
                // Prefer producing fixtures with shred_variant from unshredded input.
                // Fall back for remaining non-shreddable test-only Arrow types (currently Null).
                let typed_value: ArrayRef = Arc::new($typed_value_gen());
                if let Some(shredded) = cast_to_variant(typed_value.as_ref())
                    .ok()
                    .and_then(|unshredded| shred_variant(&unshredded, typed_value.data_type()).ok())
                {
                    return shredded.into();
                }

                let metadata = BinaryViewArray::from_iter_values(std::iter::repeat_n(
                    EMPTY_VARIANT_METADATA_BYTES,
                    typed_value.len(),
                ));
                VariantArray::from_parts(Arc::new(metadata), None, Some(typed_value), None).into()
            }
        };
    }

    perfectly_shredded_variant_array_fn!(perfectly_shredded_utf8_variant_array, || {
        StringArray::from(vec![Some("foo"), Some("bar"), Some("baz")])
    });

    perfectly_shredded_variant_array_fn!(perfectly_shredded_large_utf8_variant_array, || {
        LargeStringArray::from(vec![Some("foo"), Some("bar"), Some("baz")])
    });

    perfectly_shredded_variant_array_fn!(perfectly_shredded_utf8_view_variant_array, || {
        StringViewArray::from(vec![Some("foo"), Some("bar"), Some("baz")])
    });

    perfectly_shredded_variant_array_fn!(perfectly_shredded_bool_variant_array, || {
        BooleanArray::from(vec![Some(true), Some(false), Some(true)])
    });

    /// Return a VariantArray that represents a perfectly "shredded" variant
    /// for the given typed value.
    ///
    /// The schema of the corresponding `StructArray` would look like this:
    ///
    /// ```text
    /// StructArray {
    ///   metadata: BinaryViewArray,
    ///   typed_value: Int32Array,
    /// }
    /// ```
    macro_rules! numeric_perfectly_shredded_variant_array_fn {
        ($func:ident, $array_type:ident, $primitive_type:ty) => {
            perfectly_shredded_variant_array_fn!($func, || {
                $array_type::from(vec![
                    Some(<$primitive_type>::try_from(1u8).unwrap()),
                    Some(<$primitive_type>::try_from(2u8).unwrap()),
                    Some(<$primitive_type>::try_from(3u8).unwrap()),
                ])
            });
        };
    }

    numeric_perfectly_shredded_variant_array_fn!(
        perfectly_shredded_int8_variant_array,
        Int8Array,
        i8
    );
    numeric_perfectly_shredded_variant_array_fn!(
        perfectly_shredded_int16_variant_array,
        Int16Array,
        i16
    );
    numeric_perfectly_shredded_variant_array_fn!(
        perfectly_shredded_int32_variant_array,
        Int32Array,
        i32
    );
    numeric_perfectly_shredded_variant_array_fn!(
        perfectly_shredded_int64_variant_array,
        Int64Array,
        i64
    );
    numeric_perfectly_shredded_variant_array_fn!(
        perfectly_shredded_float32_variant_array,
        Float32Array,
        f32
    );
    numeric_perfectly_shredded_variant_array_fn!(
        perfectly_shredded_float64_variant_array,
        Float64Array,
        f64
    );

    perfectly_shredded_variant_array_fn!(
        perfectly_shredded_timestamp_micro_ntz_variant_array,
        || {
            arrow::array::TimestampMicrosecondArray::from(vec![
                Some(-456000),
                Some(1758602096000001),
                Some(1758602096000002),
            ])
        }
    );

    perfectly_shredded_to_arrow_primitive_test!(
        get_variant_perfectly_shredded_timestamp_micro_ntz_as_timestamp_micro_ntz,
        DataType::Timestamp(TimeUnit::Microsecond, None),
        perfectly_shredded_timestamp_micro_ntz_variant_array,
        arrow::array::TimestampMicrosecondArray::from(vec![
            Some(-456000),
            Some(1758602096000001),
            Some(1758602096000002),
        ])
    );

    // test converting micro to nano
    perfectly_shredded_to_arrow_primitive_test!(
        get_variant_perfectly_shredded_timestamp_micro_ntz_as_nano_ntz,
        DataType::Timestamp(TimeUnit::Nanosecond, None),
        perfectly_shredded_timestamp_micro_ntz_variant_array,
        arrow::array::TimestampNanosecondArray::from(vec![
            Some(-456000000),
            Some(1758602096000001000),
            Some(1758602096000002000)
        ])
    );

    perfectly_shredded_variant_array_fn!(perfectly_shredded_timestamp_micro_variant_array, || {
        arrow::array::TimestampMicrosecondArray::from(vec![
            Some(-456000),
            Some(1758602096000001),
            Some(1758602096000002),
        ])
        .with_timezone("+00:00")
    });

    perfectly_shredded_to_arrow_primitive_test!(
        get_variant_perfectly_shredded_timestamp_micro_as_timestamp_micro,
        DataType::Timestamp(TimeUnit::Microsecond, Some(Arc::from("+00:00"))),
        perfectly_shredded_timestamp_micro_variant_array,
        arrow::array::TimestampMicrosecondArray::from(vec![
            Some(-456000),
            Some(1758602096000001),
            Some(1758602096000002),
        ])
        .with_timezone("+00:00")
    );

    // test converting micro to nano
    perfectly_shredded_to_arrow_primitive_test!(
        get_variant_perfectly_shredded_timestamp_micro_as_nano,
        DataType::Timestamp(TimeUnit::Nanosecond, Some(Arc::from("+00:00"))),
        perfectly_shredded_timestamp_micro_variant_array,
        arrow::array::TimestampNanosecondArray::from(vec![
            Some(-456000000),
            Some(1758602096000001000),
            Some(1758602096000002000)
        ])
        .with_timezone("+00:00")
    );

    perfectly_shredded_variant_array_fn!(
        perfectly_shredded_timestamp_nano_ntz_variant_array,
        || {
            arrow::array::TimestampNanosecondArray::from(vec![
                Some(-4999999561),
                Some(1758602096000000001),
                Some(1758602096000000002),
            ])
        }
    );

    perfectly_shredded_variant_array_fn!(
        perfectly_shredded_timestamp_micro_variant_array_for_second_and_milli_second,
        || {
            arrow::array::TimestampMicrosecondArray::from(vec![
                Some(1234),       // can't be cast to second & millisecond
                Some(1234000),    // can be cast to millisecond, but not second
                Some(1234000000), // can be cast to second & millisecond
            ])
            .with_timezone("+00:00")
        }
    );

    // The following two tests wants to cover the micro with timezone -> milli/second cases
    // there are three test items, which contains some items can be cast safely, and some can't
    perfectly_shredded_to_arrow_primitive_test!(
        get_variant_perfectly_shredded_timestamp_micro_as_timestamp_second,
        DataType::Timestamp(TimeUnit::Second, Some(Arc::from("+00:00"))),
        perfectly_shredded_timestamp_micro_variant_array_for_second_and_milli_second,
        arrow::array::TimestampSecondArray::from(vec![
            None,
            None, // Return None if can't be cast to second safely
            Some(1234)
        ])
        .with_timezone("+00:00")
    );

    perfectly_shredded_to_arrow_primitive_test!(
        get_variant_perfectly_shredded_timestamp_micro_as_timestamp_milli,
        DataType::Timestamp(TimeUnit::Millisecond, Some(Arc::from("+00:00"))),
        perfectly_shredded_timestamp_micro_variant_array_for_second_and_milli_second,
        arrow::array::TimestampMillisecondArray::from(vec![
            None, // Return None if can't be cast to millisecond safely
            Some(1234),
            Some(1234000)
        ])
        .with_timezone("+00:00")
    );

    perfectly_shredded_variant_array_fn!(
        perfectly_shredded_timestamp_micro_ntz_variant_array_for_second_and_milli_second,
        || {
            arrow::array::TimestampMicrosecondArray::from(vec![
                Some(1234),       // can't be cast to second & millisecond
                Some(1234000),    // can be cast to millisecond, but not second
                Some(1234000000), // can be cast to second & millisecond
            ])
        }
    );

    // The following two tests wants to cover the micro_ntz -> milli/second cases
    // there are three test items, which contains some items can be cast safely, and some can't
    perfectly_shredded_to_arrow_primitive_test!(
        get_variant_perfectly_shredded_timestamp_micro_ntz_as_timestamp_second,
        DataType::Timestamp(TimeUnit::Second, None),
        perfectly_shredded_timestamp_micro_ntz_variant_array_for_second_and_milli_second,
        arrow::array::TimestampSecondArray::from(vec![
            None,
            None, // Return None if can't be cast to second safely
            Some(1234)
        ])
    );

    perfectly_shredded_to_arrow_primitive_test!(
        get_variant_perfectly_shredded_timestamp_micro_ntz_as_timestamp_milli,
        DataType::Timestamp(TimeUnit::Millisecond, None),
        perfectly_shredded_timestamp_micro_ntz_variant_array_for_second_and_milli_second,
        arrow::array::TimestampMillisecondArray::from(vec![
            None, // Return None if can't be cast to millisecond safely
            Some(1234),
            Some(1234000)
        ])
    );

    perfectly_shredded_variant_array_fn!(
        perfectly_shredded_timestamp_nano_variant_array_for_second_and_milli_second,
        || {
            arrow::array::TimestampNanosecondArray::from(vec![
                Some(1234000),       // can't be cast to second & millisecond
                Some(1234000000),    // can be cast to millisecond, but not second
                Some(1234000000000), // can be cast to second & millisecond
            ])
            .with_timezone("+00:00")
        }
    );

    // The following two tests wants to cover the nano with timezone -> milli/second cases
    // there are three test items, which contains some items can be cast safely, and some can't
    perfectly_shredded_to_arrow_primitive_test!(
        get_variant_perfectly_shredded_timestamp_nano_as_timestamp_second,
        DataType::Timestamp(TimeUnit::Second, Some(Arc::from("+00:00"))),
        perfectly_shredded_timestamp_nano_variant_array_for_second_and_milli_second,
        arrow::array::TimestampSecondArray::from(vec![
            None,
            None, // Return None if can't be cast to second safely
            Some(1234)
        ])
        .with_timezone("+00:00")
    );

    perfectly_shredded_to_arrow_primitive_test!(
        get_variant_perfectly_shredded_timestamp_nano_as_timestamp_milli,
        DataType::Timestamp(TimeUnit::Millisecond, Some(Arc::from("+00:00"))),
        perfectly_shredded_timestamp_nano_variant_array_for_second_and_milli_second,
        arrow::array::TimestampMillisecondArray::from(vec![
            None, // Return None if can't be cast to millisecond safely
            Some(1234),
            Some(1234000)
        ])
        .with_timezone("+00:00")
    );

    perfectly_shredded_variant_array_fn!(
        perfectly_shredded_timestamp_nano_ntz_variant_array_for_second_and_milli_second,
        || {
            arrow::array::TimestampNanosecondArray::from(vec![
                Some(1234000),       // can't be cast to second & millisecond
                Some(1234000000),    // can be cast to millisecond, but not second
                Some(1234000000000), // can be cast to second & millisecond
            ])
        }
    );

    // The following two tests wants to cover the nano_ntz -> milli/second cases
    // there are three test items, which contains some items can be cast safely, and some can't
    perfectly_shredded_to_arrow_primitive_test!(
        get_variant_perfectly_shredded_timestamp_nano_ntz_as_timestamp_second,
        DataType::Timestamp(TimeUnit::Second, None),
        perfectly_shredded_timestamp_nano_ntz_variant_array_for_second_and_milli_second,
        arrow::array::TimestampSecondArray::from(vec![
            None,
            None, // Return None if can't be cast to second safely
            Some(1234)
        ])
    );

    perfectly_shredded_to_arrow_primitive_test!(
        get_variant_perfectly_shredded_timestamp_nano_ntz_as_timestamp_milli,
        DataType::Timestamp(TimeUnit::Millisecond, None),
        perfectly_shredded_timestamp_nano_ntz_variant_array_for_second_and_milli_second,
        arrow::array::TimestampMillisecondArray::from(vec![
            None, // Return None if can't be cast to millisecond safely
            Some(1234),
            Some(1234000)
        ])
    );

    perfectly_shredded_to_arrow_primitive_test!(
        get_variant_perfectly_shredded_timestamp_nano_ntz_as_timestamp_nano_ntz,
        DataType::Timestamp(TimeUnit::Nanosecond, None),
        perfectly_shredded_timestamp_nano_ntz_variant_array,
        arrow::array::TimestampNanosecondArray::from(vec![
            Some(-4999999561),
            Some(1758602096000000001),
            Some(1758602096000000002),
        ])
    );

    perfectly_shredded_variant_array_fn!(perfectly_shredded_timestamp_nano_variant_array, || {
        arrow::array::TimestampNanosecondArray::from(vec![
            Some(-4999999561),
            Some(1758602096000000001),
            Some(1758602096000000002),
        ])
        .with_timezone("+00:00")
    });

    perfectly_shredded_to_arrow_primitive_test!(
        get_variant_perfectly_shredded_timestamp_nano_as_timestamp_nano,
        DataType::Timestamp(TimeUnit::Nanosecond, Some(Arc::from("+00:00"))),
        perfectly_shredded_timestamp_nano_variant_array,
        arrow::array::TimestampNanosecondArray::from(vec![
            Some(-4999999561),
            Some(1758602096000000001),
            Some(1758602096000000002),
        ])
        .with_timezone("+00:00")
    );

    perfectly_shredded_variant_array_fn!(perfectly_shredded_date_variant_array, || {
        Date32Array::from(vec![Some(-12345), Some(17586), Some(20000)])
    });

    perfectly_shredded_to_arrow_primitive_test!(
        get_variant_perfectly_shredded_date_as_date,
        DataType::Date32,
        perfectly_shredded_date_variant_array,
        Date32Array::from(vec![Some(-12345), Some(17586), Some(20000)])
    );

    perfectly_shredded_to_arrow_primitive_test!(
        get_variant_perfectly_shredded_date_as_date64,
        DataType::Date64,
        perfectly_shredded_date_variant_array,
        Date64Array::from(vec![
            Some(-1066608000000),
            Some(1519430400000),
            Some(1728000000000)
        ])
    );

    perfectly_shredded_variant_array_fn!(perfectly_shredded_time_variant_array, || {
        Time64MicrosecondArray::from(vec![Some(12345000), Some(87654000), Some(135792000)])
    });

    perfectly_shredded_to_arrow_primitive_test!(
        get_variant_perfectly_shredded_time_as_time,
        DataType::Time64(TimeUnit::Microsecond),
        perfectly_shredded_time_variant_array,
        Time64MicrosecondArray::from(vec![Some(12345000), Some(87654000), Some(135792000)])
    );

    perfectly_shredded_to_arrow_primitive_test!(
        get_variant_perfectly_shredded_time_as_time64_nano,
        DataType::Time64(TimeUnit::Nanosecond),
        perfectly_shredded_time_variant_array,
        Time64NanosecondArray::from(vec![
            Some(12345000000),
            Some(87654000000),
            Some(135792000000)
        ])
    );

    perfectly_shredded_variant_array_fn!(perfectly_shredded_time_variant_array_for_time32, || {
        Time64MicrosecondArray::from(vec![
            Some(1234),        // This can't be cast to Time32 losslessly
            Some(7654000),     // This can be cast to Time32(Millisecond), but not Time32(Second)
            Some(35792000000), // This can be cast to Time32(Second) & Time32(Millisecond)
        ])
    });

    perfectly_shredded_to_arrow_primitive_test!(
        get_variant_perfectly_shredded_time_as_time32_second,
        DataType::Time32(TimeUnit::Second),
        perfectly_shredded_time_variant_array_for_time32,
        Time32SecondArray::from(vec![
            None,
            None, // Return None if can't be cast to Time32(Second) safely
            Some(35792)
        ])
    );

    perfectly_shredded_to_arrow_primitive_test!(
        get_variant_perfectly_shredded_time_as_time32_milli,
        DataType::Time32(TimeUnit::Millisecond),
        perfectly_shredded_time_variant_array_for_time32,
        Time32MillisecondArray::from(vec![
            None, // Return None if can't be cast to Time32(Second) safely
            Some(7654),
            Some(35792000)
        ])
    );

    perfectly_shredded_variant_array_fn!(perfectly_shredded_null_variant_array, || {
        let mut builder = NullBuilder::new();
        builder.append_nulls(3);
        builder.finish()
    });

    perfectly_shredded_to_arrow_primitive_test!(
        get_variant_perfectly_shredded_null_as_null,
        DataType::Null,
        perfectly_shredded_null_variant_array,
        arrow::array::NullArray::new(3)
    );

    perfectly_shredded_variant_array_fn!(perfectly_shredded_null_variant_array_with_int, || {
        Int32Array::from(vec![Some(32), Some(64), Some(48)])
    });

    // We append null values if type miss match happens in safe mode
    perfectly_shredded_to_arrow_primitive_test!(
        get_variant_perfectly_shredded_null_with_type_missmatch_in_safe_mode,
        DataType::Null,
        perfectly_shredded_null_variant_array_with_int,
        arrow::array::NullArray::new(3)
    );

    // We'll return an error if type miss match happens in strict mode
    #[test]
    fn get_variant_perfectly_shredded_null_as_null_with_type_missmatch_in_strict_mode() {
        let array = perfectly_shredded_null_variant_array_with_int();
        let field = Field::new("typed_value", DataType::Null, true);
        let options = GetOptions::new()
            .with_as_type(Some(FieldRef::from(field)))
            .with_cast_options(CastOptions {
                safe: false,
                format_options: FormatOptions::default(),
            });

        let result = variant_get(&array, options);

        assert!(result.is_err());
        let error_msg = format!("{}", result.unwrap_err());
        assert!(
            error_msg
                .contains("Cast error: Failed to extract primitive of type Null from variant Int32(32) at path VariantPath([])"),
            "Expected=[Cast error: Failed to extract primitive of type Null from variant Int32(32) at path VariantPath([])],\
                Got error message=[{}]",
            error_msg
        );
    }

    perfectly_shredded_variant_array_fn!(perfectly_shredded_decimal4_variant_array, || {
        Decimal32Array::from(vec![Some(12345), Some(23400), Some(-12342)])
            .with_precision_and_scale(5, 2)
            .unwrap()
    });

    perfectly_shredded_to_arrow_primitive_test!(
        get_variant_perfectly_shredded_decimal4_as_decimal4,
        DataType::Decimal32(5, 2),
        perfectly_shredded_decimal4_variant_array,
        Decimal32Array::from(vec![Some(12345), Some(23400), Some(-12342)])
            .with_precision_and_scale(5, 2)
            .unwrap()
    );

    perfectly_shredded_variant_array_fn!(
        perfectly_shredded_decimal8_variant_array_cast2decimal32,
        || {
            Decimal64Array::from(vec![Some(123456), Some(145678), Some(-123456)])
                .with_precision_and_scale(6, 1)
                .unwrap()
        }
    );

    // The input will be cast to Decimal32 when transformed to Variant
    // This tests will covert the logic DataType::Decimal64(the original array)
    // -> Variant::Decimal4(VariantArray) -> DataType::Decimal64(the result array)
    perfectly_shredded_to_arrow_primitive_test!(
        get_variant_perfectly_shredded_decimal8_through_decimal32_as_decimal8,
        DataType::Decimal64(6, 1),
        perfectly_shredded_decimal8_variant_array_cast2decimal32,
        Decimal64Array::from(vec![Some(123456), Some(145678), Some(-123456)])
            .with_precision_and_scale(6, 1)
            .unwrap()
    );

    // This tests will covert the logic DataType::Decimal64(the original array)
    //  -> Variant::Decimal8(VariantArray) -> DataType::Decimal64(the result array)
    perfectly_shredded_variant_array_fn!(perfectly_shredded_decimal8_variant_array, || {
        Decimal64Array::from(vec![Some(1234567809), Some(1456787000), Some(-1234561203)])
            .with_precision_and_scale(10, 1)
            .unwrap()
    });

    perfectly_shredded_to_arrow_primitive_test!(
        get_variant_perfectly_shredded_decimal8_as_decimal8,
        DataType::Decimal64(10, 1),
        perfectly_shredded_decimal8_variant_array,
        Decimal64Array::from(vec![Some(1234567809), Some(1456787000), Some(-1234561203)])
            .with_precision_and_scale(10, 1)
            .unwrap()
    );

    // This tests will covert the logic DataType::Decimal128(the original array)
    //  -> Variant::Decimal4(VariantArray) -> DataType::Decimal128(the result array)
    perfectly_shredded_variant_array_fn!(
        perfectly_shredded_decimal16_within_decimal4_variant_array,
        || {
            Decimal128Array::from(vec![
                Some(i128::from(1234589)),
                Some(i128::from(2344444)),
                Some(i128::from(-1234789)),
            ])
            .with_precision_and_scale(7, 3)
            .unwrap()
        }
    );

    // This tests will covert the logic DataType::Decimal128(the original array)
    // -> Variant::Decimal4(VariantArray) -> DataType::Decimal128(the result array)
    perfectly_shredded_to_arrow_primitive_test!(
        get_variant_perfectly_shredded_decimal16_within_decimal4_as_decimal16,
        DataType::Decimal128(7, 3),
        perfectly_shredded_decimal16_within_decimal4_variant_array,
        Decimal128Array::from(vec![
            Some(i128::from(1234589)),
            Some(i128::from(2344444)),
            Some(i128::from(-1234789)),
        ])
        .with_precision_and_scale(7, 3)
        .unwrap()
    );

    perfectly_shredded_variant_array_fn!(
        perfectly_shredded_decimal16_within_decimal8_variant_array,
        || {
            Decimal128Array::from(vec![Some(1234567809), Some(1456787000), Some(-1234561203)])
                .with_precision_and_scale(10, 1)
                .unwrap()
        }
    );

    // This tests will covert the logic DataType::Decimal128(the original array)
    // -> Variant::Decimal8(VariantArray) -> DataType::Decimal128(the result array)
    perfectly_shredded_to_arrow_primitive_test!(
        get_variant_perfectly_shredded_decimal16_within8_as_decimal16,
        DataType::Decimal128(10, 1),
        perfectly_shredded_decimal16_within_decimal8_variant_array,
        Decimal128Array::from(vec![Some(1234567809), Some(1456787000), Some(-1234561203)])
            .with_precision_and_scale(10, 1)
            .unwrap()
    );

    perfectly_shredded_variant_array_fn!(perfectly_shredded_decimal16_variant_array, || {
        Decimal128Array::from(vec![
            Some(i128::from_str("12345678901234567899").unwrap()),
            Some(i128::from_str("23445677483748324300").unwrap()),
            Some(i128::from_str("-12345678901234567899").unwrap()),
        ])
        .with_precision_and_scale(20, 3)
        .unwrap()
    });

    // This tests will covert the logic DataType::Decimal128(the original array)
    // -> Variant::Decimal16(VariantArray) -> DataType::Decimal128(the result array)
    perfectly_shredded_to_arrow_primitive_test!(
        get_variant_perfectly_shredded_decimal16_as_decimal16,
        DataType::Decimal128(20, 3),
        perfectly_shredded_decimal16_variant_array,
        Decimal128Array::from(vec![
            Some(i128::from_str("12345678901234567899").unwrap()),
            Some(i128::from_str("23445677483748324300").unwrap()),
            Some(i128::from_str("-12345678901234567899").unwrap())
        ])
        .with_precision_and_scale(20, 3)
        .unwrap()
    );

    perfectly_shredded_variant_array_fn!(perfectly_shredded_binary_variant_array, || {
        BinaryArray::from(vec![
            Some(b"Apache" as &[u8]),
            Some(b"Arrow-rs" as &[u8]),
            Some(b"Parquet-variant" as &[u8]),
        ])
    });

    perfectly_shredded_to_arrow_primitive_test!(
        get_variant_perfectly_shredded_binary_as_binary,
        DataType::Binary,
        perfectly_shredded_binary_variant_array,
        BinaryArray::from(vec![
            Some(b"Apache" as &[u8]),
            Some(b"Arrow-rs" as &[u8]),
            Some(b"Parquet-variant" as &[u8]),
        ])
    );

    perfectly_shredded_variant_array_fn!(perfectly_shredded_large_binary_variant_array, || {
        LargeBinaryArray::from(vec![
            Some(b"Apache" as &[u8]),
            Some(b"Arrow-rs" as &[u8]),
            Some(b"Parquet-variant" as &[u8]),
        ])
    });

    perfectly_shredded_to_arrow_primitive_test!(
        get_variant_perfectly_shredded_large_binary_as_large_binary,
        DataType::LargeBinary,
        perfectly_shredded_large_binary_variant_array,
        LargeBinaryArray::from(vec![
            Some(b"Apache" as &[u8]),
            Some(b"Arrow-rs" as &[u8]),
            Some(b"Parquet-variant" as &[u8]),
        ])
    );

    perfectly_shredded_variant_array_fn!(perfectly_shredded_binary_view_variant_array, || {
        BinaryViewArray::from(vec![
            Some(b"Apache" as &[u8]),
            Some(b"Arrow-rs" as &[u8]),
            Some(b"Parquet-variant" as &[u8]),
        ])
    });

    perfectly_shredded_to_arrow_primitive_test!(
        get_variant_perfectly_shredded_binary_view_as_binary_view,
        DataType::BinaryView,
        perfectly_shredded_binary_view_variant_array,
        BinaryViewArray::from(vec![
            Some(b"Apache" as &[u8]),
            Some(b"Arrow-rs" as &[u8]),
            Some(b"Parquet-variant" as &[u8]),
        ])
    );

    /// Return a VariantArray that represents an "all null" variant
    /// for the following example (3 null values):
    ///
    /// ```text
    /// null
    /// null
    /// null
    /// ```
    ///
    /// The schema of the corresponding `StructArray` would look like this:
    ///
    /// ```text
    /// StructArray {
    ///   metadata: BinaryViewArray,
    /// }
    /// ```
    fn all_null_variant_array() -> ArrayRef {
        let nulls = NullBuffer::from(vec![
            false, // row 0 is null
            false, // row 1 is null
            false, // row 2 is null
        ]);

        // metadata is the same for all rows (though they're all null)
        let metadata =
            BinaryViewArray::from_iter_values(std::iter::repeat_n(EMPTY_VARIANT_METADATA_BYTES, 3));

        ArrayRef::from(VariantArray::from_parts(
            Arc::new(metadata),
            None,
            None,
            Some(nulls),
        ))
    }

    /// This test manually constructs a shredded variant array representing objects
    /// like {"x": 1, "y": "foo"} and {"x": 42} and tests extracting the "x" field
    /// as VariantArray using variant_get.
    #[test]
    fn test_shredded_object_field_access() {
        let array = shredded_object_with_x_field_variant_array();

        // Test: Extract the "x" field as VariantArray first
        let options = GetOptions::new_with_path(VariantPath::try_from("x").unwrap());
        let result = variant_get(&array, options).unwrap();

        let result_variant = VariantArray::try_new(&result).unwrap();
        assert_eq!(result_variant.len(), 2);

        // Row 0: expect x=1
        assert_eq!(result_variant.value(0), Variant::Int32(1));
        // Row 1: expect x=42
        assert_eq!(result_variant.value(1), Variant::Int32(42));
    }

    /// Test extracting shredded object field with type conversion
    #[test]
    fn test_shredded_object_field_as_int32() {
        let array = shredded_object_with_x_field_variant_array();

        // Test: Extract the "x" field as Int32Array (type conversion)
        let field = Field::new("x", DataType::Int32, false);
        let options = GetOptions::new_with_path(VariantPath::try_from("x").unwrap())
            .with_as_type(Some(FieldRef::from(field)));
        let result = variant_get(&array, options).unwrap();

        // Should get Int32Array
        let expected: ArrayRef = Arc::new(Int32Array::from(vec![Some(1), Some(42)]));
        assert_eq!(&result, &expected);
    }

    type ShreddedListLikeArrayGen = fn() -> ArrayRef;
    type ShreddedListLikeCase = (&'static str, ShreddedListLikeArrayGen);

    fn shredded_list_like_cases() -> [ShreddedListLikeCase; 4] {
        [
            ("list", shredded_list_variant_array),
            ("large_list", shredded_large_list_variant_array),
            ("list_view", shredded_list_view_variant_array),
            ("large_list_view", shredded_large_list_view_variant_array),
        ]
    }

    #[test]
    fn test_shredded_list_like_index_access_from_value_field() {
        let options = GetOptions::new_with_path(VariantPath::from(1));

        for (case, array_gen) in shredded_list_like_cases() {
            let array = array_gen();
            let result = variant_get(&array, options.clone()).unwrap();
            let result_variant = VariantArray::try_new(&result).unwrap();

            assert_eq!(result_variant.value(0), Variant::from("drama"), "{case}");
            assert_eq!(result_variant.value(1).as_int64(), Some(123), "{case}");
        }
    }

    #[test]
    fn test_shredded_list_like_index_out_of_bounds_unsafe_cast_returns_null() {
        let options =
            GetOptions::new_with_path(VariantPath::from(10)).with_cast_options(CastOptions {
                safe: false,
                ..Default::default()
            });

        for (case, array_gen) in shredded_list_like_cases() {
            let result = variant_get(&array_gen(), options.clone()).unwrap();
            let result_variant = VariantArray::try_new(&result).unwrap();
            assert_eq!(result_variant.value(0), Variant::Null, "{case}");
            assert_eq!(result_variant.value(1), Variant::Null, "{case}");
        }
    }

    /// Test extracting shredded list-like field with type conversion.
    #[test]
    fn test_shredded_list_like_as_string() {
        let field = Field::new("typed_value", DataType::Utf8, false);
        let options = GetOptions::new_with_path(VariantPath::from(0))
            .with_as_type(Some(FieldRef::from(field)));
        let expected: ArrayRef = Arc::new(StringArray::from(vec![Some("comedy"), Some("horror")]));

        for (case, array_gen) in shredded_list_like_cases() {
            let result = variant_get(&array_gen(), options.clone()).unwrap();
            assert_eq!(&result, &expected, "{case}");
        }
    }

    #[test]
    fn test_shredded_list_like_index_access_from_value_field_as_int64() {
        let field = Field::new("typed_value", DataType::Int64, true);
        let options = GetOptions::new_with_path(VariantPath::from(1))
            .with_as_type(Some(FieldRef::from(field)));
        let expected: ArrayRef = Arc::new(Int64Array::from(vec![None, Some(123)]));

        for (case, array_gen) in shredded_list_like_cases() {
            let result = variant_get(&array_gen(), options.clone()).unwrap();
            // "drama" -> NULL, 123 -> 123.
            assert_eq!(&result, &expected, "{case}");
        }
    }

    #[test]
    fn test_shredded_list_in_struct_index_access() {
        let array = shredded_struct_with_list_variant_array();
        let options = GetOptions::new_with_path(VariantPath::try_from("a[1]").unwrap());
        let result = variant_get(&array, options).unwrap();
        let result_variant = VariantArray::try_new(&result).unwrap();

        assert_eq!(result_variant.value(0), Variant::from("drama"));
        assert_eq!(result_variant.value(1).as_int64(), Some(123));
    }

    #[test]
    fn test_shredded_struct_in_list_field_access() {
        let array = shredded_list_of_struct_variant_array();
        let field = Field::new("x", DataType::Int32, true);
        let path = VariantPath::from(0).join("x");
        let options = GetOptions::new_with_path(path).with_as_type(Some(FieldRef::from(field)));
        let result = variant_get(&array, options).unwrap();

        let expected: ArrayRef = Arc::new(Int32Array::from(vec![Some(1), Some(3)]));
        assert_eq!(&result, &expected);
    }

    #[test]
    fn test_shredded_list_of_lists_index_access() {
        let array = shredded_list_of_lists_variant_array();
        let path = VariantPath::from(0).join(1);

        let result = variant_get(&array, GetOptions::new_with_path(path.clone())).unwrap();
        let result_variant = VariantArray::try_new(&result).unwrap();
        assert_eq!(result_variant.value(0), Variant::from("b"));
        assert_eq!(result_variant.value(1).as_int64(), Some(123));

        let field = Field::new("typed_value", DataType::Int64, true);
        let casted = variant_get(
            &array,
            GetOptions::new_with_path(path).with_as_type(Some(FieldRef::from(field))),
        )
        .unwrap();
        let expected: ArrayRef = Arc::new(Int64Array::from(vec![None, Some(123)]));
        assert_eq!(&casted, &expected);
    }

    /// Helper to create a shredded list-like variant array used by list index tests.
    ///
    /// Rows:
    /// 1. `["comedy", "drama"]` (fully shred-able as `Utf8`)
    /// 2. `["horror", 123]` (partially shredded, with fallback for the numeric element)
    fn shredded_list_like_variant_array(list_schema: DataType) -> ArrayRef {
        let json_rows: ArrayRef = Arc::new(StringArray::from(vec![
            Some(r#"["comedy", "drama"]"#),
            Some(r#"["horror", 123]"#),
        ]));
        let input = json_to_variant(&json_rows).unwrap();

        let shredded = shred_variant(&input, &list_schema).unwrap();
        ArrayRef::from(shredded)
    }

    fn shredded_list_of_lists_variant_array() -> ArrayRef {
        let json_rows: ArrayRef = Arc::new(StringArray::from(vec![
            Some(r#"[["a", "b"], ["c", "d"]]"#),
            Some(r#"[["x", 123], ["y", "z"]]"#),
        ]));
        let input = json_to_variant(&json_rows).unwrap();

        let inner_list = DataType::List(Arc::new(Field::new("item", DataType::Utf8, true)));
        let outer_list = DataType::List(Arc::new(Field::new("item", inner_list, true)));
        let shredded = shred_variant(&input, &outer_list).unwrap();
        ArrayRef::from(shredded)
    }

    fn shredded_list_variant_array() -> ArrayRef {
        shredded_list_like_variant_array(DataType::List(Arc::new(Field::new(
            "item",
            DataType::Utf8,
            true,
        ))))
    }

    fn shredded_large_list_variant_array() -> ArrayRef {
        shredded_list_like_variant_array(DataType::LargeList(Arc::new(Field::new(
            "item",
            DataType::Utf8,
            true,
        ))))
    }

    fn shredded_list_view_variant_array() -> ArrayRef {
        shredded_list_like_variant_array(DataType::ListView(Arc::new(Field::new(
            "item",
            DataType::Utf8,
            true,
        ))))
    }

    fn shredded_large_list_view_variant_array() -> ArrayRef {
        shredded_list_like_variant_array(DataType::LargeListView(Arc::new(Field::new(
            "item",
            DataType::Utf8,
            true,
        ))))
    }

    fn shredded_struct_with_list_variant_array() -> ArrayRef {
        let json_rows: ArrayRef = Arc::new(StringArray::from(vec![
            Some(r#"{"a": ["comedy", "drama"]}"#),
            Some(r#"{"a": ["horror", 123]}"#),
        ]));
        let input = json_to_variant(&json_rows).unwrap();

        let list_schema = DataType::List(Arc::new(Field::new("item", DataType::Utf8, true)));
        let shredding_schema = ShreddedSchemaBuilder::default()
            .with_path("a", &list_schema)
            .unwrap()
            .build();
        let shredded = shred_variant(&input, &shredding_schema).unwrap();
        ArrayRef::from(shredded)
    }

    fn shredded_list_of_struct_variant_array() -> ArrayRef {
        let json_rows: ArrayRef = Arc::new(StringArray::from(vec![
            Some(r#"[{"x": 1}, {"x": 2}]"#),
            Some(r#"[{"x": 3}, {"y": 4}]"#),
        ]));
        let input = json_to_variant(&json_rows).unwrap();

        let struct_type =
            DataType::Struct(Fields::from(vec![Field::new("x", DataType::Int32, true)]));
        let list_schema = DataType::List(Arc::new(Field::new("item", struct_type, true)));
        let shredded = shred_variant(&input, &list_schema).unwrap();
        ArrayRef::from(shredded)
    }

    /// Helper function to create a shredded variant array representing objects
    ///
    /// This creates an array that represents:
    /// Row 0: {"x": 1, "y": "foo"}  (x is shredded, y is in value field)
    /// Row 1: {"x": 42}             (x is shredded, perfect shredding)
    ///
    /// The physical layout follows the shredding spec where:
    /// - metadata: contains object metadata
    /// - typed_value: StructArray with field "x" (ShreddedVariantFieldArray)
    /// - value: contains fallback for unshredded fields like {"y": "foo"}
    /// - The "x" field has typed_value=Int32Array and value=NULL (perfect shredding)
    fn shredded_object_with_x_field_variant_array() -> ArrayRef {
        // Create the base metadata for objects
        let (metadata, y_field_value) = {
            let mut builder = parquet_variant::VariantBuilder::new();
            let mut obj = builder.new_object();
            obj.insert("x", Variant::Int32(42));
            obj.insert("y", Variant::from("foo"));
            obj.finish();
            builder.finish()
        };

        // Create metadata array (same for both rows)
        let metadata_array = BinaryViewArray::from_iter_values(std::iter::repeat_n(&metadata, 2));

        // Create the main value field per the 3-step shredding spec:
        // Step 2: If field not in shredding schema, check value field
        // Row 0: {"y": "foo"} (y is not shredded, stays in value for step 2)
        // Row 1: {} (empty object - no unshredded fields)
        let empty_object_value = {
            let mut builder = parquet_variant::VariantBuilder::new();
            let obj = builder.new_object();
            obj.finish();
            let (_, value) = builder.finish();
            value
        };

        let value_array = BinaryViewArray::from(vec![
            Some(y_field_value.as_slice()),      // Row 0 has {"y": "foo"}
            Some(empty_object_value.as_slice()), // Row 1 has {}
        ]);

        // Create the "x" field as a ShreddedVariantFieldArray
        // This represents the shredded Int32 values for the "x" field
        let x_field_typed_value = Int32Array::from(vec![Some(1), Some(42)]);

        // For perfect shredding of the x field, no "value" column, only typed_value
        let x_field_shredded = ShreddedVariantFieldArray::from_parts(
            None,
            Some(Arc::new(x_field_typed_value) as ArrayRef),
            None,
        );

        // Create the main typed_value as a struct containing the "x" field
        let typed_value_fields = Fields::from(vec![Field::new(
            "x",
            x_field_shredded.data_type().clone(),
            true,
        )]);
        let typed_value_struct = StructArray::try_new(
            typed_value_fields,
            vec![ArrayRef::from(x_field_shredded)],
            None, // No nulls - both rows have the object structure
        )
        .unwrap();

        // Create the main VariantArray
        ArrayRef::from(VariantArray::from_parts(
            Arc::new(metadata_array),
            Some(Arc::new(value_array)),
            Some(Arc::new(typed_value_struct)),
            None,
        ))
    }

    /// Simple test to check if nested paths are supported by current implementation
    #[test]
    fn test_simple_nested_path_support() {
        // Check: How does VariantPath parse different strings?
        println!("Testing path parsing:");

        let path_x = VariantPath::try_from("x").unwrap();
        let elements_x: Vec<_> = path_x.iter().collect();
        println!("  'x' -> {} elements: {:?}", elements_x.len(), elements_x);

        let path_ax = VariantPath::try_from("a.x").unwrap();
        let elements_ax: Vec<_> = path_ax.iter().collect();
        println!(
            "  'a.x' -> {} elements: {:?}",
            elements_ax.len(),
            elements_ax
        );

        let path_ax_alt = VariantPath::try_from("$.a.x").unwrap();
        let elements_ax_alt: Vec<_> = path_ax_alt.iter().collect();
        println!(
            "  '$.a.x' -> {} elements: {:?}",
            elements_ax_alt.len(),
            elements_ax_alt
        );

        let path_nested = VariantPath::try_from("a").unwrap().join("x");
        let elements_nested: Vec<_> = path_nested.iter().collect();
        println!(
            "  VariantPath::try_from('a').unwrap().join('x') -> {} elements: {:?}",
            elements_nested.len(),
            elements_nested
        );

        // Use your existing simple test data but try "a.x" instead of "x"
        let array = shredded_object_with_x_field_variant_array();

        // Test if variant_get with REAL nested path throws not implemented error
        let real_nested_path = VariantPath::try_from("a").unwrap().join("x");
        let options = GetOptions::new_with_path(real_nested_path);
        let result = variant_get(&array, options);

        match result {
            Ok(_) => {
                println!("Nested path 'a.x' works unexpectedly!");
            }
            Err(e) => {
                println!("Nested path 'a.x' error: {}", e);
                if e.to_string().contains("Not yet implemented")
                    || e.to_string().contains("NotYetImplemented")
                {
                    println!("This is expected - nested paths are not implemented");
                    return;
                }
                // Any other error is also expected for now
                println!("This shows nested paths need implementation");
            }
        }
    }

    /// Test comprehensive variant_get scenarios with Int32 conversion
    /// Test depth 0: Direct field access "x" with Int32 conversion
    /// Covers shredded vs non-shredded VariantArrays for simple field access
    #[test]
    fn test_depth_0_int32_conversion() {
        println!("=== Testing Depth 0: Direct field access ===");

        // Non-shredded test data: [{"x": 42}, {"x": "foo"}, {"y": 10}]
        let unshredded_array = create_depth_0_test_data();

        let field = Field::new("result", DataType::Int32, true);
        let path = VariantPath::try_from("x").unwrap();
        let options = GetOptions::new_with_path(path).with_as_type(Some(FieldRef::from(field)));
        let result = variant_get(&unshredded_array, options).unwrap();

        let expected: ArrayRef = Arc::new(Int32Array::from(vec![
            Some(42), // {"x": 42} -> 42
            None,     // {"x": "foo"} -> NULL (type mismatch)
            None,     // {"y": 10} -> NULL (field missing)
        ]));
        assert_eq!(&result, &expected);
        println!("Depth 0 (unshredded) passed");

        // Shredded test data: using simplified approach based on working pattern
        let shredded_array = create_depth_0_shredded_test_data_simple();

        let field = Field::new("result", DataType::Int32, true);
        let path = VariantPath::try_from("x").unwrap();
        let options = GetOptions::new_with_path(path).with_as_type(Some(FieldRef::from(field)));
        let result = variant_get(&shredded_array, options).unwrap();

        let expected: ArrayRef = Arc::new(Int32Array::from(vec![
            Some(42), // {"x": 42} -> 42 (from typed_value)
            None,     // {"x": "foo"} -> NULL (type mismatch, from value field)
        ]));
        assert_eq!(&result, &expected);
        println!("Depth 0 (shredded) passed");
    }

    /// Test depth 1: Single nested field access "a.x" with Int32 conversion
    /// Covers shredded vs non-shredded VariantArrays for nested field access
    #[test]
    fn test_depth_1_int32_conversion() {
        println!("=== Testing Depth 1: Single nested field access ===");

        // Non-shredded test data from the GitHub issue
        let unshredded_array = create_nested_path_test_data();

        let field = Field::new("result", DataType::Int32, true);
        let path = VariantPath::try_from("a.x").unwrap(); // Dot notation!
        let options = GetOptions::new_with_path(path).with_as_type(Some(FieldRef::from(field)));
        let result = variant_get(&unshredded_array, options).unwrap();

        let expected: ArrayRef = Arc::new(Int32Array::from(vec![
            Some(55), // {"a": {"x": 55}} -> 55
            None,     // {"a": {"x": "foo"}} -> NULL (type mismatch)
        ]));
        assert_eq!(&result, &expected);
        println!("Depth 1 (unshredded) passed");

        // Shredded test data: depth 1 nested shredding
        let shredded_array = create_depth_1_shredded_test_data_working();

        let field = Field::new("result", DataType::Int32, true);
        let path = VariantPath::try_from("a.x").unwrap(); // Dot notation!
        let options = GetOptions::new_with_path(path).with_as_type(Some(FieldRef::from(field)));
        let result = variant_get(&shredded_array, options).unwrap();

        let expected: ArrayRef = Arc::new(Int32Array::from(vec![
            Some(55), // {"a": {"x": 55}} -> 55 (from nested shredded x)
            None,     // {"a": {"x": "foo"}} -> NULL (type mismatch in nested value)
        ]));
        assert_eq!(&result, &expected);
        println!("Depth 1 (shredded) passed");
    }

    /// Test depth 2: Double nested field access "a.b.x" with Int32 conversion  
    /// Covers shredded vs non-shredded VariantArrays for deeply nested field access
    #[test]
    fn test_depth_2_int32_conversion() {
        println!("=== Testing Depth 2: Double nested field access ===");

        // Non-shredded test data: [{"a": {"b": {"x": 100}}}, {"a": {"b": {"x": "bar"}}}, {"a": {"b": {"y": 200}}}]
        let unshredded_array = create_depth_2_test_data();

        let field = Field::new("result", DataType::Int32, true);
        let path = VariantPath::try_from("a.b.x").unwrap(); // Double nested dot notation!
        let options = GetOptions::new_with_path(path).with_as_type(Some(FieldRef::from(field)));
        let result = variant_get(&unshredded_array, options).unwrap();

        let expected: ArrayRef = Arc::new(Int32Array::from(vec![
            Some(100), // {"a": {"b": {"x": 100}}} -> 100
            None,      // {"a": {"b": {"x": "bar"}}} -> NULL (type mismatch)
            None,      // {"a": {"b": {"y": 200}}} -> NULL (field missing)
        ]));
        assert_eq!(&result, &expected);
        println!("Depth 2 (unshredded) passed");

        // Shredded test data: depth 2 nested shredding
        let shredded_array = create_depth_2_shredded_test_data_working();

        let field = Field::new("result", DataType::Int32, true);
        let path = VariantPath::try_from("a.b.x").unwrap(); // Double nested dot notation!
        let options = GetOptions::new_with_path(path).with_as_type(Some(FieldRef::from(field)));
        let result = variant_get(&shredded_array, options).unwrap();

        let expected: ArrayRef = Arc::new(Int32Array::from(vec![
            Some(100), // {"a": {"b": {"x": 100}}} -> 100 (from deeply nested shredded x)
            None,      // {"a": {"b": {"x": "bar"}}} -> NULL (type mismatch in deep value)
            None,      // {"a": {"b": {"y": 200}}} -> NULL (field missing in deep structure)
        ]));
        assert_eq!(&result, &expected);
        println!("Depth 2 (shredded) passed");
    }

    /// Test that demonstrates what CURRENTLY WORKS
    ///
    /// This shows that nested path functionality does work, but only when the
    /// test data matches what the current implementation expects
    #[test]
    fn test_current_nested_path_functionality() {
        let array = shredded_object_with_x_field_variant_array();

        // Test: Extract the "x" field (single level) - this works
        let single_path = VariantPath::try_from("x").unwrap();
        let field = Field::new("result", DataType::Int32, true);
        let options =
            GetOptions::new_with_path(single_path).with_as_type(Some(FieldRef::from(field)));
        let result = variant_get(&array, options).unwrap();

        println!("Single path 'x' works - result: {:?}", result);

        // Test: Try nested path "a.x" - this is what we need to implement
        let nested_path = VariantPath::try_from("a").unwrap().join("x");
        let field = Field::new("result", DataType::Int32, true);
        let options =
            GetOptions::new_with_path(nested_path).with_as_type(Some(FieldRef::from(field)));
        let result = variant_get(&array, options).unwrap();

        println!("Nested path 'a.x' result: {:?}", result);
    }

    #[test]
    fn test_variant_get_as_variant_from_unshredded_input() {
        let (unshredded, _) = create_variant_get_as_variant_test_data();
        let unshredded_field = VariantArray::try_new(&unshredded).unwrap().field("result");
        assert_variant_field_extraction_returns_unshredded_variant(&unshredded, &unshredded_field);
    }

    #[test]
    fn test_variant_get_as_variant_from_shredded_input() {
        let (unshredded, shredded) = create_variant_get_as_variant_test_data();
        let unshredded_field = VariantArray::try_new(&unshredded).unwrap().field("result");
        assert_variant_field_extraction_returns_unshredded_variant(&shredded, &unshredded_field);
    }

    #[test]
    fn test_variant_get_as_shredded_variant_is_not_yet_supported() {
        let (_, shredded) = create_variant_get_as_variant_test_data();
        // Deriving the request field from the shredded array yields a `VariantType` field whose
        // struct carries a `typed_value` -- a request to shred the output. That is unsupported
        // (https://github.com/apache/arrow-rs/issues/8153) and must error, not silently return a
        // plain binary variant.
        let shredded_field = VariantArray::try_new(&shredded).unwrap().field("result");
        assert!(requested_field_is_shredded(Some(&shredded_field)));

        let options = GetOptions::new_with_path(VariantPath::try_from("field_name").unwrap())
            .with_as_type(Some(FieldRef::from(shredded_field)));
        let err = variant_get(&shredded, options).unwrap_err();
        assert!(
            matches!(err, ArrowError::NotYetImplemented(_)),
            "expected NotYetImplemented, got {err:?}"
        );
    }

    fn create_variant_get_as_variant_test_data() -> (ArrayRef, ArrayRef) {
        let input_json: ArrayRef = Arc::new(StringArray::from(vec![
            Some(r#"{"field_name": {"k": 100000}}"#),
            Some(r#"{"field_name": {"k": "s"}}"#),
        ]));

        let unshredded = ArrayRef::from(json_to_variant(&input_json).unwrap());
        let unshredded_variant = VariantArray::try_new(&unshredded).unwrap();

        let as_type = DataType::Struct(Fields::from(vec![Field::new(
            "field_name",
            DataType::Struct(Fields::from(vec![Field::new("k", DataType::Int32, true)])),
            true,
        )]));
        let shredded = ArrayRef::from(shred_variant(&unshredded_variant, &as_type).unwrap());

        (unshredded, shredded)
    }

    fn assert_variant_field_extraction_returns_unshredded_variant(
        input: &ArrayRef,
        variant_field: &Field,
    ) {
        let options = GetOptions::new_with_path(VariantPath::try_from("field_name").unwrap())
            .with_as_type(Some(FieldRef::from(variant_field.clone())));

        let result = variant_get(input, options).unwrap();
        let result_variant = VariantArray::try_new(&result).unwrap();

        assert!(result_variant.typed_value_column().is_none());
        assert!(result_variant.value_column().is_some());

        let expected_json: ArrayRef = Arc::new(StringArray::from(vec![
            Some(r#"{"k":100000}"#),
            Some(r#"{"k":"s"}"#),
        ]));
        let expected = json_to_variant(&expected_json).unwrap();

        assert_eq!(result_variant.len(), expected.len());
        for i in 0..result_variant.len() {
            assert_eq!(result_variant.is_null(i), expected.is_null(i));
            if !result_variant.is_null(i) {
                assert_eq!(result_variant.value(i), expected.value(i));
            }
        }
    }

    /// Create test data for depth 0 (direct field access)
    /// [{"x": 42}, {"x": "foo"}, {"y": 10}]
    fn create_depth_0_test_data() -> ArrayRef {
        let mut builder = crate::VariantArrayBuilder::new(3);

        // Row 1: {"x": 42}
        {
            let json_str = r#"{"x": 42}"#;
            let string_array: ArrayRef = Arc::new(StringArray::from(vec![json_str]));
            if let Ok(variant_array) = json_to_variant(&string_array) {
                builder.append_variant(variant_array.value(0));
            } else {
                builder.append_null();
            }
        }

        // Row 2: {"x": "foo"}
        {
            let json_str = r#"{"x": "foo"}"#;
            let string_array: ArrayRef = Arc::new(StringArray::from(vec![json_str]));
            if let Ok(variant_array) = json_to_variant(&string_array) {
                builder.append_variant(variant_array.value(0));
            } else {
                builder.append_null();
            }
        }

        // Row 3: {"y": 10} (missing "x" field)
        {
            let json_str = r#"{"y": 10}"#;
            let string_array: ArrayRef = Arc::new(StringArray::from(vec![json_str]));
            if let Ok(variant_array) = json_to_variant(&string_array) {
                builder.append_variant(variant_array.value(0));
            } else {
                builder.append_null();
            }
        }

        ArrayRef::from(builder.build())
    }

    /// Create test data for depth 1 (single nested field)
    /// This represents the exact scenarios from the GitHub issue: "a.x"
    fn create_nested_path_test_data() -> ArrayRef {
        let mut builder = crate::VariantArrayBuilder::new(2);

        // Row 1: {"a": {"x": 55}, "b": 42}
        {
            let json_str = r#"{"a": {"x": 55}, "b": 42}"#;
            let string_array: ArrayRef = Arc::new(StringArray::from(vec![json_str]));
            if let Ok(variant_array) = json_to_variant(&string_array) {
                builder.append_variant(variant_array.value(0));
            } else {
                builder.append_null();
            }
        }

        // Row 2: {"a": {"x": "foo"}, "b": 42}
        {
            let json_str = r#"{"a": {"x": "foo"}, "b": 42}"#;
            let string_array: ArrayRef = Arc::new(StringArray::from(vec![json_str]));
            if let Ok(variant_array) = json_to_variant(&string_array) {
                builder.append_variant(variant_array.value(0));
            } else {
                builder.append_null();
            }
        }

        ArrayRef::from(builder.build())
    }

    /// Create test data for depth 2 (double nested field)
    /// [{"a": {"b": {"x": 100}}}, {"a": {"b": {"x": "bar"}}}, {"a": {"b": {"y": 200}}}]
    fn create_depth_2_test_data() -> ArrayRef {
        let mut builder = crate::VariantArrayBuilder::new(3);

        // Row 1: {"a": {"b": {"x": 100}}}
        {
            let json_str = r#"{"a": {"b": {"x": 100}}}"#;
            let string_array: ArrayRef = Arc::new(StringArray::from(vec![json_str]));
            if let Ok(variant_array) = json_to_variant(&string_array) {
                builder.append_variant(variant_array.value(0));
            } else {
                builder.append_null();
            }
        }

        // Row 2: {"a": {"b": {"x": "bar"}}}
        {
            let json_str = r#"{"a": {"b": {"x": "bar"}}}"#;
            let string_array: ArrayRef = Arc::new(StringArray::from(vec![json_str]));
            if let Ok(variant_array) = json_to_variant(&string_array) {
                builder.append_variant(variant_array.value(0));
            } else {
                builder.append_null();
            }
        }

        // Row 3: {"a": {"b": {"y": 200}}} (missing "x" field)
        {
            let json_str = r#"{"a": {"b": {"y": 200}}}"#;
            let string_array: ArrayRef = Arc::new(StringArray::from(vec![json_str]));
            if let Ok(variant_array) = json_to_variant(&string_array) {
                builder.append_variant(variant_array.value(0));
            } else {
                builder.append_null();
            }
        }

        ArrayRef::from(builder.build())
    }

    /// Create simple shredded test data for depth 0 using a simplified working pattern
    /// Creates 2 rows: [{"x": 42}, {"x": "foo"}] with "x" shredded where possible
    fn create_depth_0_shredded_test_data_simple() -> ArrayRef {
        // Create base metadata using the working pattern
        let (metadata, string_x_value) = {
            let mut builder = parquet_variant::VariantBuilder::new();
            let mut obj = builder.new_object();
            obj.insert("x", Variant::from("foo"));
            obj.finish();
            builder.finish()
        };

        // Metadata array (same for both rows)
        let metadata_array = BinaryViewArray::from_iter_values(std::iter::repeat_n(&metadata, 2));

        // Value array following the 3-step shredding spec:
        // Row 0: {} (x is shredded, no unshredded fields)
        // Row 1: {"x": "foo"} (x is a string, can't be shredded to Int32)
        let empty_object_value = {
            let mut builder = parquet_variant::VariantBuilder::new();
            let obj = builder.new_object();
            obj.finish();
            let (_, value) = builder.finish();
            value
        };

        let value_array = BinaryViewArray::from(vec![
            Some(empty_object_value.as_slice()), // Row 0: {} (x shredded out)
            Some(string_x_value.as_slice()),     // Row 1: {"x": "foo"} (fallback)
        ]);

        // Create the "x" field as a ShreddedVariantFieldArray
        let x_field_typed_value = Int32Array::from(vec![Some(42), None]);

        // For the x field, only typed_value (perfect shredding when possible)
        let x_field_shredded = ShreddedVariantFieldArray::from_parts(
            None,
            Some(Arc::new(x_field_typed_value) as ArrayRef),
            None,
        );

        // Create the main typed_value as a struct containing the "x" field
        let typed_value_fields = Fields::from(vec![Field::new(
            "x",
            x_field_shredded.data_type().clone(),
            true,
        )]);
        let typed_value_struct = StructArray::try_new(
            typed_value_fields,
            vec![ArrayRef::from(x_field_shredded)],
            None,
        )
        .unwrap();

        // Build final VariantArray
        ArrayRef::from(VariantArray::from_parts(
            Arc::new(metadata_array),
            Some(Arc::new(value_array)),
            Some(Arc::new(typed_value_struct)),
            None,
        ))
    }

    /// Create working depth 1 shredded test data based on the existing working pattern
    /// This creates a properly structured shredded variant for "a.x" where:
    /// - Row 0: {"a": {"x": 55}, "b": 42} with a.x shredded into typed_value
    /// - Row 1: {"a": {"x": "foo"}, "b": 42} with a.x fallback to value field due to type mismatch
    fn create_depth_1_shredded_test_data_working() -> ArrayRef {
        // Create metadata following the working pattern from shredded_object_with_x_field_variant_array
        let (metadata, _) = {
            // Create nested structure: {"a": {"x": 55}, "b": 42}
            let mut builder = parquet_variant::VariantBuilder::new();
            let mut obj = builder.new_object();

            // Create the nested "a" object
            let mut a_obj = obj.new_object("a");
            a_obj.insert("x", Variant::Int32(55));
            a_obj.finish();

            obj.insert("b", Variant::Int32(42));
            obj.finish();
            builder.finish()
        };

        let metadata_array = BinaryViewArray::from_iter_values(std::iter::repeat_n(&metadata, 2));

        // Create value arrays for the fallback case
        // Following the spec: if field cannot be shredded, it stays in value
        let empty_object_value = {
            let mut builder = parquet_variant::VariantBuilder::new();
            let obj = builder.new_object();
            obj.finish();
            let (_, value) = builder.finish();
            value
        };

        // Row 1 fallback: use the working pattern from the existing shredded test
        // This avoids metadata issues by using the simple fallback approach
        let row1_fallback = {
            let mut builder = parquet_variant::VariantBuilder::new();
            let mut obj = builder.new_object();
            obj.insert("fallback", Variant::from("data"));
            obj.finish();
            let (_, value) = builder.finish();
            value
        };

        let value_array = BinaryViewArray::from(vec![
            Some(empty_object_value.as_slice()), // Row 0: {} (everything shredded except b in unshredded fields)
            Some(row1_fallback.as_slice()), // Row 1: {"a": {"x": "foo"}, "b": 42} (a.x can't be shredded)
        ]);

        // Create the nested shredded structure
        // Level 2: x field (the deepest level)
        let x_typed_value = Int32Array::from(vec![Some(55), None]);
        let x_field_shredded = ShreddedVariantFieldArray::from_parts(
            None,
            Some(Arc::new(x_typed_value) as ArrayRef),
            None,
        );

        // Level 1: a field containing x field + value field for fallbacks
        // The "a" field needs both typed_value (for shredded x) and value (for fallback cases)

        // Create the value field for "a" (for cases where a.x can't be shredded)
        let a_value_data = {
            let mut builder = parquet_variant::VariantBuilder::new();
            let obj = builder.new_object();
            obj.finish();
            let (_, value) = builder.finish();
            value
        };
        let a_value_array = BinaryViewArray::from(vec![
            None,                          // Row 0: x is shredded, so no value fallback needed
            Some(a_value_data.as_slice()), // Row 1: fallback for a.x="foo" (but logic will check typed_value first)
        ]);

        let a_inner_fields = Fields::from(vec![Field::new(
            "x",
            x_field_shredded.data_type().clone(),
            true,
        )]);
        let a_inner_typed_value = Arc::new(
            StructArray::try_new(a_inner_fields, vec![ArrayRef::from(x_field_shredded)], None)
                .unwrap(),
        ) as ArrayRef;
        let a_field_shredded = ShreddedVariantFieldArray::from_parts(
            Some(Arc::new(a_value_array)),
            Some(a_inner_typed_value),
            None,
        );

        // Level 0: main typed_value struct containing a field
        let typed_value_fields = Fields::from(vec![Field::new(
            "a",
            a_field_shredded.data_type().clone(),
            true,
        )]);
        let typed_value_struct = StructArray::try_new(
            typed_value_fields,
            vec![ArrayRef::from(a_field_shredded)],
            None,
        )
        .unwrap();

        // Build final VariantArray
        ArrayRef::from(VariantArray::from_parts(
            Arc::new(metadata_array),
            Some(Arc::new(value_array)),
            Some(Arc::new(typed_value_struct)),
            None,
        ))
    }

    /// Create working depth 2 shredded test data for "a.b.x" paths
    /// This creates a 3-level nested shredded structure where:
    /// - Row 0: {"a": {"b": {"x": 100}}} with a.b.x shredded into typed_value
    /// - Row 1: {"a": {"b": {"x": "bar"}}} with type mismatch fallback
    /// - Row 2: {"a": {"b": {"y": 200}}} with missing field fallback
    fn create_depth_2_shredded_test_data_working() -> ArrayRef {
        // Create metadata following the working pattern
        let (metadata, _) = {
            // Create deeply nested structure: {"a": {"b": {"x": 100}}}
            let mut builder = parquet_variant::VariantBuilder::new();
            let mut obj = builder.new_object();

            // Create the nested "a.b" structure
            let mut a_obj = obj.new_object("a");
            let mut b_obj = a_obj.new_object("b");
            b_obj.insert("x", Variant::Int32(100));
            b_obj.finish();
            a_obj.finish();

            obj.finish();
            builder.finish()
        };

        let metadata_array = BinaryViewArray::from_iter_values(std::iter::repeat_n(&metadata, 3));

        // Create value arrays for fallback cases
        let empty_object_value = {
            let mut builder = parquet_variant::VariantBuilder::new();
            let obj = builder.new_object();
            obj.finish();
            let (_, value) = builder.finish();
            value
        };

        // Simple fallback values - avoiding complex nested metadata
        let value_array = BinaryViewArray::from(vec![
            Some(empty_object_value.as_slice()), // Row 0: fully shredded
            Some(empty_object_value.as_slice()), // Row 1: fallback (simplified)
            Some(empty_object_value.as_slice()), // Row 2: fallback (simplified)
        ]);

        // Create the deeply nested shredded structure: a.b.x

        // Level 3: x field (deepest level)
        let x_typed_value = Int32Array::from(vec![Some(100), None, None]);
        let x_field_shredded = ShreddedVariantFieldArray::from_parts(
            None,
            Some(Arc::new(x_typed_value) as ArrayRef),
            None,
        );

        // Level 2: b field containing x field + value field
        let b_value_data = {
            let mut builder = parquet_variant::VariantBuilder::new();
            let obj = builder.new_object();
            obj.finish();
            let (_, value) = builder.finish();
            value
        };
        let b_value_array = BinaryViewArray::from(vec![
            None,                          // Row 0: x is shredded
            Some(b_value_data.as_slice()), // Row 1: fallback for b.x="bar"
            Some(b_value_data.as_slice()), // Row 2: fallback for b.y=200
        ]);

        let b_inner_fields = Fields::from(vec![Field::new(
            "x",
            x_field_shredded.data_type().clone(),
            true,
        )]);
        let b_inner_typed_value = Arc::new(
            StructArray::try_new(b_inner_fields, vec![ArrayRef::from(x_field_shredded)], None)
                .unwrap(),
        ) as ArrayRef;
        let b_field_shredded = ShreddedVariantFieldArray::from_parts(
            Some(Arc::new(b_value_array)),
            Some(b_inner_typed_value),
            None,
        );

        // Level 1: a field containing b field + value field
        let a_value_data = {
            let mut builder = parquet_variant::VariantBuilder::new();
            let obj = builder.new_object();
            obj.finish();
            let (_, value) = builder.finish();
            value
        };
        let a_value_array = BinaryViewArray::from(vec![
            None,                          // Row 0: b is shredded
            Some(a_value_data.as_slice()), // Row 1: fallback for a.b.*
            Some(a_value_data.as_slice()), // Row 2: fallback for a.b.*
        ]);

        let a_inner_fields = Fields::from(vec![Field::new(
            "b",
            b_field_shredded.data_type().clone(),
            true,
        )]);
        let a_inner_typed_value = Arc::new(
            StructArray::try_new(a_inner_fields, vec![ArrayRef::from(b_field_shredded)], None)
                .unwrap(),
        ) as ArrayRef;
        let a_field_shredded = ShreddedVariantFieldArray::from_parts(
            Some(Arc::new(a_value_array)),
            Some(a_inner_typed_value),
            None,
        );

        // Level 0: main typed_value struct containing a field
        let typed_value_fields = Fields::from(vec![Field::new(
            "a",
            a_field_shredded.data_type().clone(),
            true,
        )]);
        let typed_value_struct = StructArray::try_new(
            typed_value_fields,
            vec![ArrayRef::from(a_field_shredded)],
            None,
        )
        .unwrap();

        // Build final VariantArray
        ArrayRef::from(VariantArray::from_parts(
            Arc::new(metadata_array),
            Some(Arc::new(value_array)),
            Some(Arc::new(typed_value_struct)),
            None,
        ))
    }

    #[test]
    fn test_field_path_non_struct_returns_missing_path_step() {
        // Use the existing simple test data that has Int32 as typed_value
        let variant_array = perfectly_shredded_int32_variant_array();

        for safe in [true, false] {
            let options = GetOptions {
                path: VariantPath::try_from("nonexistent_field").unwrap(),
                as_type: Some(Arc::new(Field::new("result", DataType::Int32, true))),
                cast_options: CastOptions {
                    safe,
                    ..Default::default()
                },
            };

            let result_array = variant_get(&variant_array, options).unwrap();
            assert_eq!(result_array.len(), 3);
            assert!(result_array.is_null(0));
            assert!(result_array.is_null(1));
            assert!(result_array.is_null(2));
        }
    }

    #[test]
    fn test_strict_cast_options_index_on_non_list_returns_null() {
        use arrow::compute::CastOptions;
        use arrow::datatypes::{DataType, Field};
        use parquet_variant::VariantPath;
        use std::sync::Arc;

        // Use existing test data that has Int32 typed_value at the top level.
        let variant_array = perfectly_shredded_int32_variant_array();
        let options = GetOptions {
            path: VariantPath::from(0),
            as_type: Some(Arc::new(Field::new("result", DataType::Int32, true))),
            cast_options: CastOptions {
                safe: false,
                ..Default::default()
            },
        };

        let variant_array_ref: Arc<dyn Array> = variant_array.clone();
        let result = variant_get(&variant_array_ref, options).unwrap();

        assert_eq!(result.len(), 3);
        assert!(result.is_null(0));
        assert!(result.is_null(1));
        assert!(result.is_null(2));
    }

    #[test]
    fn test_error_message_boolean_type_display() {
        let mut builder = VariantArrayBuilder::new(1);
        builder.append_variant(Variant::from("abcd"));
        let variant_array: ArrayRef = ArrayRef::from(builder.build());

        // Request Boolean with strict casting to force an error
        let options = GetOptions {
            path: VariantPath::default(),
            as_type: Some(Arc::new(Field::new("result", DataType::Boolean, true))),
            cast_options: CastOptions {
                safe: false,
                ..Default::default()
            },
        };

        let err = variant_get(&variant_array, options).unwrap_err();
        let msg = err.to_string();
        assert!(msg.contains("Failed to extract primitive of type Boolean"));
    }

    #[test]
    fn test_error_message_numeric_type_display() {
        let mut builder = VariantArrayBuilder::new(1);
        builder.append_variant(Variant::from("abcd"));
        let variant_array: ArrayRef = ArrayRef::from(builder.build());

        // Request Float32 with strict casting to force an error
        let options = GetOptions {
            path: VariantPath::default(),
            as_type: Some(Arc::new(Field::new("result", DataType::Float32, true))),
            cast_options: CastOptions {
                safe: false,
                ..Default::default()
            },
        };

        let err = variant_get(&variant_array, options).unwrap_err();
        let msg = err.to_string();
        assert!(msg.contains("Failed to extract primitive of type Float32"));
    }

    #[test]
    fn test_error_message_temporal_type_display() {
        let mut builder = VariantArrayBuilder::new(1);
        builder.append_variant(Variant::BooleanFalse);
        let variant_array: ArrayRef = ArrayRef::from(builder.build());

        // Request Timestamp with strict casting to force an error
        let options = GetOptions {
            path: VariantPath::default(),
            as_type: Some(Arc::new(Field::new(
                "result",
                DataType::Timestamp(TimeUnit::Nanosecond, None),
                true,
            ))),
            cast_options: CastOptions {
                safe: false,
                ..Default::default()
            },
        };

        let err = variant_get(&variant_array, options).unwrap_err();
        let msg = err.to_string();
        assert!(msg.contains("Failed to extract primitive of type Timestamp(ns)"));
    }

    #[test]
    fn test_null_buffer_union_for_shredded_paths() {
        // Test that null buffers are properly unioned when traversing shredded paths
        // This test verifies scovich's null buffer union requirement

        // Create a depth-1 shredded variant array where:
        // - The top-level variant array has some nulls
        // - The nested typed_value also has some nulls
        // - The result should be the union of both null buffers

        let variant_array = create_depth_1_shredded_test_data_working();

        // Get the field "x" which should union nulls from:
        // 1. The top-level variant array nulls
        // 2. The "a" field's typed_value nulls
        // 3. The "x" field's typed_value nulls
        let options = GetOptions {
            path: VariantPath::try_from("a.x").unwrap(),
            as_type: Some(Arc::new(Field::new("result", DataType::Int32, true))),
            cast_options: CastOptions::default(),
        };

        let result = variant_get(&variant_array, options).unwrap();

        // Verify the result length matches input
        assert_eq!(result.len(), variant_array.len());

        // The null pattern should reflect the union of all ancestor nulls
        // Row 0: Should have valid data (path exists and is shredded as Int32)
        // Row 1: Should be null (due to type mismatch - "foo" can't cast to Int32)
        assert!(!result.is_null(0), "Row 0 should have valid Int32 data");
        assert!(
            result.is_null(1),
            "Row 1 should be null due to type casting failure"
        );

        // Verify the actual values
        let int32_result = result.as_any().downcast_ref::<Int32Array>().unwrap();
        assert_eq!(int32_result.value(0), 55); // The valid Int32 value
    }

    #[test]
    fn test_struct_null_mask_union_from_children() {
        // Test that struct null masks properly union nulls from children field extractions
        // This verifies scovich's concern about incomplete null masks in struct construction

        // Create test data where some fields will fail type casting
        let json_strings = vec![
            r#"{"a": 42, "b": "hello"}"#, // Row 0: a=42 (castable to int), b="hello" (not castable to int)
            r#"{"a": "world", "b": 100}"#, // Row 1: a="world" (not castable to int), b=100 (castable to int)
            r#"{"a": 55, "b": 77}"#,       // Row 2: a=55 (castable to int), b=77 (castable to int)
        ];

        let string_array: Arc<dyn arrow::array::Array> = Arc::new(StringArray::from(json_strings));
        let variant_array = json_to_variant(&string_array).unwrap();

        // Request extraction as a struct with both fields as Int32
        // This should create child arrays where some fields are null due to casting failures
        let struct_fields = Fields::from(vec![
            Field::new("a", DataType::Int32, true),
            Field::new("b", DataType::Int32, true),
        ]);
        let struct_type = DataType::Struct(struct_fields);

        let options = GetOptions {
            path: VariantPath::default(), // Extract the whole object as struct
            as_type: Some(Arc::new(Field::new("result", struct_type, true))),
            cast_options: CastOptions::default(),
        };

        let variant_array_ref = ArrayRef::from(variant_array);
        let result = variant_get(&variant_array_ref, options).unwrap();

        // Verify the result is a StructArray
        let struct_result = result.as_struct();
        assert_eq!(struct_result.len(), 3);

        // Get the individual field arrays
        let field_a = struct_result
            .column(0)
            .as_any()
            .downcast_ref::<Int32Array>()
            .unwrap();
        let field_b = struct_result
            .column(1)
            .as_any()
            .downcast_ref::<Int32Array>()
            .unwrap();

        // Verify field values and nulls
        // Row 0: a=42 (valid), b=null (casting failure)
        assert!(!field_a.is_null(0));
        assert_eq!(field_a.value(0), 42);
        assert!(field_b.is_null(0)); // "hello" can't cast to int

        // Row 1: a=null (casting failure), b=100 (valid)
        assert!(field_a.is_null(1)); // "world" can't cast to int
        assert!(!field_b.is_null(1));
        assert_eq!(field_b.value(1), 100);

        // Row 2: a=55 (valid), b=77 (valid)
        assert!(!field_a.is_null(2));
        assert_eq!(field_a.value(2), 55);
        assert!(!field_b.is_null(2));
        assert_eq!(field_b.value(2), 77);

        // Verify the struct-level null mask properly unions child nulls
        // The struct should NOT be null in any row because each row has at least one valid field
        // (This tests that we're not incorrectly making the entire struct null when children fail)
        assert!(!struct_result.is_null(0)); // Has valid field 'a'
        assert!(!struct_result.is_null(1)); // Has valid field 'b'
        assert!(!struct_result.is_null(2)); // Has both valid fields
    }

    #[test]
    fn test_field_nullability_preservation() {
        // Test that field nullability from GetOptions.as_type is preserved in the result

        let json_strings = vec![
            r#"{"x": 42}"#,                  // Row 0: Valid int that should convert to Int32
            r#"{"x": "not_a_number"}"#,      // Row 1: String that can't cast to Int32
            r#"{"x": null}"#,                // Row 2: Explicit null value
            r#"{"x": "hello"}"#,             // Row 3: Another string (wrong type)
            r#"{"y": 100}"#,                 // Row 4: Missing "x" field (SQL NULL case)
            r#"{"x": 127}"#, // Row 5: Small int (could be Int8, widening cast candidate)
            r#"{"x": 32767}"#, // Row 6: Medium int (could be Int16, widening cast candidate)
            r#"{"x": 2147483647}"#, // Row 7: Max Int32 value (fits in Int32)
            r#"{"x": 9223372036854775807}"#, // Row 8: Large Int64 value (cannot convert to Int32)
        ];

        let string_array: Arc<dyn arrow::array::Array> = Arc::new(StringArray::from(json_strings));
        let variant_array = json_to_variant(&string_array).unwrap();

        // Test 1: nullable field (should allow nulls from cast failures)
        let nullable_field = Arc::new(Field::new("result", DataType::Int32, true));
        let options_nullable = GetOptions {
            path: VariantPath::try_from("x").unwrap(),
            as_type: Some(nullable_field.clone()),
            cast_options: CastOptions::default(),
        };

        let variant_array_ref = ArrayRef::from(variant_array);
        let result_nullable = variant_get(&variant_array_ref, options_nullable).unwrap();

        // Verify we get an Int32Array with nulls for cast failures
        let int32_result = result_nullable
            .as_any()
            .downcast_ref::<Int32Array>()
            .unwrap();
        assert_eq!(int32_result.len(), 9);

        // Row 0: 42 converts successfully to Int32
        assert!(!int32_result.is_null(0));
        assert_eq!(int32_result.value(0), 42);

        // Row 1: "not_a_number" fails to convert -> NULL
        assert!(int32_result.is_null(1));

        // Row 2: explicit null value -> NULL
        assert!(int32_result.is_null(2));

        // Row 3: "hello" (wrong type) fails to convert -> NULL
        assert!(int32_result.is_null(3));

        // Row 4: missing "x" field (SQL NULL case) -> NULL
        assert!(int32_result.is_null(4));

        // Row 5: 127 (small int, potential Int8 -> Int32 widening)
        // Current behavior: JSON parses to Int8, should convert to Int32
        assert!(!int32_result.is_null(5));
        assert_eq!(int32_result.value(5), 127);

        // Row 6: 32767 (medium int, potential Int16 -> Int32 widening)
        // Current behavior: JSON parses to Int16, should convert to Int32
        assert!(!int32_result.is_null(6));
        assert_eq!(int32_result.value(6), 32767);

        // Row 7: 2147483647 (max Int32, fits exactly)
        // Current behavior: Should convert successfully
        assert!(!int32_result.is_null(7));
        assert_eq!(int32_result.value(7), 2147483647);

        // Row 8: 9223372036854775807 (large Int64, cannot fit in Int32)
        // Current behavior: Should fail conversion -> NULL
        assert!(int32_result.is_null(8));

        // Test 2: non-nullable field (behavior should be the same with safe casting)
        let non_nullable_field = Arc::new(Field::new("result", DataType::Int32, false));
        let options_non_nullable = GetOptions {
            path: VariantPath::try_from("x").unwrap(),
            as_type: Some(non_nullable_field.clone()),
            cast_options: CastOptions::default(), // safe=true by default
        };

        // Create variant array again since we moved it
        let variant_array_2 = json_to_variant(&string_array).unwrap();
        let variant_array_ref_2 = ArrayRef::from(variant_array_2);
        let result_non_nullable = variant_get(&variant_array_ref_2, options_non_nullable).unwrap();
        let int32_result_2 = result_non_nullable
            .as_any()
            .downcast_ref::<Int32Array>()
            .unwrap();

        // Even with a non-nullable field, safe casting should still produce nulls for failures
        assert_eq!(int32_result_2.len(), 9);

        // Row 0: 42 converts successfully to Int32
        assert!(!int32_result_2.is_null(0));
        assert_eq!(int32_result_2.value(0), 42);

        // Rows 1-4: All should be null due to safe casting behavior
        // (non-nullable field specification doesn't override safe casting behavior)
        assert!(int32_result_2.is_null(1)); // "not_a_number"
        assert!(int32_result_2.is_null(2)); // explicit null
        assert!(int32_result_2.is_null(3)); // "hello"
        assert!(int32_result_2.is_null(4)); // missing field

        // Rows 5-7: These should also convert successfully (numeric widening/fitting)
        assert!(!int32_result_2.is_null(5)); // 127 (Int8 -> Int32)
        assert_eq!(int32_result_2.value(5), 127);
        assert!(!int32_result_2.is_null(6)); // 32767 (Int16 -> Int32)
        assert_eq!(int32_result_2.value(6), 32767);
        assert!(!int32_result_2.is_null(7)); // 2147483647 (fits in Int32)
        assert_eq!(int32_result_2.value(7), 2147483647);

        // Row 8: Large Int64 should fail conversion -> NULL
        assert!(int32_result_2.is_null(8)); // 9223372036854775807 (too large for Int32)
    }

    #[test]
    fn test_struct_extraction_subset_superset_schema_perfectly_shredded() {
        // Create variant with diverse null patterns and empty objects
        let variant_array = create_comprehensive_shredded_variant();

        // Request struct with fields "a", "b", "d" (skip existing "c", add missing "d")
        let struct_fields = Fields::from(vec![
            Field::new("a", DataType::Int32, true),
            Field::new("b", DataType::Int32, true),
            Field::new("d", DataType::Int32, true),
        ]);
        let struct_type = DataType::Struct(struct_fields);

        let options = GetOptions {
            path: VariantPath::default(),
            as_type: Some(Arc::new(Field::new("result", struct_type, true))),
            cast_options: CastOptions::default(),
        };

        let result = variant_get(&variant_array, options).unwrap();

        // Verify the result is a StructArray with 3 fields and 5 rows
        let struct_result = result.as_any().downcast_ref::<StructArray>().unwrap();
        assert_eq!(struct_result.len(), 5);
        assert_eq!(struct_result.num_columns(), 3);

        let field_a = struct_result
            .column(0)
            .as_any()
            .downcast_ref::<Int32Array>()
            .unwrap();
        let field_b = struct_result
            .column(1)
            .as_any()
            .downcast_ref::<Int32Array>()
            .unwrap();
        let field_d = struct_result
            .column(2)
            .as_any()
            .downcast_ref::<Int32Array>()
            .unwrap();

        // Row 0: Normal values {"a": 1, "b": 2, "c": 3} → {a: 1, b: 2, d: NULL}
        assert!(!struct_result.is_null(0));
        assert_eq!(field_a.value(0), 1);
        assert_eq!(field_b.value(0), 2);
        assert!(field_d.is_null(0)); // Missing field "d"

        // Row 1: Top-level NULL → struct-level NULL
        assert!(struct_result.is_null(1));

        // Row 2: Field "a" missing → {a: NULL, b: 2, d: NULL}
        assert!(!struct_result.is_null(2));
        assert!(field_a.is_null(2)); // Missing field "a"
        assert_eq!(field_b.value(2), 2);
        assert!(field_d.is_null(2)); // Missing field "d"

        // Row 3: Field "b" missing → {a: 1, b: NULL, d: NULL}
        assert!(!struct_result.is_null(3));
        assert_eq!(field_a.value(3), 1);
        assert!(field_b.is_null(3)); // Missing field "b"
        assert!(field_d.is_null(3)); // Missing field "d"

        // Row 4: Empty object {} → {a: NULL, b: NULL, d: NULL}
        assert!(!struct_result.is_null(4));
        assert!(field_a.is_null(4)); // Empty object
        assert!(field_b.is_null(4)); // Empty object
        assert!(field_d.is_null(4)); // Missing field "d"
    }

    #[test]
    fn test_nested_struct_extraction_perfectly_shredded() {
        // Create nested variant with diverse null patterns
        let variant_array = create_comprehensive_nested_shredded_variant();
        println!("variant_array: {variant_array:?}");

        // Request 3-level nested struct type {"outer": {"inner": INT}}
        let inner_field = Field::new("inner", DataType::Int32, true);
        let inner_type = DataType::Struct(Fields::from(vec![inner_field]));
        let outer_field = Field::new("outer", inner_type, true);
        let result_type = DataType::Struct(Fields::from(vec![outer_field]));

        let options = GetOptions {
            path: VariantPath::default(),
            as_type: Some(Arc::new(Field::new("result", result_type, true))),
            cast_options: CastOptions::default(),
        };

        let result = variant_get(&variant_array, options).unwrap();
        println!("result: {result:?}");

        // Verify the result is a StructArray with "outer" field and 4 rows
        let outer_struct = result.as_any().downcast_ref::<StructArray>().unwrap();
        assert_eq!(outer_struct.len(), 4);
        assert_eq!(outer_struct.num_columns(), 1);

        // Get the "inner" struct column
        let inner_struct = outer_struct
            .column(0)
            .as_any()
            .downcast_ref::<StructArray>()
            .unwrap();
        assert_eq!(inner_struct.num_columns(), 1);

        // Get the "leaf" field (Int32 values)
        let leaf_field = inner_struct
            .column(0)
            .as_any()
            .downcast_ref::<Int32Array>()
            .unwrap();

        // Row 0: Normal nested {"outer": {"inner": {"leaf": 42}}}
        assert!(!outer_struct.is_null(0));
        assert!(!inner_struct.is_null(0));
        assert_eq!(leaf_field.value(0), 42);

        // Row 1: "inner" field missing → {outer: {inner: NULL}}
        assert!(!outer_struct.is_null(1));
        assert!(!inner_struct.is_null(1)); // outer exists, inner exists but leaf is NULL
        assert!(leaf_field.is_null(1)); // leaf field is NULL

        // Row 2: "outer" field missing → {outer: NULL}
        assert!(!outer_struct.is_null(2));
        assert!(inner_struct.is_null(2)); // outer field is NULL

        // Row 3: Top-level NULL → struct-level NULL
        assert!(outer_struct.is_null(3));
    }

    #[test]
    fn test_path_based_null_masks_one_step() {
        // Create nested variant with diverse null patterns
        let variant_array = create_comprehensive_nested_shredded_variant();

        // Extract "outer" field using path-based variant_get
        let path = VariantPath::try_from("outer").unwrap();
        let inner_field = Field::new("inner", DataType::Int32, true);
        let result_type = DataType::Struct(Fields::from(vec![inner_field]));

        let options = GetOptions {
            path,
            as_type: Some(Arc::new(Field::new("result", result_type, true))),
            cast_options: CastOptions::default(),
        };

        let result = variant_get(&variant_array, options).unwrap();

        // Verify the result is a StructArray with "inner" field and 4 rows
        let outer_result = result.as_any().downcast_ref::<StructArray>().unwrap();
        assert_eq!(outer_result.len(), 4);
        assert_eq!(outer_result.num_columns(), 1);

        // Get the "inner" field (Int32 values)
        let inner_field = outer_result
            .column(0)
            .as_any()
            .downcast_ref::<Int32Array>()
            .unwrap();

        // Row 0: Normal nested {"outer": {"inner": 42}} → {"inner": 42}
        assert!(!outer_result.is_null(0));
        assert_eq!(inner_field.value(0), 42);

        // Row 1: Inner field null {"outer": {"inner": null}} → {"inner": null}
        assert!(!outer_result.is_null(1));
        assert!(inner_field.is_null(1));

        // Row 2: Outer field null {"outer": null} → null (entire struct is null)
        assert!(outer_result.is_null(2));

        // Row 3: Top-level null → null (entire struct is null)
        assert!(outer_result.is_null(3));
    }

    #[test]
    fn test_path_based_null_masks_two_steps() {
        // Create nested variant with diverse null patterns
        let variant_array = create_comprehensive_nested_shredded_variant();

        // Extract "outer.inner" field using path-based variant_get
        let path = VariantPath::try_from("outer").unwrap().join("inner");

        let options = GetOptions {
            path,
            as_type: Some(Arc::new(Field::new("result", DataType::Int32, true))),
            cast_options: CastOptions::default(),
        };

        let result = variant_get(&variant_array, options).unwrap();

        // Verify the result is an Int32Array with 4 rows
        let int_result = result.as_any().downcast_ref::<Int32Array>().unwrap();
        assert_eq!(int_result.len(), 4);

        // Row 0: Normal nested {"outer": {"inner": 42}} → 42
        assert!(!int_result.is_null(0));
        assert_eq!(int_result.value(0), 42);

        // Row 1: Inner field null {"outer": {"inner": null}} → null
        assert!(int_result.is_null(1));

        // Row 2: Outer field null {"outer": null} → null (path traversal fails)
        assert!(int_result.is_null(2));

        // Row 3: Top-level null → null (path traversal fails)
        assert!(int_result.is_null(3));
    }

    #[test]
    fn test_struct_extraction_mixed_and_unshredded() {
        // Create a partially shredded variant (x shredded, y not)
        let variant_array = create_mixed_and_unshredded_variant();

        // Request struct with both shredded and unshredded fields
        let struct_fields = Fields::from(vec![
            Field::new("x", DataType::Int32, true),
            Field::new("y", DataType::Int32, true),
        ]);
        let struct_type = DataType::Struct(struct_fields);

        let options = GetOptions {
            path: VariantPath::default(),
            as_type: Some(Arc::new(Field::new("result", struct_type, true))),
            cast_options: CastOptions::default(),
        };

        let result = variant_get(&variant_array, options).unwrap();

        // Verify the mixed shredding works (should succeed with current implementation)
        let struct_result = result.as_any().downcast_ref::<StructArray>().unwrap();
        assert_eq!(struct_result.len(), 4);
        assert_eq!(struct_result.num_columns(), 2);

        let field_x = struct_result
            .column(0)
            .as_any()
            .downcast_ref::<Int32Array>()
            .unwrap();
        let field_y = struct_result
            .column(1)
            .as_any()
            .downcast_ref::<Int32Array>()
            .unwrap();

        // Row 0: {"x": 1, "y": 42} - x from shredded, y from value field
        assert_eq!(field_x.value(0), 1);
        assert_eq!(field_y.value(0), 42);

        // Row 1: {"x": 2} - x from shredded, y missing (perfect shredding)
        assert_eq!(field_x.value(1), 2);
        assert!(field_y.is_null(1));

        // Row 2: {"x": 3, "y": null} - x from shredded, y explicitly null in value
        assert_eq!(field_x.value(2), 3);
        assert!(field_y.is_null(2));

        // Row 3: top-level null - entire struct row should be null
        assert!(struct_result.is_null(3));
    }

    #[test]
    fn test_struct_row_builder_handles_unshredded_nested_structs() {
        // Create completely unshredded JSON variant (no typed_value at all)
        let json_strings = vec![
            r#"{"outer": {"inner": 42}}"#,
            r#"{"outer": {"inner": 100}}"#,
        ];
        let string_array: Arc<dyn Array> = Arc::new(StringArray::from(json_strings));
        let variant_array = json_to_variant(&string_array).unwrap();

        // Request nested struct
        let inner_fields = Fields::from(vec![Field::new("inner", DataType::Int32, true)]);
        let inner_struct_type = DataType::Struct(inner_fields);
        let outer_fields = Fields::from(vec![Field::new("outer", inner_struct_type, true)]);
        let outer_struct_type = DataType::Struct(outer_fields);

        let options = GetOptions {
            path: VariantPath::default(),
            as_type: Some(Arc::new(Field::new("result", outer_struct_type, true))),
            cast_options: CastOptions::default(),
        };

        let variant_array_ref = ArrayRef::from(variant_array);
        let result = variant_get(&variant_array_ref, options).unwrap();

        let outer_struct = result.as_struct();
        assert_eq!(outer_struct.len(), 2);
        assert_eq!(outer_struct.num_columns(), 1);

        let inner_struct = outer_struct.column(0).as_struct();
        assert_eq!(inner_struct.num_columns(), 1);

        let inner_values = inner_struct
            .column(0)
            .as_any()
            .downcast_ref::<Int32Array>()
            .unwrap();
        assert_eq!(inner_values.value(0), 42);
        assert_eq!(inner_values.value(1), 100);
    }

    #[test]
    fn test_unshredded_struct_safe_cast_non_object_rows_are_null() {
        let json_strings = vec![r#"{"a": 1, "b": 2}"#, "123", "{}"];
        let string_array: Arc<dyn Array> = Arc::new(StringArray::from(json_strings));
        let variant_array_ref = ArrayRef::from(json_to_variant(&string_array).unwrap());

        let struct_fields = Fields::from(vec![
            Field::new("a", DataType::Int32, true),
            Field::new("b", DataType::Int32, true),
        ]);
        let options = GetOptions {
            path: VariantPath::default(),
            as_type: Some(Arc::new(Field::new(
                "result",
                DataType::Struct(struct_fields),
                true,
            ))),
            cast_options: CastOptions::default(),
        };

        let result = variant_get(&variant_array_ref, options).unwrap();
        let struct_result = result.as_struct();
        let field_a = struct_result
            .column(0)
            .as_primitive::<arrow::datatypes::Int32Type>();
        let field_b = struct_result
            .column(1)
            .as_primitive::<arrow::datatypes::Int32Type>();

        // Row 0 is an object, so the struct row is valid with extracted fields.
        assert!(!struct_result.is_null(0));
        assert_eq!(field_a.value(0), 1);
        assert_eq!(field_b.value(0), 2);

        // Row 1 is a scalar, so safe struct cast should produce a NULL struct row.
        assert!(struct_result.is_null(1));
        assert!(field_a.is_null(1));
        assert!(field_b.is_null(1));

        // Row 2 is an empty object, so the struct row is valid with missing fields as NULL.
        assert!(!struct_result.is_null(2));
        assert!(field_a.is_null(2));
        assert!(field_b.is_null(2));
    }

    #[test]
    fn test_unshredded_struct_strict_cast_non_object_errors() {
        let json_strings = vec![r#"{"a": 1, "b": 2}"#, "123"];
        let string_array: Arc<dyn Array> = Arc::new(StringArray::from(json_strings));
        let variant_array_ref = ArrayRef::from(json_to_variant(&string_array).unwrap());

        let struct_fields = Fields::from(vec![
            Field::new("a", DataType::Int32, true),
            Field::new("b", DataType::Int32, true),
        ]);
        let options = GetOptions {
            path: VariantPath::default(),
            as_type: Some(Arc::new(Field::new(
                "result",
                DataType::Struct(struct_fields),
                true,
            ))),
            cast_options: CastOptions {
                safe: false,
                ..Default::default()
            },
        };

        let err = variant_get(&variant_array_ref, options).unwrap_err();
        assert!(
            err.to_string()
                .contains("Failed to extract struct from variant")
        );
    }

    /// Create comprehensive shredded variant with diverse null patterns and empty objects
    /// Rows: normal values, top-level null, missing field a, missing field b, empty object
    fn create_comprehensive_shredded_variant() -> ArrayRef {
        let (metadata, _) = {
            let mut builder = parquet_variant::VariantBuilder::new();
            let obj = builder.new_object();
            obj.finish();
            builder.finish()
        };

        // Create null buffer for top-level nulls
        let nulls = NullBuffer::from(vec![
            true,  // row 0: normal values
            false, // row 1: top-level null
            true,  // row 2: missing field a
            true,  // row 3: missing field b
            true,  // row 4: empty object
        ]);

        let metadata_array = BinaryViewArray::from_iter_values(std::iter::repeat_n(&metadata, 5));

        // Create shredded fields with different null patterns
        // Field "a": present in rows 0,3 (missing in rows 1,2,4)
        let a_field_typed_value = Int32Array::from(vec![Some(1), None, None, Some(1), None]);
        let a_field_shredded = ShreddedVariantFieldArray::from_parts(
            None,
            Some(Arc::new(a_field_typed_value) as ArrayRef),
            None,
        );

        // Field "b": present in rows 0,2 (missing in rows 1,3,4)
        let b_field_typed_value = Int32Array::from(vec![Some(2), None, Some(2), None, None]);
        let b_field_shredded = ShreddedVariantFieldArray::from_parts(
            None,
            Some(Arc::new(b_field_typed_value) as ArrayRef),
            None,
        );

        // Field "c": present in row 0 only (missing in all other rows)
        let c_field_typed_value = Int32Array::from(vec![Some(3), None, None, None, None]);
        let c_field_shredded = ShreddedVariantFieldArray::from_parts(
            None,
            Some(Arc::new(c_field_typed_value) as ArrayRef),
            None,
        );

        // Create main typed_value struct
        let typed_value_fields = Fields::from(vec![
            Field::new("a", a_field_shredded.data_type().clone(), true),
            Field::new("b", b_field_shredded.data_type().clone(), true),
            Field::new("c", c_field_shredded.data_type().clone(), true),
        ]);
        let typed_value_struct = StructArray::try_new(
            typed_value_fields,
            vec![
                ArrayRef::from(a_field_shredded),
                ArrayRef::from(b_field_shredded),
                ArrayRef::from(c_field_shredded),
            ],
            None,
        )
        .unwrap();

        // Build final VariantArray with top-level nulls
        ArrayRef::from(VariantArray::from_parts(
            Arc::new(metadata_array),
            None,
            Some(Arc::new(typed_value_struct)),
            Some(nulls),
        ))
    }

    /// Create comprehensive nested shredded variant with diverse null patterns
    /// Represents 3-level structure: variant -> outer -> inner (INT value)
    /// The shredding schema is: {"metadata": BINARY, "typed_value": {"outer": {"typed_value": {"inner": {"typed_value": INT}}}}}
    /// Rows: normal nested value, inner field null, outer field null, top-level null
    fn create_comprehensive_nested_shredded_variant() -> ArrayRef {
        // Create the inner level: contains typed_value with Int32 values
        // Row 0: has value 42, Row 1: inner null, Row 2: outer null, Row 3: top-level null
        let inner_typed_value = Int32Array::from(vec![Some(42), None, None, None]); // dummy value for row 2
        let inner = ShreddedVariantFieldArray::from_parts(
            None,
            Some(Arc::new(inner_typed_value) as ArrayRef),
            None,
        );

        let outer_typed_value_nulls = NullBuffer::from(vec![
            true,  // row 0: inner struct exists with typed_value=42
            false, // row 1: inner field NULL
            false, // row 2: outer field NULL
            false, // row 3: top-level NULL
        ]);
        let outer_typed_value = StructArrayBuilder::new()
            .with_field("inner", ArrayRef::from(inner), false)
            .with_nulls(outer_typed_value_nulls)
            .build();

        let outer = ShreddedVariantFieldArray::from_parts(
            None,
            Some(Arc::new(outer_typed_value) as ArrayRef),
            None,
        );

        let typed_value_nulls = NullBuffer::from(vec![
            true,  // row 0: inner struct exists with typed_value=42
            true,  // row 1: inner field NULL
            false, // row 2: outer field NULL
            false, // row 3: top-level NULL
        ]);
        let typed_value = StructArrayBuilder::new()
            .with_field("outer", ArrayRef::from(outer), false)
            .with_nulls(typed_value_nulls)
            .build();

        // Build final VariantArray with top-level nulls
        let metadata_array =
            BinaryViewArray::from_iter_values(std::iter::repeat_n(EMPTY_VARIANT_METADATA_BYTES, 4));
        let nulls = NullBuffer::from(vec![
            true,  // row 0: inner struct exists with typed_value=42
            true,  // row 1: inner field NULL
            true,  // row 2: outer field NULL
            false, // row 3: top-level NULL
        ]);
        ArrayRef::from(VariantArray::from_parts(
            Arc::new(metadata_array),
            None,
            Some(Arc::new(typed_value)),
            Some(nulls),
        ))
    }

    /// Create variant with mixed shredding (spec-compliant) including null scenarios
    /// Field "x" is globally shredded, field "y" is never shredded
    fn create_mixed_and_unshredded_variant() -> ArrayRef {
        // Create spec-compliant mixed shredding:
        // - Field "x" is globally shredded (has typed_value column)
        // - Field "y" is never shredded (only appears in value field when present)

        let (metadata, y_field_value) = {
            let mut builder = parquet_variant::VariantBuilder::new();
            let mut obj = builder.new_object();
            obj.insert("y", Variant::from(42));
            obj.finish();
            builder.finish()
        };

        let metadata_array = BinaryViewArray::from_iter_values(std::iter::repeat_n(&metadata, 4));

        // Value field contains objects with unshredded fields only (never contains "x")
        // Row 0: {"y": "foo"} - x is shredded out, y remains in value
        // Row 1: {} - both x and y are absent (perfect shredding for x, y missing)
        // Row 2: {"y": null} - x is shredded out, y explicitly null
        // Row 3: top-level null (encoded in VariantArray's null mask, but fields contain valid data)

        let empty_object_value = {
            let mut builder = parquet_variant::VariantBuilder::new();
            builder.new_object().finish();
            let (_, value) = builder.finish();
            value
        };

        let y_null_value = {
            let mut builder = parquet_variant::VariantBuilder::new();
            builder.new_object().with_field("y", Variant::Null).finish();
            let (_, value) = builder.finish();
            value
        };

        let value_array = BinaryViewArray::from(vec![
            Some(y_field_value.as_slice()),      // Row 0: {"y": 42}
            Some(empty_object_value.as_slice()), // Row 1: {}
            Some(y_null_value.as_slice()),       // Row 2: {"y": null}
            Some(empty_object_value.as_slice()), // Row 3: top-level null (but value field contains valid data)
        ]);

        // Create shredded field "x" (globally shredded - never appears in value field)
        // For top-level null row, the field still needs valid content (not null)
        let x_field_typed_value = Int32Array::from(vec![Some(1), Some(2), Some(3), Some(0)]);
        let x_field_shredded = ShreddedVariantFieldArray::from_parts(
            None,
            Some(Arc::new(x_field_typed_value) as ArrayRef),
            None,
        );

        // Create main typed_value struct (only contains shredded fields)
        let typed_value_struct = StructArrayBuilder::new()
            .with_field("x", ArrayRef::from(x_field_shredded), false)
            .build();

        // Build VariantArray with both value and typed_value (PartiallyShredded)
        // Top-level null is encoded in the main StructArray's null mask
        let variant_nulls = NullBuffer::from(vec![true, true, true, false]); // Row 3 is top-level null
        ArrayRef::from(VariantArray::from_parts(
            Arc::new(metadata_array),
            Some(Arc::new(value_array)),
            Some(Arc::new(typed_value_struct)),
            Some(variant_nulls),
        ))
    }

    #[test]
    fn get_decimal32_rescaled_to_scale2() {
        // Build unshredded variant values with different scales
        let mut builder = crate::VariantArrayBuilder::new(5);
        builder.append_variant(VariantDecimal4::try_new(1234, 2).unwrap().into()); // 12.34
        builder.append_variant(VariantDecimal4::try_new(1234, 3).unwrap().into()); // 1.234
        builder.append_variant(VariantDecimal4::try_new(1234, 0).unwrap().into()); // 1234
        builder.append_null();
        builder.append_variant(
            VariantDecimal8::try_new((VariantDecimal4::MAX_UNSCALED_VALUE as i64) + 1, 3)
                .unwrap()
                .into(),
        ); // should fit into Decimal32
        let variant_array: ArrayRef = ArrayRef::from(builder.build());

        let field = Field::new("result", DataType::Decimal32(9, 2), true);
        let options = GetOptions::new().with_as_type(Some(FieldRef::from(field)));
        let result = variant_get(&variant_array, options).unwrap();
        let result = result.as_any().downcast_ref::<Decimal32Array>().unwrap();

        assert_eq!(result.precision(), 9);
        assert_eq!(result.scale(), 2);
        assert_eq!(result.value(0), 1234);
        assert_eq!(result.value(1), 123);
        assert_eq!(result.value(2), 123400);
        assert!(result.is_null(3));
        assert_eq!(
            result.value(4),
            VariantDecimal4::MAX_UNSCALED_VALUE / 10 + 1
        ); // should not be null as the final result fits into Decimal32
    }

    #[test]
    fn get_decimal32_scale_down_rounding() {
        let mut builder = crate::VariantArrayBuilder::new(7);
        builder.append_variant(VariantDecimal4::try_new(1235, 0).unwrap().into());
        builder.append_variant(VariantDecimal4::try_new(1245, 0).unwrap().into());
        builder.append_variant(VariantDecimal4::try_new(-1235, 0).unwrap().into());
        builder.append_variant(VariantDecimal4::try_new(-1245, 0).unwrap().into());
        builder.append_variant(VariantDecimal4::try_new(1235, 2).unwrap().into()); // 12.35 rounded down to 10 for scale -1
        builder.append_variant(VariantDecimal4::try_new(1235, 3).unwrap().into()); // 1.235 rounded down to 0 for scale -1
        builder.append_variant(VariantDecimal4::try_new(5235, 3).unwrap().into()); // 5.235 rounded up to 10 for scale -1
        let variant_array: ArrayRef = ArrayRef::from(builder.build());

        let field = Field::new("result", DataType::Decimal32(9, -1), true);
        let options = GetOptions::new().with_as_type(Some(FieldRef::from(field)));
        let result = variant_get(&variant_array, options).unwrap();
        let result = result.as_any().downcast_ref::<Decimal32Array>().unwrap();

        assert_eq!(result.precision(), 9);
        assert_eq!(result.scale(), -1);
        assert_eq!(result.value(0), 124);
        assert_eq!(result.value(1), 125);
        assert_eq!(result.value(2), -124);
        assert_eq!(result.value(3), -125);
        assert_eq!(result.value(4), 1);
        assert!(result.is_valid(5));
        assert_eq!(result.value(5), 0);
        assert_eq!(result.value(6), 1);
    }

    #[test]
    fn get_decimal32_large_scale_reduction() {
        let mut builder = crate::VariantArrayBuilder::new(2);
        builder.append_variant(
            VariantDecimal4::try_new(-VariantDecimal4::MAX_UNSCALED_VALUE, 0)
                .unwrap()
                .into(),
        );
        builder.append_variant(
            VariantDecimal4::try_new(VariantDecimal4::MAX_UNSCALED_VALUE, 0)
                .unwrap()
                .into(),
        );
        let variant_array: ArrayRef = ArrayRef::from(builder.build());

        let field = Field::new("result", DataType::Decimal32(9, -9), true);
        let options = GetOptions::new().with_as_type(Some(FieldRef::from(field)));
        let result = variant_get(&variant_array, options).unwrap();
        let result = result.as_any().downcast_ref::<Decimal32Array>().unwrap();

        assert_eq!(result.precision(), 9);
        assert_eq!(result.scale(), -9);
        assert_eq!(result.value(0), -1);
        assert_eq!(result.value(1), 1);

        let field = Field::new("result", DataType::Decimal32(9, -10), true);
        let options = GetOptions::new().with_as_type(Some(FieldRef::from(field)));
        let result = variant_get(&variant_array, options).unwrap();
        let result = result.as_any().downcast_ref::<Decimal32Array>().unwrap();

        assert_eq!(result.precision(), 9);
        assert_eq!(result.scale(), -10);
        assert!(result.is_valid(0));
        assert_eq!(result.value(0), 0);
        assert!(result.is_valid(1));
        assert_eq!(result.value(1), 0);
    }

    #[test]
    fn get_decimal32_precision_overflow_safe() {
        // Exceed Decimal32 after scaling and rounding
        let mut builder = crate::VariantArrayBuilder::new(2);
        builder.append_variant(
            VariantDecimal4::try_new(VariantDecimal4::MAX_UNSCALED_VALUE, 0)
                .unwrap()
                .into(),
        );
        builder.append_variant(
            VariantDecimal4::try_new(VariantDecimal4::MAX_UNSCALED_VALUE, 9)
                .unwrap()
                .into(),
        ); // integer value round up overflows
        let variant_array: ArrayRef = ArrayRef::from(builder.build());

        let field = Field::new("result", DataType::Decimal32(2, 2), true);
        let options = GetOptions::new().with_as_type(Some(FieldRef::from(field)));
        let result = variant_get(&variant_array, options).unwrap();
        let result = result.as_any().downcast_ref::<Decimal32Array>().unwrap();

        assert!(result.is_null(0));
        assert!(result.is_null(1)); // should overflow because 1.00 does not fit into precision (2)
    }

    #[test]
    fn get_decimal32_precision_overflow_unsafe_errors() {
        let mut builder = crate::VariantArrayBuilder::new(1);
        builder.append_variant(
            VariantDecimal4::try_new(VariantDecimal4::MAX_UNSCALED_VALUE, 0)
                .unwrap()
                .into(),
        );
        let variant_array: ArrayRef = ArrayRef::from(builder.build());

        let field = Field::new("result", DataType::Decimal32(9, 2), true);
        let cast_options = CastOptions {
            safe: false,
            ..Default::default()
        };
        let options = GetOptions::new()
            .with_as_type(Some(FieldRef::from(field)))
            .with_cast_options(cast_options);
        let err = variant_get(&variant_array, options).unwrap_err();

        assert!(
            err.to_string().contains(
                "Failed to cast to Decimal32(precision=9, scale=2) from variant Decimal4"
            )
        );
    }

    #[test]
    fn get_decimal64_rescaled_to_scale2() {
        let mut builder = crate::VariantArrayBuilder::new(5);
        builder.append_variant(VariantDecimal8::try_new(1234, 2).unwrap().into()); // 12.34
        builder.append_variant(VariantDecimal8::try_new(1234, 3).unwrap().into()); // 1.234
        builder.append_variant(VariantDecimal8::try_new(1234, 0).unwrap().into()); // 1234
        builder.append_null();
        builder.append_variant(
            VariantDecimal16::try_new((VariantDecimal8::MAX_UNSCALED_VALUE as i128) + 1, 3)
                .unwrap()
                .into(),
        ); // should fit into Decimal64
        let variant_array: ArrayRef = ArrayRef::from(builder.build());

        let field = Field::new("result", DataType::Decimal64(18, 2), true);
        let options = GetOptions::new().with_as_type(Some(FieldRef::from(field)));
        let result = variant_get(&variant_array, options).unwrap();
        let result = result.as_any().downcast_ref::<Decimal64Array>().unwrap();

        assert_eq!(result.precision(), 18);
        assert_eq!(result.scale(), 2);
        assert_eq!(result.value(0), 1234);
        assert_eq!(result.value(1), 123);
        assert_eq!(result.value(2), 123400);
        assert!(result.is_null(3));
        assert_eq!(
            result.value(4),
            VariantDecimal8::MAX_UNSCALED_VALUE / 10 + 1
        ); // should not be null as the final result fits into Decimal64
    }

    #[test]
    fn get_decimal64_scale_down_rounding() {
        let mut builder = crate::VariantArrayBuilder::new(7);
        builder.append_variant(VariantDecimal8::try_new(1235, 0).unwrap().into());
        builder.append_variant(VariantDecimal8::try_new(1245, 0).unwrap().into());
        builder.append_variant(VariantDecimal8::try_new(-1235, 0).unwrap().into());
        builder.append_variant(VariantDecimal8::try_new(-1245, 0).unwrap().into());
        builder.append_variant(VariantDecimal8::try_new(1235, 2).unwrap().into()); // 12.35 rounded down to 10 for scale -1
        builder.append_variant(VariantDecimal8::try_new(1235, 3).unwrap().into()); // 1.235 rounded down to 0 for scale -1
        builder.append_variant(VariantDecimal8::try_new(5235, 3).unwrap().into()); // 5.235 rounded up to 10 for scale -1
        let variant_array: ArrayRef = ArrayRef::from(builder.build());

        let field = Field::new("result", DataType::Decimal64(18, -1), true);
        let options = GetOptions::new().with_as_type(Some(FieldRef::from(field)));
        let result = variant_get(&variant_array, options).unwrap();
        let result = result.as_any().downcast_ref::<Decimal64Array>().unwrap();

        assert_eq!(result.precision(), 18);
        assert_eq!(result.scale(), -1);
        assert_eq!(result.value(0), 124);
        assert_eq!(result.value(1), 125);
        assert_eq!(result.value(2), -124);
        assert_eq!(result.value(3), -125);
        assert_eq!(result.value(4), 1);
        assert!(result.is_valid(5));
        assert_eq!(result.value(5), 0);
        assert_eq!(result.value(6), 1);
    }

    #[test]
    fn get_decimal64_large_scale_reduction() {
        let mut builder = crate::VariantArrayBuilder::new(2);
        builder.append_variant(
            VariantDecimal8::try_new(-VariantDecimal8::MAX_UNSCALED_VALUE, 0)
                .unwrap()
                .into(),
        );
        builder.append_variant(
            VariantDecimal8::try_new(VariantDecimal8::MAX_UNSCALED_VALUE, 0)
                .unwrap()
                .into(),
        );
        let variant_array: ArrayRef = ArrayRef::from(builder.build());

        let field = Field::new("result", DataType::Decimal64(18, -18), true);
        let options = GetOptions::new().with_as_type(Some(FieldRef::from(field)));
        let result = variant_get(&variant_array, options).unwrap();
        let result = result.as_any().downcast_ref::<Decimal64Array>().unwrap();

        assert_eq!(result.precision(), 18);
        assert_eq!(result.scale(), -18);
        assert_eq!(result.value(0), -1);
        assert_eq!(result.value(1), 1);

        let field = Field::new("result", DataType::Decimal64(18, -19), true);
        let options = GetOptions::new().with_as_type(Some(FieldRef::from(field)));
        let result = variant_get(&variant_array, options).unwrap();
        let result = result.as_any().downcast_ref::<Decimal64Array>().unwrap();

        assert_eq!(result.precision(), 18);
        assert_eq!(result.scale(), -19);
        assert!(result.is_valid(0));
        assert_eq!(result.value(0), 0);
        assert!(result.is_valid(1));
        assert_eq!(result.value(1), 0);
    }

    #[test]
    fn get_decimal64_precision_overflow_safe() {
        // Exceed Decimal64 after scaling and rounding
        let mut builder = crate::VariantArrayBuilder::new(2);
        builder.append_variant(
            VariantDecimal8::try_new(VariantDecimal8::MAX_UNSCALED_VALUE, 0)
                .unwrap()
                .into(),
        );
        builder.append_variant(
            VariantDecimal8::try_new(VariantDecimal8::MAX_UNSCALED_VALUE, 18)
                .unwrap()
                .into(),
        ); // integer value round up overflows
        let variant_array: ArrayRef = ArrayRef::from(builder.build());

        let field = Field::new("result", DataType::Decimal64(2, 2), true);
        let options = GetOptions::new().with_as_type(Some(FieldRef::from(field)));
        let result = variant_get(&variant_array, options).unwrap();
        let result = result.as_any().downcast_ref::<Decimal64Array>().unwrap();

        assert!(result.is_null(0));
        assert!(result.is_null(1));
    }

    #[test]
    fn get_decimal64_precision_overflow_unsafe_errors() {
        let mut builder = crate::VariantArrayBuilder::new(1);
        builder.append_variant(
            VariantDecimal8::try_new(VariantDecimal8::MAX_UNSCALED_VALUE, 0)
                .unwrap()
                .into(),
        );
        let variant_array: ArrayRef = ArrayRef::from(builder.build());

        let field = Field::new("result", DataType::Decimal64(18, 2), true);
        let cast_options = CastOptions {
            safe: false,
            ..Default::default()
        };
        let options = GetOptions::new()
            .with_as_type(Some(FieldRef::from(field)))
            .with_cast_options(cast_options);
        let err = variant_get(&variant_array, options).unwrap_err();

        assert!(
            err.to_string().contains(
                "Failed to cast to Decimal64(precision=18, scale=2) from variant Decimal8"
            )
        );
    }

    #[test]
    fn get_decimal128_rescaled_to_scale2() {
        let mut builder = crate::VariantArrayBuilder::new(4);
        builder.append_variant(VariantDecimal16::try_new(1234, 2).unwrap().into());
        builder.append_variant(VariantDecimal16::try_new(1234, 3).unwrap().into());
        builder.append_variant(VariantDecimal16::try_new(1234, 0).unwrap().into());
        builder.append_null();
        let variant_array: ArrayRef = ArrayRef::from(builder.build());

        let field = Field::new("result", DataType::Decimal128(38, 2), true);
        let options = GetOptions::new().with_as_type(Some(FieldRef::from(field)));
        let result = variant_get(&variant_array, options).unwrap();
        let result = result.as_any().downcast_ref::<Decimal128Array>().unwrap();

        assert_eq!(result.precision(), 38);
        assert_eq!(result.scale(), 2);
        assert_eq!(result.value(0), 1234);
        assert_eq!(result.value(1), 123);
        assert_eq!(result.value(2), 123400);
        assert!(result.is_null(3));
    }

    #[test]
    fn get_decimal128_scale_down_rounding() {
        let mut builder = crate::VariantArrayBuilder::new(7);
        builder.append_variant(VariantDecimal16::try_new(1235, 0).unwrap().into());
        builder.append_variant(VariantDecimal16::try_new(1245, 0).unwrap().into());
        builder.append_variant(VariantDecimal16::try_new(-1235, 0).unwrap().into());
        builder.append_variant(VariantDecimal16::try_new(-1245, 0).unwrap().into());
        builder.append_variant(VariantDecimal16::try_new(1235, 2).unwrap().into()); // 12.35 rounded down to 10 for scale -1
        builder.append_variant(VariantDecimal16::try_new(1235, 3).unwrap().into()); // 1.235 rounded down to 0 for scale -1
        builder.append_variant(VariantDecimal16::try_new(5235, 3).unwrap().into()); // 5.235 rounded up to 10 for scale -1
        let variant_array: ArrayRef = ArrayRef::from(builder.build());

        let field = Field::new("result", DataType::Decimal128(38, -1), true);
        let options = GetOptions::new().with_as_type(Some(FieldRef::from(field)));
        let result = variant_get(&variant_array, options).unwrap();
        let result = result.as_any().downcast_ref::<Decimal128Array>().unwrap();

        assert_eq!(result.precision(), 38);
        assert_eq!(result.scale(), -1);
        assert_eq!(result.value(0), 124);
        assert_eq!(result.value(1), 125);
        assert_eq!(result.value(2), -124);
        assert_eq!(result.value(3), -125);
        assert_eq!(result.value(4), 1);
        assert!(result.is_valid(5));
        assert_eq!(result.value(5), 0);
        assert_eq!(result.value(6), 1);
    }

    #[test]
    fn get_decimal128_precision_overflow_safe() {
        // Exceed Decimal128 after scaling and rounding
        let mut builder = crate::VariantArrayBuilder::new(2);
        builder.append_variant(
            VariantDecimal16::try_new(VariantDecimal16::MAX_UNSCALED_VALUE, 0)
                .unwrap()
                .into(),
        );
        builder.append_variant(
            VariantDecimal16::try_new(VariantDecimal16::MAX_UNSCALED_VALUE, 38)
                .unwrap()
                .into(),
        ); // integer value round up overflows
        let variant_array: ArrayRef = ArrayRef::from(builder.build());

        let field = Field::new("result", DataType::Decimal128(2, 2), true);
        let options = GetOptions::new().with_as_type(Some(FieldRef::from(field)));
        let result = variant_get(&variant_array, options).unwrap();
        let result = result.as_any().downcast_ref::<Decimal128Array>().unwrap();

        assert!(result.is_null(0));
        assert!(result.is_null(1)); // should overflow because 1.00 does not fit into precision (2)
    }

    #[test]
    fn get_decimal128_precision_overflow_unsafe_errors() {
        let mut builder = crate::VariantArrayBuilder::new(1);
        builder.append_variant(
            VariantDecimal16::try_new(VariantDecimal16::MAX_UNSCALED_VALUE, 0)
                .unwrap()
                .into(),
        );
        let variant_array: ArrayRef = ArrayRef::from(builder.build());

        let field = Field::new("result", DataType::Decimal128(38, 2), true);
        let cast_options = CastOptions {
            safe: false,
            ..Default::default()
        };
        let options = GetOptions::new()
            .with_as_type(Some(FieldRef::from(field)))
            .with_cast_options(cast_options);
        let err = variant_get(&variant_array, options).unwrap_err();

        assert!(err.to_string().contains(
            "Failed to cast to Decimal128(precision=38, scale=2) from variant Decimal16"
        ));
    }

    #[test]
    fn get_decimal256_rescaled_to_scale2() {
        // Build unshredded variant values with different scales using Decimal16 source
        let mut builder = crate::VariantArrayBuilder::new(4);
        builder.append_variant(VariantDecimal16::try_new(1234, 2).unwrap().into()); // 12.34
        builder.append_variant(VariantDecimal16::try_new(1234, 3).unwrap().into()); // 1.234
        builder.append_variant(VariantDecimal16::try_new(1234, 0).unwrap().into()); // 1234
        builder.append_null();
        let variant_array: ArrayRef = ArrayRef::from(builder.build());

        let field = Field::new("result", DataType::Decimal256(76, 2), true);
        let options = GetOptions::new().with_as_type(Some(FieldRef::from(field)));
        let result = variant_get(&variant_array, options).unwrap();
        let result = result.as_any().downcast_ref::<Decimal256Array>().unwrap();

        assert_eq!(result.precision(), 76);
        assert_eq!(result.scale(), 2);
        assert_eq!(result.value(0), i256::from_i128(1234));
        assert_eq!(result.value(1), i256::from_i128(123));
        assert_eq!(result.value(2), i256::from_i128(123400));
        assert!(result.is_null(3));
    }

    #[test]
    fn get_decimal256_scale_down_rounding() {
        let mut builder = crate::VariantArrayBuilder::new(7);
        builder.append_variant(VariantDecimal16::try_new(1235, 0).unwrap().into());
        builder.append_variant(VariantDecimal16::try_new(1245, 0).unwrap().into());
        builder.append_variant(VariantDecimal16::try_new(-1235, 0).unwrap().into());
        builder.append_variant(VariantDecimal16::try_new(-1245, 0).unwrap().into());
        builder.append_variant(VariantDecimal16::try_new(1235, 2).unwrap().into()); // 12.35 rounded down to 10 for scale -1
        builder.append_variant(VariantDecimal16::try_new(1235, 3).unwrap().into()); // 1.235 rounded down to 0 for scale -1
        builder.append_variant(VariantDecimal16::try_new(5235, 3).unwrap().into()); // 5.235 rounded up to 10 for scale -1
        let variant_array: ArrayRef = ArrayRef::from(builder.build());

        let field = Field::new("result", DataType::Decimal256(76, -1), true);
        let options = GetOptions::new().with_as_type(Some(FieldRef::from(field)));
        let result = variant_get(&variant_array, options).unwrap();
        let result = result.as_any().downcast_ref::<Decimal256Array>().unwrap();

        assert_eq!(result.precision(), 76);
        assert_eq!(result.scale(), -1);
        assert_eq!(result.value(0), i256::from_i128(124));
        assert_eq!(result.value(1), i256::from_i128(125));
        assert_eq!(result.value(2), i256::from_i128(-124));
        assert_eq!(result.value(3), i256::from_i128(-125));
        assert_eq!(result.value(4), i256::from_i128(1));
        assert!(result.is_valid(5));
        assert_eq!(result.value(5), i256::from_i128(0));
        assert_eq!(result.value(6), i256::from_i128(1));
    }

    #[test]
    fn get_decimal256_precision_overflow_safe() {
        // Exceed Decimal128 max precision (38) after scaling
        let mut builder = crate::VariantArrayBuilder::new(2);
        builder.append_variant(
            VariantDecimal16::try_new(VariantDecimal16::MAX_UNSCALED_VALUE, 1)
                .unwrap()
                .into(),
        );
        builder.append_variant(
            VariantDecimal16::try_new(VariantDecimal16::MAX_UNSCALED_VALUE, 0)
                .unwrap()
                .into(),
        );
        let variant_array: ArrayRef = ArrayRef::from(builder.build());

        let field = Field::new("result", DataType::Decimal256(76, 39), true);
        let options = GetOptions::new().with_as_type(Some(FieldRef::from(field)));
        let result = variant_get(&variant_array, options).unwrap();
        let result = result.as_any().downcast_ref::<Decimal256Array>().unwrap();

        // Input is Decimal16 with integer = 10^38-1 and scale = 1, target scale = 39
        // So expected integer is (10^38-1) * 10^(39-1) = (10^38-1) * 10^38
        let base = i256::from_i128(10);
        let factor = base.checked_pow(38).unwrap();
        let expected = i256::from_i128(VariantDecimal16::MAX_UNSCALED_VALUE)
            .checked_mul(factor)
            .unwrap();
        assert_eq!(result.value(0), expected);
        assert!(result.is_null(1));
    }

    #[test]
    fn get_decimal256_precision_overflow_unsafe_errors() {
        // Exceed Decimal128 max precision (38) after scaling
        let mut builder = crate::VariantArrayBuilder::new(2);
        builder.append_variant(
            VariantDecimal16::try_new(VariantDecimal16::MAX_UNSCALED_VALUE, 1)
                .unwrap()
                .into(),
        );
        builder.append_variant(
            VariantDecimal16::try_new(VariantDecimal16::MAX_UNSCALED_VALUE, 0)
                .unwrap()
                .into(),
        );
        let variant_array: ArrayRef = ArrayRef::from(builder.build());

        let field = Field::new("result", DataType::Decimal256(76, 39), true);
        let cast_options = CastOptions {
            safe: false,
            ..Default::default()
        };
        let options = GetOptions::new()
            .with_as_type(Some(FieldRef::from(field)))
            .with_cast_options(cast_options);
        let err = variant_get(&variant_array, options).unwrap_err();

        assert!(err.to_string().contains(
            "Failed to cast to Decimal256(precision=76, scale=39) from variant Decimal16"
        ));
    }

    #[test]
    fn get_non_supported_temporal_types_error() {
        let values = vec![None, Some(Variant::Null), Some(Variant::BooleanFalse)];
        let variant_array: ArrayRef = ArrayRef::from(VariantArray::from_iter(values));

        let test_cases = vec![
            FieldRef::from(Field::new(
                "result",
                DataType::Duration(TimeUnit::Microsecond),
                true,
            )),
            FieldRef::from(Field::new(
                "result",
                DataType::Interval(IntervalUnit::YearMonth),
                true,
            )),
        ];

        for field in test_cases {
            let options = GetOptions::new().with_as_type(Some(field));
            let err = variant_get(&variant_array, options).unwrap_err();
            assert!(
                err.to_string()
                    .contains("Casting Variant to duration/interval types is not supported")
            );
        }
    }

    #[test]
    fn get_variant_as_dictionary() {
        let variant_array: ArrayRef = ArrayRef::from(VariantArray::from_iter(vec![
            Some(Variant::from("apple")),
            Some(Variant::from("banana")),
            None,
            Some(Variant::from("apple")),
        ]));
        let data_type = DataType::Dictionary(Box::new(DataType::Int32), Box::new(DataType::Utf8));
        let options = GetOptions::new().with_as_type(Some(FieldRef::from(Field::new(
            "dict",
            data_type.clone(),
            true,
        ))));

        let result = variant_get(&variant_array, options).unwrap();
        assert_eq!(result.data_type(), &data_type);

        let decoded = cast(result.as_ref(), &DataType::Utf8).unwrap();
        let expected = StringArray::from(vec![Some("apple"), Some("banana"), None, Some("apple")]);
        assert_eq!(decoded.as_ref(), &expected);
    }

    #[test]
    fn get_variant_as_numeric_dictionary() {
        let variant_array: ArrayRef = ArrayRef::from(VariantArray::from_iter(vec![
            Some(Variant::from(42)),
            Some(Variant::from(7)),
            None,
            Some(Variant::from(42)),
        ]));
        let data_type = DataType::Dictionary(Box::new(DataType::Int16), Box::new(DataType::Int32));
        let options = GetOptions::new().with_as_type(Some(FieldRef::from(Field::new(
            "dict",
            data_type.clone(),
            true,
        ))));

        let result = variant_get(&variant_array, options).unwrap();
        assert_eq!(result.data_type(), &data_type);

        let decoded = cast(result.as_ref(), &DataType::Int32).unwrap();
        let expected = Int32Array::from(vec![Some(42), Some(7), None, Some(42)]);
        assert_eq!(decoded.as_ref(), &expected);
    }

    #[test]
    fn get_variant_as_run_end_encoded() {
        let variant_array: ArrayRef = ArrayRef::from(VariantArray::from_iter(vec![
            Some(Variant::from("apple")),
            Some(Variant::from("apple")),
            None,
            Some(Variant::from("banana")),
            Some(Variant::from("banana")),
        ]));
        let run_ends = Arc::new(Field::new("run_ends", DataType::Int32, false));
        let values = Arc::new(Field::new("values", DataType::Utf8, true));
        let data_type = DataType::RunEndEncoded(run_ends, values);
        let options = GetOptions::new().with_as_type(Some(FieldRef::from(Field::new(
            "ree",
            data_type.clone(),
            true,
        ))));

        let result = variant_get(&variant_array, options).unwrap();
        assert_eq!(result.data_type(), &data_type);

        let decoded = cast(result.as_ref(), &DataType::Utf8).unwrap();
        let expected = StringArray::from(vec![
            Some("apple"),
            Some("apple"),
            None,
            Some("banana"),
            Some("banana"),
        ]);
        assert_eq!(decoded.as_ref(), &expected);
    }

    fn invalid_time_variant_array() -> ArrayRef {
        let mut builder = VariantArrayBuilder::new(3);
        // 86401000000 is invalid for Time64Microsecond (max is 86400000000)
        builder.append_variant(Variant::Int64(86401000000));
        builder.append_variant(Variant::Int64(86401000000));
        builder.append_variant(Variant::Int64(86401000000));
        Arc::new(builder.build().into_inner())
    }

    #[test]
    fn test_variant_get_error_when_cast_failure_and_safe_false() {
        let variant_array = invalid_time_variant_array();

        let field = Field::new("result", DataType::Time64(TimeUnit::Microsecond), true);
        let cast_options = CastOptions {
            safe: false, // Will error on cast failure
            ..Default::default()
        };
        let options = GetOptions::new()
            .with_as_type(Some(FieldRef::from(field)))
            .with_cast_options(cast_options);
        let err = variant_get(&variant_array, options).unwrap_err();
        assert!(
            err.to_string().contains(
                "Cast error: Failed to extract primitive of type Time64(µs) from variant Int64(86401000000) at path VariantPath([])"
            ),
            "actual: {err}",
        );
    }

    #[test]
    fn test_variant_get_return_null_when_cast_failure_and_safe_true() {
        let variant_array = invalid_time_variant_array();

        let field = Field::new("result", DataType::Time64(TimeUnit::Microsecond), true);
        let cast_options = CastOptions {
            safe: true, // Will return null on cast failure
            ..Default::default()
        };
        let options = GetOptions::new()
            .with_as_type(Some(FieldRef::from(field)))
            .with_cast_options(cast_options);
        let result = variant_get(&variant_array, options).unwrap();
        assert_eq!(3, result.len());

        for i in 0..3 {
            assert!(result.is_null(i));
        }
    }

    #[test]
    fn test_perfect_shredding_returns_same_arc_ptr() {
        let variant_array = perfectly_shredded_int32_variant_array();

        let variant_array_ref = VariantArray::try_new(&variant_array).unwrap();
        let typed_value_arc = variant_array_ref.typed_value_column().unwrap().clone();

        let field = Field::new("result", DataType::Int32, true);
        let options = GetOptions::new().with_as_type(Some(FieldRef::from(field)));
        let result = variant_get(&variant_array, options).unwrap();

        assert!(Arc::ptr_eq(&typed_value_arc, &result));
    }

    #[test]
    fn test_perfect_shredding_three_typed_value_columns() {
        // Column 1: perfectly shredded primitive with all nulls
        let all_nulls_values: Arc<Int32Array> = Arc::new(Int32Array::from(vec![
            Option::<i32>::None,
            Option::<i32>::None,
            Option::<i32>::None,
        ]));
        let all_nulls_erased: ArrayRef = all_nulls_values.clone();
        let all_nulls_field =
            ShreddedVariantFieldArray::from_parts(None, Some(all_nulls_erased.clone()), None);
        let all_nulls_type = all_nulls_field.data_type().clone();
        let all_nulls_struct: ArrayRef = ArrayRef::from(all_nulls_field);

        // Column 2: perfectly shredded primitive with some nulls
        let some_nulls_values: Arc<Int32Array> =
            Arc::new(Int32Array::from(vec![Some(10), None, Some(30)]));
        let some_nulls_erased: ArrayRef = some_nulls_values.clone();
        let some_nulls_field =
            ShreddedVariantFieldArray::from_parts(None, Some(some_nulls_erased.clone()), None);
        let some_nulls_type = some_nulls_field.data_type().clone();
        let some_nulls_struct: ArrayRef = ArrayRef::from(some_nulls_field);

        // Column 3: perfectly shredded nested struct
        let inner_values: Arc<Int32Array> =
            Arc::new(Int32Array::from(vec![Some(111), None, Some(333)]));
        let inner_erased: ArrayRef = inner_values.clone();
        let inner_field =
            ShreddedVariantFieldArray::from_parts(None, Some(inner_erased.clone()), None);
        let inner_field_type = inner_field.data_type().clone();
        let inner_struct_array: ArrayRef = ArrayRef::from(inner_field);

        let nested_struct = Arc::new(
            StructArray::try_new(
                Fields::from(vec![Field::new("inner", inner_field_type, true)]),
                vec![inner_struct_array],
                None,
            )
            .unwrap(),
        );
        let nested_struct_erased: ArrayRef = nested_struct.clone();
        let struct_field =
            ShreddedVariantFieldArray::from_parts(None, Some(nested_struct_erased.clone()), None);
        let struct_field_type = struct_field.data_type().clone();
        let struct_field_struct: ArrayRef = ArrayRef::from(struct_field);

        // Assemble the top-level typed_value struct with the three columns above
        let typed_value_struct = StructArray::try_new(
            Fields::from(vec![
                Field::new("all_nulls", all_nulls_type, true),
                Field::new("some_nulls", some_nulls_type, true),
                Field::new("struct_field", struct_field_type, true),
            ]),
            vec![all_nulls_struct, some_nulls_struct, struct_field_struct],
            None,
        )
        .unwrap();

        let metadata = BinaryViewArray::from_iter_values(std::iter::repeat_n(
            EMPTY_VARIANT_METADATA_BYTES,
            all_nulls_values.len(),
        ));
        let variant_array: ArrayRef = VariantArray::from_parts(
            Arc::new(metadata),
            None,
            Some(Arc::new(typed_value_struct)),
            None,
        )
        .into();

        // Case 1: all-null primitive column should reuse the typed_value Arc directly
        let all_nulls_field_ref = FieldRef::from(Field::new("result", DataType::Int32, true));
        let all_nulls_result = variant_get(
            &variant_array,
            GetOptions::new_with_path(VariantPath::try_from("all_nulls").unwrap())
                .with_as_type(Some(all_nulls_field_ref)),
        )
        .unwrap();
        assert!(Arc::ptr_eq(&all_nulls_result, &all_nulls_erased));

        // Case 2: primitive column with some nulls should also reuse its typed_value Arc
        let some_nulls_field_ref = FieldRef::from(Field::new("result", DataType::Int32, true));
        let some_nulls_result = variant_get(
            &variant_array,
            GetOptions::new_with_path(VariantPath::try_from("some_nulls").unwrap())
                .with_as_type(Some(some_nulls_field_ref)),
        )
        .unwrap();
        assert!(Arc::ptr_eq(&some_nulls_result, &some_nulls_erased));

        // Case 3: struct column should return a StructArray composed from the nested field
        let struct_child_fields = Fields::from(vec![Field::new("inner", DataType::Int32, true)]);
        let struct_field_ref = FieldRef::from(Field::new(
            "result",
            DataType::Struct(struct_child_fields.clone()),
            true,
        ));
        let struct_result = variant_get(
            &variant_array,
            GetOptions::new_with_path(VariantPath::try_from("struct_field").unwrap())
                .with_as_type(Some(struct_field_ref)),
        )
        .unwrap();
        let struct_array = struct_result
            .as_any()
            .downcast_ref::<StructArray>()
            .unwrap();
        assert_eq!(struct_array.len(), 3);
        assert_eq!(struct_array.null_count(), 0);

        let inner_values_result = struct_array
            .column(0)
            .as_any()
            .downcast_ref::<Int32Array>()
            .unwrap();
        assert_eq!(inner_values_result.len(), 3);
        assert_eq!(inner_values_result.value(0), 111);
        assert!(inner_values_result.is_null(1));
        assert_eq!(inner_values_result.value(2), 333);
    }

    #[test]
    fn test_variant_get_list_like_safe_cast() {
        let string_array: ArrayRef = Arc::new(StringArray::from(vec![
            r#"{"outer":{"list":[1, "two", 3]}}"#,
            r#"{"outer":{"list":"not a list"}}"#,
        ]));
        let variant_array = ArrayRef::from(json_to_variant(&string_array).unwrap());

        let element_array: ArrayRef = Arc::new(Int64Array::from(vec![Some(1), None, Some(3)]));
        let field = Arc::new(Field::new("item", Int64, true));

        let expectations = vec![
            (
                DataType::List(field.clone()),
                Arc::new(ListArray::new(
                    field.clone(),
                    OffsetBuffer::new(ScalarBuffer::from(vec![0, 3, 3])),
                    element_array.clone(),
                    Some(NullBuffer::from(vec![true, false])),
                )) as ArrayRef,
            ),
            (
                DataType::LargeList(field.clone()),
                Arc::new(LargeListArray::new(
                    field.clone(),
                    OffsetBuffer::new(ScalarBuffer::from(vec![0, 3, 3])),
                    element_array.clone(),
                    Some(NullBuffer::from(vec![true, false])),
                )) as ArrayRef,
            ),
            (
                DataType::ListView(field.clone()),
                Arc::new(ListViewArray::new(
                    field.clone(),
                    ScalarBuffer::from(vec![0, 3]),
                    ScalarBuffer::from(vec![3, 0]),
                    element_array.clone(),
                    Some(NullBuffer::from(vec![true, false])),
                )) as ArrayRef,
            ),
            (
                DataType::LargeListView(field.clone()),
                Arc::new(LargeListViewArray::new(
                    field.clone(),
                    ScalarBuffer::from(vec![0, 3]),
                    ScalarBuffer::from(vec![3, 0]),
                    element_array,
                    Some(NullBuffer::from(vec![true, false])),
                )) as ArrayRef,
            ),
            (
                DataType::FixedSizeList(field.clone(), 3),
                Arc::new(FixedSizeListArray::new(
                    field,
                    3,
                    Arc::new(Int64Array::from(vec![
                        Some(1),
                        None,
                        Some(3),
                        None,
                        None,
                        None,
                    ])),
                    Some(NullBuffer::from(vec![true, false])),
                )) as ArrayRef,
            ),
        ];

        for (request_type, expected) in expectations {
            let options =
                GetOptions::new_with_path(VariantPath::try_from("outer").unwrap().join("list"))
                    .with_as_type(Some(FieldRef::from(Field::new(
                        "result",
                        request_type.clone(),
                        true,
                    ))));

            let result = variant_get(&variant_array, options).unwrap();
            assert_eq!(result.data_type(), expected.data_type());
            assert_eq!(&result, &expected);
        }

        for (idx, expected) in [
            (0, vec![Some(1), None]),
            (1, vec![None, None]),
            (2, vec![Some(3), None]),
        ] {
            let index_options = GetOptions::new_with_path(
                VariantPath::try_from("outer")
                    .unwrap()
                    .join("list")
                    .join(idx),
            )
            .with_as_type(Some(FieldRef::from(Field::new(
                "result",
                DataType::Int64,
                true,
            ))));
            let index_result = variant_get(&variant_array, index_options).unwrap();
            let index_expected: ArrayRef = Arc::new(Int64Array::from(expected));
            assert_eq!(&index_result, &index_expected);
        }
    }

    #[test]
    fn test_variant_get_nested_list() {
        use arrow::datatypes::Int64Type;

        let string_array: ArrayRef = Arc::new(StringArray::from(vec![
            r#"[[1, 2], [3]]"#,
            r#"[[4], "not a list", [5, 6]]"#,
        ]));
        let variant_array = ArrayRef::from(json_to_variant(&string_array).unwrap());

        let inner_field = Arc::new(Field::new("item", Int64, true));
        let outer_field = Arc::new(Field::new(
            "item",
            DataType::List(inner_field.clone()),
            true,
        ));
        let request_type = DataType::List(outer_field.clone());

        let options = GetOptions::new().with_as_type(Some(FieldRef::from(Field::new(
            "result",
            request_type,
            true,
        ))));
        let result = variant_get(&variant_array, options).unwrap();
        let outer = result.as_list::<i32>();

        // Row 0: [[1, 2], [3]]
        let row0 = outer.value(0);
        let row0 = row0.as_list::<i32>();
        assert_eq!(row0.len(), 2);
        let elem0 = row0.value(0);
        assert_eq!(elem0.as_primitive::<Int64Type>().values(), &[1, 2]);
        let elem1 = row0.value(1);
        assert_eq!(elem1.as_primitive::<Int64Type>().values(), &[3]);

        // Row 1: [[4], null, [5, 6]] — "not a list" becomes null inner list
        let row1 = outer.value(1);
        let row1 = row1.as_list::<i32>();
        assert_eq!(row1.len(), 3);
        let elem0 = row1.value(0);
        assert_eq!(elem0.as_primitive::<Int64Type>().values(), &[4]);
        assert!(row1.is_null(1));
        let elem2 = row1.value(2);
        assert_eq!(elem2.as_primitive::<Int64Type>().values(), &[5, 6]);
    }

    #[test]
    fn test_variant_get_list_like_unsafe_cast_errors_on_element_mismatch() {
        let string_array: ArrayRef =
            Arc::new(StringArray::from(vec![r#"[1, "two", 3]"#, "[4, 5]"]));
        let variant_array = ArrayRef::from(json_to_variant(&string_array).unwrap());
        let cast_options = CastOptions {
            safe: false,
            ..Default::default()
        };

        let item_field = Arc::new(Field::new("item", DataType::Int64, true));
        let request_types = vec![
            DataType::List(item_field.clone()),
            DataType::LargeList(item_field.clone()),
            DataType::ListView(item_field.clone()),
            DataType::LargeListView(item_field),
        ];

        for request_type in request_types {
            let options = GetOptions::new()
                .with_as_type(Some(FieldRef::from(Field::new(
                    "result",
                    request_type.clone(),
                    true,
                ))))
                .with_cast_options(cast_options.clone());

            let err = variant_get(&variant_array, options).unwrap_err();
            assert!(
                err.to_string()
                    .contains("Failed to extract primitive of type Int64")
            );
        }
    }

    #[test]
    fn test_variant_get_list_like_unsafe_cast_preserves_null_elements() {
        let string_array: ArrayRef = Arc::new(StringArray::from(vec![r#"[1, null, 3]"#]));
        let variant_array = ArrayRef::from(json_to_variant(&string_array).unwrap());
        let cast_options = CastOptions {
            safe: false,
            ..Default::default()
        };
        let options = GetOptions::new()
            .with_as_type(Some(FieldRef::from(Field::new(
                "result",
                DataType::List(Arc::new(Field::new("item", DataType::Int64, true))),
                true,
            ))))
            .with_cast_options(cast_options);

        let result = variant_get(&variant_array, options).unwrap();
        let list_array = result.as_any().downcast_ref::<ListArray>().unwrap();
        let values = list_array
            .values()
            .as_any()
            .downcast_ref::<Int64Array>()
            .unwrap();

        assert_eq!(values.len(), 3);
        assert_eq!(values.value(0), 1);
        assert!(values.is_null(1));
        assert_eq!(values.value(2), 3);
    }

    #[test]
    fn test_variant_get_list_like_unsafe_cast_errors_on_non_list() {
        let string_array: ArrayRef = Arc::new(StringArray::from(vec!["[1, 2]", "\"not a list\""]));
        let variant_array = ArrayRef::from(json_to_variant(&string_array).unwrap());
        let cast_options = CastOptions {
            safe: false,
            ..Default::default()
        };
        let item_field = Arc::new(Field::new("item", Int64, true));
        let data_types = vec![
            DataType::List(item_field.clone()),
            DataType::LargeList(item_field.clone()),
            DataType::ListView(item_field.clone()),
            DataType::LargeListView(item_field.clone()),
            DataType::FixedSizeList(item_field, 2),
        ];

        for data_type in data_types {
            let options = GetOptions::new()
                .with_as_type(Some(FieldRef::from(Field::new("result", data_type, true))))
                .with_cast_options(cast_options.clone());

            let err = variant_get(&variant_array, options).unwrap_err();
            assert!(
                err.to_string()
                    .contains("Failed to extract list from variant"),
            );
        }
    }

    #[test]
    fn test_variant_get_fixed_size_list_wrong_size() {
        let string_array: ArrayRef = Arc::new(StringArray::from(vec!["[1, 2, 3]"]));
        let variant_array = ArrayRef::from(json_to_variant(&string_array).unwrap());
        let item_field = Arc::new(Field::new("item", Int64, true));

        // With `safe` set to true, size mismatch should return Null.
        let options = GetOptions::new()
            .with_as_type(Some(FieldRef::from(Field::new(
                "result",
                DataType::FixedSizeList(item_field.clone(), 2),
                true,
            ))))
            .with_cast_options(CastOptions {
                safe: true,
                ..Default::default()
            });
        let result = variant_get(&variant_array, options).unwrap();
        let fixed_size_list = result
            .as_any()
            .downcast_ref::<FixedSizeListArray>()
            .expect("Expected FixedSizeListArray");
        assert_eq!(fixed_size_list.len(), 1);
        assert!(fixed_size_list.is_null(0));

        // With `safe` set to false, error should be raised on wrong sized fixed list.
        let options = GetOptions::new()
            .with_as_type(Some(FieldRef::from(Field::new(
                "result",
                DataType::FixedSizeList(item_field.clone(), 2),
                true,
            ))))
            .with_cast_options(CastOptions {
                safe: false,
                ..Default::default()
            });
        let err = variant_get(&variant_array, options).unwrap_err();
        assert!(
            err.to_string()
                .contains("Expected fixed size list of size 2, got size 3"),
            "got: {err}",
        );
    }

    macro_rules! perfectly_shredded_preserves_top_level_nulls_test {
        ($name:ident, $result_type:expr, $typed_value:expr, $expected_array:expr) => {
            perfectly_shredded_preserves_top_level_nulls_test!(
                $name,
                $result_type,
                $typed_value,
                Some(NullBuffer::from(vec![true, false, true])),
                $expected_array
            );
        };
        ($name:ident, $result_type:expr, $typed_value:expr, $parent_nulls:expr, $expected_array:expr) => {
            #[test]
            fn $name() {
                let metadata = Arc::new(BinaryViewArray::from_iter_values(std::iter::repeat_n(
                    EMPTY_VARIANT_METADATA_BYTES,
                    3,
                )));
                let typed_value: ArrayRef = Arc::new($typed_value);
                let variant_array: ArrayRef =
                    VariantArray::from_parts(metadata, None, Some(typed_value), $parent_nulls)
                        .into();

                let result = variant_get(
                    &variant_array,
                    GetOptions::new().with_as_type(Some(FieldRef::from(Field::new(
                        "result",
                        $result_type,
                        true,
                    )))),
                )
                .unwrap();

                let expected_array: ArrayRef = Arc::new($expected_array);
                assert_eq!(&result, &expected_array);
            }
        };
    }

    perfectly_shredded_preserves_top_level_nulls_test!(
        test_variant_get_perfectly_shredded_integer_preserves_top_level_nulls,
        DataType::Int32,
        Int32Array::from(vec![Some(0_i32), Some(1_i32), Some(2_i32)]),
        Int32Array::from(vec![Some(0_i32), None, Some(2_i32)])
    );

    perfectly_shredded_preserves_top_level_nulls_test!(
        test_variant_get_perfectly_shredded_integer_unions_child_and_top_level_nulls,
        DataType::Int32,
        Int32Array::from(vec![None, Some(1_i32), Some(2_i32)]),
        Some(NullBuffer::from(vec![true, false, true])),
        Int32Array::from(vec![None, None, Some(2_i32)])
    );

    perfectly_shredded_preserves_top_level_nulls_test!(
        test_variant_get_perfectly_shredded_null_preserves_top_level_nulls,
        DataType::Null,
        NullArray::new(3),
        NullArray::new(3)
    );

    perfectly_shredded_preserves_top_level_nulls_test!(
        test_variant_get_perfectly_shredded_binary_view_preserves_top_level_nulls,
        DataType::BinaryView,
        BinaryViewArray::from(vec![
            Some(b"Apache" as &[u8]),
            Some(b"masked-null" as &[u8]),
            Some(b"Parquet-variant" as &[u8]),
        ]),
        BinaryViewArray::from(vec![
            Some(b"Apache" as &[u8]),
            None,
            Some(b"Parquet-variant" as &[u8]),
        ])
    );

    perfectly_shredded_preserves_top_level_nulls_test!(
        test_variant_get_perfectly_shredded_binary_preserves_top_level_nulls,
        DataType::Binary,
        BinaryArray::from(vec![
            Some(b"Apache" as &[u8]),
            Some(b"masked-null" as &[u8]),
            Some(b"Parquet-variant" as &[u8]),
        ]),
        BinaryArray::from(vec![
            Some(b"Apache" as &[u8]),
            None,
            Some(b"Parquet-variant" as &[u8]),
        ])
    );

    perfectly_shredded_preserves_top_level_nulls_test!(
        test_variant_get_perfectly_shredded_decimal4_preserves_top_level_nulls,
        DataType::Decimal32(5, 2),
        Decimal32Array::from(vec![Some(12345), Some(23400), Some(-12342)])
            .with_precision_and_scale(5, 2)
            .unwrap(),
        Decimal32Array::from(vec![Some(12345), None, Some(-12342)])
            .with_precision_and_scale(5, 2)
            .unwrap()
    );

    perfectly_shredded_preserves_top_level_nulls_test!(
        test_variant_get_perfectly_shredded_decimal8_preserves_top_level_nulls,
        DataType::Decimal64(10, 1),
        Decimal64Array::from(vec![Some(1234567809), Some(1456787000), Some(-1234561203)])
            .with_precision_and_scale(10, 1)
            .unwrap(),
        Decimal64Array::from(vec![Some(1234567809), None, Some(-1234561203)])
            .with_precision_and_scale(10, 1)
            .unwrap()
    );

    perfectly_shredded_preserves_top_level_nulls_test!(
        test_variant_get_perfectly_shredded_decimal16_preserves_top_level_nulls,
        DataType::Decimal128(20, 3),
        Decimal128Array::from(vec![
            Some(i128::from_str("12345678901234567899").unwrap()),
            Some(i128::from_str("23445677483748324300").unwrap()),
            Some(i128::from_str("-12345678901234567899").unwrap()),
        ])
        .with_precision_and_scale(20, 3)
        .unwrap(),
        Decimal128Array::from(vec![
            Some(i128::from_str("12345678901234567899").unwrap()),
            None,
            Some(i128::from_str("-12345678901234567899").unwrap()),
        ])
        .with_precision_and_scale(20, 3)
        .unwrap()
    );
}