//! Cast from Variant UDF for DataFusion.
//!
//! Wraps `parquet_variant_compute::variant_get()` to cast Variant to a typed Arrow array.
//! For complex types (List, Struct, Map), we first shred the variant to the target type
//! before extraction, since `variant_get` doesn't support extracting complex types from
//! unshredded variants directly.

use std::any::Any;
use std::sync::Arc;

use datafusion::arrow::array::{
    ArrayBuilder, ArrayRef, BooleanBuilder, Decimal128Builder, Float32Builder, Float64Builder,
    GenericListBuilder, Int16Builder, Int32Builder, Int64Builder, Int8Builder, LargeListBuilder,
    ListBuilder, MapBuilder, OffsetSizeTrait, StringViewBuilder, StructBuilder,
    TimestampMicrosecondBuilder,
};

use datafusion::arrow::datatypes::{DataType, Field, FieldRef, Fields, TimeUnit};
use datafusion::common::{Result, ScalarValue};
use datafusion::logical_expr::{
    ColumnarValue, ReturnFieldArgs, ScalarFunctionArgs, ScalarUDF, ScalarUDFImpl, Signature,
    TypeSignature, Volatility,
};
use datafusion_common::DataFusionError;
use hamelin_lib::{f64_to_i64, parse_string_as_bool};
use parquet_variant::Variant;
use parquet_variant_compute::{VariantArray, VariantArrayBuilder};

use super::{normalize_variant_struct, write_variant_display};
use crate::struct_expansion::map_field_names;

type BoxedArrayBuilder = Box<dyn ArrayBuilder>;

/// Arrow `ArrayBuilder` wrapper around `VariantArrayBuilder`.
///
/// `VariantArrayBuilder::build` consumes `self`, but the `ArrayBuilder::finish`
/// contract is `&mut self` + reusable. Reconcile with `std::mem::replace`
/// against a fresh empty inner. Row count is tracked externally since the
/// upstream type doesn't expose `len`.
///
/// Used by `make_builder_for_type` so the generic compound builders
/// (`ListBuilder`, `StructBuilder`, `MapBuilder`) can carry a variant child
/// anywhere in the target shape.
struct VariantBuilder {
    inner: VariantArrayBuilder,
    len: usize,
}

impl VariantBuilder {
    fn with_capacity(capacity: usize) -> Self {
        Self {
            inner: VariantArrayBuilder::new(capacity),
            len: 0,
        }
    }

    fn append_variant(&mut self, variant: &Variant) {
        self.inner.append_variant(variant.clone());
        self.len += 1;
    }

    fn append_null(&mut self) {
        self.inner.append_null();
        self.len += 1;
    }

    fn build_empty() -> ArrayRef {
        Arc::new(normalize_variant_struct(
            VariantArrayBuilder::new(0).build().into(),
        ))
    }
}

impl ArrayBuilder for VariantBuilder {
    fn len(&self) -> usize {
        self.len
    }

    fn finish(&mut self) -> ArrayRef {
        let taken = std::mem::replace(&mut self.inner, VariantArrayBuilder::new(0));
        self.len = 0;
        Arc::new(normalize_variant_struct(taken.build().into()))
    }

    fn finish_cloned(&self) -> ArrayRef {
        // The upstream `VariantArrayBuilder` has no non-consuming snapshot, and
        // this wrapper is only reached through the UDF's one-shot build → finish
        // flow (never via an accumulator that would call finish_cloned mid-batch).
        // Return an empty canonical variant array as a defensive fallback if
        // something ever does invoke this path.
        Self::build_empty()
    }

    fn as_any(&self) -> &dyn Any {
        self
    }

    fn as_any_mut(&mut self) -> &mut dyn Any {
        self
    }

    fn into_box_any(self: Box<Self>) -> Box<dyn Any> {
        self
    }
}

/// UDF that casts Variant to a specific target type.
///
/// Exported for use by `hamelin_array_cast` when handling nested variant casts.
#[derive(Debug, Hash, PartialEq, Eq)]
pub struct FromVariantUdf {
    signature: Signature,
    target_type: DataType,
}

impl FromVariantUdf {
    pub fn new(target_type: DataType) -> Self {
        Self {
            signature: Signature::new(TypeSignature::Any(1), Volatility::Immutable),
            target_type,
        }
    }

    pub fn target_type(&self) -> &DataType {
        &self.target_type
    }
}

impl ScalarUDFImpl for FromVariantUdf {
    fn as_any(&self) -> &dyn Any {
        self
    }

    fn name(&self) -> &str {
        "hamelin_from_variant"
    }

    fn signature(&self) -> &Signature {
        &self.signature
    }

    fn return_type(&self, _arg_types: &[DataType]) -> Result<DataType> {
        Ok(self.target_type.clone())
    }

    fn return_field_from_args(&self, _args: ReturnFieldArgs) -> Result<Arc<Field>> {
        Ok(Arc::new(Field::new(
            self.name(),
            self.target_type.clone(),
            true,
        )))
    }

    fn invoke_with_args(&self, args: ScalarFunctionArgs) -> Result<ColumnarValue> {
        if args.args.len() != 1 {
            return Err(DataFusionError::Execution(format!(
                "hamelin_from_variant expects 1 argument, got {}",
                args.args.len()
            )));
        }

        match &args.args[0] {
            ColumnarValue::Scalar(scalar) => {
                let array = scalar.to_array_of_size(1)?;
                let result = extract_from_variant(&array, &self.target_type)?;
                let scalar = ScalarValue::try_from_array(&result, 0)?;
                Ok(ColumnarValue::Scalar(scalar))
            }
            ColumnarValue::Array(array) => {
                let result = extract_from_variant(array, &self.target_type)?;
                Ok(ColumnarValue::Array(result))
            }
        }
    }
}

pub fn from_variant_udf(target_type: DataType) -> ScalarUDF {
    ScalarUDF::new_from_impl(FromVariantUdf::new(target_type))
}

/// Extract a typed Arrow array from a variant array using the given target type.
///
/// Implements the "boxed type" rule: a variant value `v` with runtime type `T`
/// cast to target `U` behaves as `(v AS T) AS U` — so any scalar target reachable
/// through Hamelin's `string AS U` cast also works when the variant's runtime
/// type is string (double-cast via the bare-string display form).
///
/// For complex types (List, Struct, Map), custom extraction is used since
/// `shred_variant` doesn't support all types.
fn extract_from_variant(array: &ArrayRef, target_type: &DataType) -> Result<ArrayRef> {
    let variant_array =
        VariantArray::try_new(array).map_err(|e| DataFusionError::Execution(e.to_string()))?;

    // Scalar targets: one pass through the variant array, dispatching per row
    // on (target, variant runtime type) directly into the typed builder — no
    // intermediate string rendering, no Arrow cast kernel invocation.
    // Only Timestamp(Microsecond, _) is supported: hamelin_type_to_arrow
    // always maps Type::Timestamp to Timestamp(Microsecond, Some("+00:00")),
    // so this narrow match advertises exactly what the builder layer handles.
    if matches!(
        target_type,
        DataType::Utf8View
            | DataType::Boolean
            | DataType::Int8
            | DataType::Int16
            | DataType::Int32
            | DataType::Int64
            | DataType::Float32
            | DataType::Float64
            | DataType::Decimal128(_, _)
            | DataType::Timestamp(TimeUnit::Microsecond, _)
    ) {
        return variant_to_scalar_array(&variant_array, target_type);
    }

    // Map / List / LargeList use custom extraction since shred_variant doesn't
    // support them. List and LargeList thread their offset width through the
    // generic `variant_to_list`.
    if let DataType::Map(entry_field, _sorted) = target_type {
        return variant_to_map(&variant_array, entry_field);
    }
    if let DataType::List(element_field) = target_type {
        return variant_to_list::<i32>(&variant_array, element_field);
    }
    if let DataType::LargeList(element_field) = target_type {
        return variant_to_list::<i64>(&variant_array, element_field);
    }

    // Struct: row-by-row extraction so each field gets the full boxed-type rule
    // (double-cast via string, etc.). Rejects already-shredded variants
    // explicitly — the upstream ingest path in `arrow.rs` unshreds before
    // reaching the UDF, so a shredded input here means the invariant broke
    // and silently matching exact runtime types would be wrong.
    if let DataType::Struct(fields) = target_type {
        if variant_array.typed_value_field().is_some() {
            return Err(DataFusionError::Execution(
                "FromVariantUdf received a pre-shredded variant; \
                 unshredding must happen upstream (see hamelin_datafusion::arrow)"
                    .to_string(),
            ));
        }
        return variant_to_struct(&variant_array, fields);
    }

    Err(DataFusionError::Execution(format!(
        "FromVariantUdf does not support target type {target_type}"
    )))
}

/// One-pass scalar extraction: allocate the right typed builder, dispatch per row.
fn variant_to_scalar_array(
    variant_array: &VariantArray,
    target_type: &DataType,
) -> Result<ArrayRef> {
    let len = variant_array.len();
    let mut builder = make_builder_for_type(target_type, len)?;

    for i in 0..len {
        if !variant_array.is_valid(i) {
            append_null_to_builder(builder.as_mut(), target_type)?;
            continue;
        }
        let variant = variant_array.value(i);
        append_value_to_builder(builder.as_mut(), target_type, &variant)?;
    }

    Ok(builder.finish())
}

/// Convert a VariantArray containing objects to a StructArray using per-field
/// extraction. Each field is pulled through `append_value_to_builder`, which
/// applies the full boxed-type cast rule (including double-cast via display form
/// for variant-string → scalar targets).
fn variant_to_struct(variant_array: &VariantArray, fields: &Fields) -> Result<ArrayRef> {
    let len = variant_array.len();
    let mut builder = {
        let field_builders: Vec<BoxedArrayBuilder> = fields
            .iter()
            .map(|f| make_builder_for_type(f.data_type(), len))
            .collect::<Result<Vec<_>>>()?;
        StructBuilder::new(fields.clone(), field_builders)
    };

    for i in 0..variant_array.len() {
        if !variant_array.is_valid(i) {
            // Must advance all child field builders before calling append_null
            let field_builders = builder.field_builders_mut();
            for (j, field) in fields.iter().enumerate() {
                append_null_to_builder(field_builders[j].as_mut(), field.data_type())?;
            }
            builder.append_null();
            continue;
        }

        let variant = variant_array.value(i);
        match variant {
            Variant::Object(obj) => {
                let field_builders = builder.field_builders_mut();
                for (j, field) in fields.iter().enumerate() {
                    let field_value = obj.get(field.name()).unwrap_or(Variant::Null);
                    append_value_to_builder(
                        field_builders[j].as_mut(),
                        field.data_type(),
                        &field_value,
                    )?;
                }
                builder.append(true);
            }
            // Shape mismatch (non-object, non-null runtime type) and JSON null
            // both collapse to a SQL null struct per the boxed-type rule;
            // casts never error at runtime. Must advance all child field
            // builders before calling append_null to keep offsets aligned.
            _ => {
                let field_builders = builder.field_builders_mut();
                for (j, field) in fields.iter().enumerate() {
                    append_null_to_builder(field_builders[j].as_mut(), field.data_type())?;
                }
                builder.append_null();
            }
        }
    }

    Ok(Arc::new(builder.finish()))
}

/// Convert a VariantArray containing objects to a MapArray.
/// Each variant object becomes a map entry with string keys.
fn variant_to_map(variant_array: &VariantArray, entry_field: &FieldRef) -> Result<ArrayRef> {
    let DataType::Struct(entry_fields) = entry_field.data_type() else {
        return Err(DataFusionError::Execution(
            "Map entry field must be a struct".to_string(),
        ));
    };

    let key_field = entry_fields
        .iter()
        .next()
        .ok_or_else(|| DataFusionError::Execution("Map entry must have key field".to_string()))?;
    let value_field = entry_fields
        .iter()
        .nth(1)
        .ok_or_else(|| DataFusionError::Execution("Map entry must have value field".to_string()))?;

    // Build the map using singular field names (key, value) to match our convention
    let len = variant_array.len();
    let key_builder = make_builder_for_type(key_field.data_type(), len)?;
    let value_builder = make_builder_for_type(value_field.data_type(), len)?;
    let mut map_builder = MapBuilder::new(Some(map_field_names()), key_builder, value_builder);

    for i in 0..variant_array.len() {
        if !variant_array.is_valid(i) {
            map_builder.append(false)?;
            continue;
        }

        let variant = variant_array.value(i);
        match variant {
            Variant::Object(obj) => {
                for (key, val) in obj.iter() {
                    append_value_to_builder(
                        map_builder.keys(),
                        key_field.data_type(),
                        &Variant::String(key),
                    )?;
                    append_value_to_builder(map_builder.values(), value_field.data_type(), &val)?;
                }
                map_builder.append(true)?;
            }
            // Shape mismatch (non-object runtime type) → null per the boxed-type
            // rule; casts never error at runtime.
            _ => {
                map_builder.append(false)?;
            }
        }
    }

    Ok(Arc::new(map_builder.finish()))
}

/// Convert a VariantArray containing lists to a (Large)ListArray.
/// Generic over `OffsetSize` so the same body serves both `List` (i32) and
/// `LargeList` (i64) targets.
fn variant_to_list<O: OffsetSizeTrait>(
    variant_array: &VariantArray,
    element_field: &FieldRef,
) -> Result<ArrayRef> {
    let len = variant_array.len();
    let inner_builder = make_builder_for_type(element_field.data_type(), len)?;
    let mut list_builder = GenericListBuilder::<O, _>::new(inner_builder);

    for i in 0..variant_array.len() {
        if !variant_array.is_valid(i) {
            list_builder.append(false);
            continue;
        }

        let variant = variant_array.value(i);
        match variant {
            Variant::List(list) => {
                for elem in list.iter() {
                    append_value_to_builder(
                        list_builder.values().as_mut(),
                        element_field.data_type(),
                        &elem,
                    )?;
                }
                list_builder.append(true);
            }
            // Shape mismatch (non-list runtime type) → null per the boxed-type
            // rule; casts never error at runtime.
            _ => {
                list_builder.append(false);
            }
        }
    }

    Ok(Arc::new(list_builder.finish()))
}

/// Create a boxed array builder for the given data type, preallocating for
/// `row_capacity` rows so the per-batch allocations don't have to grow.
///
/// Compound builders (List/Struct/Map) pass `row_capacity` to their children
/// untouched: for fixed-shape nested fields the per-row count is the same as
/// the parent's. For variable-fan-out children (map values where each row
/// can have many entries) the hint is a lower bound — Arrow will still grow
/// past it if needed, but we save the initial doublings.
fn make_builder_for_type(data_type: &DataType, row_capacity: usize) -> Result<BoxedArrayBuilder> {
    // Canonical variant struct — route through VariantBuilder so compound
    // targets can carry a variant leaf anywhere in their shape. Handled before
    // the match since a variant is structurally `DataType::Struct(...)` with
    // specific field shape, which the generic `DataType::Struct` arm wouldn't
    // distinguish.
    if super::is_variant_data_type(data_type) {
        return Ok(Box::new(VariantBuilder::with_capacity(row_capacity)));
    }
    match data_type {
        DataType::Boolean => Ok(Box::new(BooleanBuilder::with_capacity(row_capacity))),
        DataType::Int8 => Ok(Box::new(Int8Builder::with_capacity(row_capacity))),
        DataType::Int16 => Ok(Box::new(Int16Builder::with_capacity(row_capacity))),
        DataType::Int32 => Ok(Box::new(Int32Builder::with_capacity(row_capacity))),
        DataType::Int64 => Ok(Box::new(Int64Builder::with_capacity(row_capacity))),
        DataType::Float32 => Ok(Box::new(Float32Builder::with_capacity(row_capacity))),
        DataType::Float64 => Ok(Box::new(Float64Builder::with_capacity(row_capacity))),
        DataType::Utf8View => Ok(Box::new(StringViewBuilder::with_capacity(row_capacity))),
        DataType::Timestamp(TimeUnit::Microsecond, tz) => Ok(Box::new(
            TimestampMicrosecondBuilder::with_capacity(row_capacity).with_timezone_opt(tz.clone()),
        )),
        DataType::Decimal128(precision, scale) => Ok(Box::new(
            Decimal128Builder::with_capacity(row_capacity)
                .with_data_type(DataType::Decimal128(*precision, *scale)),
        )),
        DataType::List(element_field) => {
            let inner_builder = make_builder_for_type(element_field.data_type(), row_capacity)?;
            Ok(Box::new(ListBuilder::new(inner_builder)))
        }
        DataType::LargeList(element_field) => {
            let inner_builder = make_builder_for_type(element_field.data_type(), row_capacity)?;
            Ok(Box::new(LargeListBuilder::new(inner_builder)))
        }
        DataType::Struct(fields) => {
            let field_builders: Vec<BoxedArrayBuilder> = fields
                .iter()
                .map(|f| make_builder_for_type(f.data_type(), row_capacity))
                .collect::<Result<Vec<_>>>()?;
            Ok(Box::new(StructBuilder::new(fields.clone(), field_builders)))
        }
        DataType::Map(entry_field, _) => {
            let DataType::Struct(entry_fields) = entry_field.data_type() else {
                return Err(DataFusionError::Execution(
                    "Map entry must be a struct".to_string(),
                ));
            };
            let key_field = entry_fields
                .iter()
                .next()
                .ok_or_else(|| DataFusionError::Execution("Map must have key field".to_string()))?;
            let value_field = entry_fields.iter().nth(1).ok_or_else(|| {
                DataFusionError::Execution("Map must have value field".to_string())
            })?;
            let key_builder = make_builder_for_type(key_field.data_type(), row_capacity)?;
            let value_builder = make_builder_for_type(value_field.data_type(), row_capacity)?;
            Ok(Box::new(MapBuilder::new(
                Some(map_field_names()),
                key_builder,
                value_builder,
            )))
        }
        other => Err(DataFusionError::Execution(format!(
            "Unsupported element type for variant list/map conversion: {other}"
        ))),
    }
}

/// Append a variant value to a builder of the given Arrow type, implementing the
/// Hamelin "boxed type" cast rule in one pass per row. No intermediate string
/// allocation or Arrow cast-kernel invocation: each `(target_type, variant)`
/// pair dispatches directly to the matching typed builder.
///
/// Conversions supported (matching Hamelin's scalar cast matrix):
/// - Identity: variant holds the exact target type (e.g. `Variant::Int64` → `Int64`).
/// - Numeric widening/narrowing: int→int, int→float, float→float via the upstream
///   `Variant::as_int*` / `as_f64` helpers which handle in-variant widening.
/// - Boolean → numeric: `BooleanTrue`/`BooleanFalse` emits `1`/`0`.
/// - Any primitive → string (`Utf8View`): writes the bare display form.
/// - Variant-string → scalar: `str::parse::<T>()` per Hamelin's `string → T` rule;
///   null on parse failure.
/// - Object/array → scalar and every other unsupported pair: null.
/// - List/Struct/Map targets: recurse into the builder's child builders.
fn append_value_to_builder(
    builder: &mut dyn ArrayBuilder,
    data_type: &DataType,
    variant: &Variant,
) -> Result<()> {
    // Variant target: store the row in canonical metadata/value form via
    // VariantBuilder. Handled before the match since a variant is structurally
    // `DataType::Struct(...)` with a specific field shape that the generic
    // Struct arm wouldn't distinguish.
    //
    // Variant::Null is a *value* (the JSON null literal), not a SQL null —
    // preserve it. SQL nulls arrive through append_null_to_builder instead.
    if super::is_variant_data_type(data_type) {
        let variant_builder = builder
            .as_any_mut()
            .downcast_mut::<VariantBuilder>()
            .ok_or_else(|| DataFusionError::Execution("Type mismatch for variant".to_string()))?;
        variant_builder.append_variant(variant);
        return Ok(());
    }
    match data_type {
        DataType::Boolean => {
            let b = builder
                .as_any_mut()
                .downcast_mut::<BooleanBuilder>()
                .ok_or_else(|| DataFusionError::Execution("Type mismatch".to_string()))?;
            // Boxed-type rule for boolean:
            // - bool variants pass through
            // - int / double / decimal variants follow IntToBoolean/DoubleToBoolean
            //   semantics (zero → false, any other value → true)
            // - string variants double-cast using the same acceptance set as
            //   Arrow's string→bool kernel (case-insensitive, trim, accepts
            //   "true"/"false"/"1"/"0"/"yes"/"no"/"on"/"off" and prefixes).
            // - everything else → null
            let parsed = match variant {
                Variant::BooleanTrue => Some(true),
                Variant::BooleanFalse => Some(false),
                Variant::String(s) => parse_string_as_bool(s),
                Variant::ShortString(s) => parse_string_as_bool(s.as_ref()),
                _ => variant_to_f64(variant).map(|f| f != 0.0),
            };
            match parsed {
                Some(v) => b.append_value(v),
                None => b.append_null(),
            }
        }
        DataType::Int8 => {
            let b = builder
                .as_any_mut()
                .downcast_mut::<Int8Builder>()
                .ok_or_else(|| DataFusionError::Execution("Type mismatch".to_string()))?;
            match variant_to_i64(variant).and_then(|n| i8::try_from(n).ok()) {
                Some(v) => b.append_value(v),
                None => b.append_null(),
            }
        }
        DataType::Int16 => {
            let b = builder
                .as_any_mut()
                .downcast_mut::<Int16Builder>()
                .ok_or_else(|| DataFusionError::Execution("Type mismatch".to_string()))?;
            match variant_to_i64(variant).and_then(|n| i16::try_from(n).ok()) {
                Some(v) => b.append_value(v),
                None => b.append_null(),
            }
        }
        DataType::Int32 => {
            let b = builder
                .as_any_mut()
                .downcast_mut::<Int32Builder>()
                .ok_or_else(|| DataFusionError::Execution("Type mismatch".to_string()))?;
            match variant_to_i64(variant).and_then(|n| i32::try_from(n).ok()) {
                Some(v) => b.append_value(v),
                None => b.append_null(),
            }
        }
        DataType::Int64 => {
            let b = builder
                .as_any_mut()
                .downcast_mut::<Int64Builder>()
                .ok_or_else(|| DataFusionError::Execution("Type mismatch".to_string()))?;
            match variant_to_i64(variant) {
                Some(v) => b.append_value(v),
                None => b.append_null(),
            }
        }
        DataType::Float32 => {
            let b = builder
                .as_any_mut()
                .downcast_mut::<Float32Builder>()
                .ok_or_else(|| DataFusionError::Execution("Type mismatch".to_string()))?;
            match variant_to_f64(variant) {
                Some(v) => b.append_value(v as f32),
                None => b.append_null(),
            }
        }
        DataType::Float64 => {
            let b = builder
                .as_any_mut()
                .downcast_mut::<Float64Builder>()
                .ok_or_else(|| DataFusionError::Execution("Type mismatch".to_string()))?;
            match variant_to_f64(variant) {
                Some(v) => b.append_value(v),
                None => b.append_null(),
            }
        }
        DataType::Utf8View => {
            let b = builder
                .as_any_mut()
                .downcast_mut::<StringViewBuilder>()
                .ok_or_else(|| DataFusionError::Execution("Type mismatch".to_string()))?;
            append_variant_to_stringview(b, variant)?;
        }
        DataType::Timestamp(TimeUnit::Microsecond, _) => {
            let b = builder
                .as_any_mut()
                .downcast_mut::<TimestampMicrosecondBuilder>()
                .ok_or_else(|| DataFusionError::Execution("Type mismatch".to_string()))?;
            // Boxed-type rule for timestamp: only string variants parse into a
            // timestamp (matching Hamelin's `string AS timestamp`). Other runtime
            // types → null, consistent with the ticket rule that the underlying
            // primitive must have a defined `T AS timestamp` cast.
            let micros = match variant {
                Variant::String(s) => super::parse_timestamp_micros(s).ok(),
                Variant::ShortString(s) => super::parse_timestamp_micros(s.as_ref()).ok(),
                _ => None,
            };
            match micros {
                Some(v) => b.append_value(v),
                None => b.append_null(),
            }
        }
        DataType::Decimal128(precision, scale) => {
            let b = builder
                .as_any_mut()
                .downcast_mut::<Decimal128Builder>()
                .ok_or_else(|| DataFusionError::Execution("Type mismatch".to_string()))?;
            match variant_to_decimal128(variant, *precision, *scale) {
                Some(unscaled) => b.append_value(unscaled),
                None => b.append_null(),
            }
        }
        DataType::List(element_field) => {
            append_variant_to_list_builder::<i32>(builder, element_field.data_type(), variant)?;
        }
        DataType::LargeList(element_field) => {
            append_variant_to_list_builder::<i64>(builder, element_field.data_type(), variant)?;
        }
        DataType::Struct(fields) => {
            let struct_builder = builder
                .as_any_mut()
                .downcast_mut::<StructBuilder>()
                .ok_or_else(|| {
                    DataFusionError::Execution("Type mismatch for struct".to_string())
                })?;
            match variant {
                Variant::Object(obj) => {
                    let field_builders = struct_builder.field_builders_mut();
                    for (i, field) in fields.iter().enumerate() {
                        let field_builder = field_builders.get_mut(i).ok_or_else(|| {
                            DataFusionError::Execution(format!(
                                "Missing field builder for {}",
                                field.name()
                            ))
                        })?;
                        let field_value = obj.get(field.name()).unwrap_or(Variant::Null);
                        append_value_to_builder(
                            field_builder.as_mut(),
                            field.data_type(),
                            &field_value,
                        )?;
                    }
                    struct_builder.append(true);
                }
                _ => {
                    // Null struct: advance all child builders to keep offsets aligned.
                    let field_builders = struct_builder.field_builders_mut();
                    for (i, field) in fields.iter().enumerate() {
                        append_null_to_builder(field_builders[i].as_mut(), field.data_type())?;
                    }
                    struct_builder.append_null();
                }
            }
        }
        DataType::Map(entry_field, _) => {
            let DataType::Struct(entry_fields) = entry_field.data_type() else {
                return Err(DataFusionError::Execution(
                    "Map entry must be a struct".to_string(),
                ));
            };
            let key_field = entry_fields
                .iter()
                .next()
                .ok_or_else(|| DataFusionError::Execution("Map must have key field".to_string()))?;
            let value_field = entry_fields.iter().nth(1).ok_or_else(|| {
                DataFusionError::Execution("Map must have value field".to_string())
            })?;
            let map_builder = builder
                .as_any_mut()
                .downcast_mut::<MapBuilder<BoxedArrayBuilder, BoxedArrayBuilder>>()
                .ok_or_else(|| DataFusionError::Execution("Type mismatch for map".to_string()))?;
            match variant {
                Variant::Object(obj) => {
                    for (key, val) in obj.iter() {
                        append_value_to_builder(
                            map_builder.keys(),
                            key_field.data_type(),
                            &Variant::String(key),
                        )?;
                        append_value_to_builder(
                            map_builder.values(),
                            value_field.data_type(),
                            &val,
                        )?;
                    }
                    map_builder.append(true)?;
                }
                _ => map_builder.append(false)?,
            }
        }
        _ => {
            return Err(DataFusionError::Execution(format!(
                "Unsupported target type in append_value_to_builder: {data_type}"
            )));
        }
    }
    Ok(())
}

/// Extract an integer value from a variant per the Hamelin scalar cast rule:
/// - Int variants widen to i64.
/// - Bool variants emit 1/0.
/// - Float/Double variants truncate toward zero (matching native `double AS int`).
/// - Decimal variants truncate toward zero (matching native `decimal AS int`).
/// - String variants parse via `str::parse::<i64>()`.
/// - Everything else (null, object, array, timestamp, etc.) → None.
fn variant_to_i64(variant: &Variant) -> Option<i64> {
    match variant {
        Variant::BooleanTrue => Some(1),
        Variant::BooleanFalse => Some(0),
        Variant::String(s) => s.parse::<i64>().ok(),
        Variant::ShortString(s) => s.as_ref().parse::<i64>().ok(),
        Variant::Float(f) => f64_to_i64(*f as f64),
        Variant::Double(f) => f64_to_i64(*f),
        // Decimal variants: integer / 10^scale truncates toward zero in Rust
        // signed-integer division, matching native DecimalToInt semantics.
        // Upstream `Variant::as_int64` only accepts decimals with scale 0,
        // so we handle non-zero scale here.
        Variant::Decimal4(d) => decimal_to_i64(d.integer() as i128, d.scale()),
        Variant::Decimal8(d) => decimal_to_i64(d.integer() as i128, d.scale()),
        Variant::Decimal16(d) => decimal_to_i64(d.integer(), d.scale()),
        _ => variant.as_int64(),
    }
}

/// Truncate a `(unscaled, scale)` decimal toward zero into i64. Returns None
/// if the truncated quotient overflows i64.
fn decimal_to_i64(integer: i128, scale: u8) -> Option<i64> {
    let divisor = 10_i128.checked_pow(scale as u32)?;
    let quotient = integer / divisor;
    quotient.try_into().ok()
}

/// Extract an f64 from a variant per the Hamelin scalar cast rule:
/// - Numeric variants use the upstream `as_f64` (int→float, float widening).
/// - Bool variants emit 1.0/0.0.
/// - Decimal variants compute integer / 10^scale as f64 (matching native
///   `decimal AS double`); upstream `as_f64` only handles ints/floats.
/// - String variants parse via `str::parse::<f64>()`.
/// - Everything else → None.
fn variant_to_f64(variant: &Variant) -> Option<f64> {
    match variant {
        Variant::BooleanTrue => Some(1.0),
        Variant::BooleanFalse => Some(0.0),
        Variant::String(s) => s.parse::<f64>().ok(),
        Variant::ShortString(s) => s.as_ref().parse::<f64>().ok(),
        Variant::Decimal4(d) => Some(d.integer() as f64 / 10f64.powi(d.scale() as i32)),
        Variant::Decimal8(d) => Some(d.integer() as f64 / 10f64.powi(d.scale() as i32)),
        Variant::Decimal16(d) => Some(d.integer() as f64 / 10f64.powi(d.scale() as i32)),
        _ => variant.as_f64(),
    }
}

/// Extract a Decimal128 unscaled value from a variant per the Hamelin scalar
/// cast rule, scaling/rescaling to match the target `(precision, scale)`:
/// - Int variants: `n × 10^scale`, bounds-checked against `precision`.
/// - Bool variants: `0` or `10^scale`, matching native `BooleanToDecimal`.
/// - Decimal variants: rescale `(integer, src_scale)` to `target_scale` —
///   widening multiplies; narrowing truncates toward zero. All in i128 so
///   precision is preserved up to Decimal16's 38-digit range.
/// - String variants: delegate to Arrow's `parse_decimal::<Decimal128Type>`,
///   which handles arbitrary precision strings exactly as Hamelin's native
///   `string AS decimal` cast does.
/// - Float/Double variants: f64 → scaled i128 with `f.round()`. Inherent
///   precision loss past ~17 significant digits — same precision profile as
///   any double-rooted path.
/// - Everything else → None.
fn variant_to_decimal128(variant: &Variant, precision: u8, scale: i8) -> Option<i128> {
    match variant {
        Variant::BooleanTrue => scale_int_to_decimal(1i128, precision, scale),
        Variant::BooleanFalse => scale_int_to_decimal(0i128, precision, scale),
        Variant::Int8(n) => scale_int_to_decimal(*n as i128, precision, scale),
        Variant::Int16(n) => scale_int_to_decimal(*n as i128, precision, scale),
        Variant::Int32(n) => scale_int_to_decimal(*n as i128, precision, scale),
        Variant::Int64(n) => scale_int_to_decimal(*n as i128, precision, scale),
        Variant::Decimal4(d) => {
            rescale_decimal_to_decimal(d.integer() as i128, d.scale() as i8, precision, scale)
        }
        Variant::Decimal8(d) => {
            rescale_decimal_to_decimal(d.integer() as i128, d.scale() as i8, precision, scale)
        }
        Variant::Decimal16(d) => {
            rescale_decimal_to_decimal(d.integer(), d.scale() as i8, precision, scale)
        }
        Variant::Float(f) => f64_to_decimal128(*f as f64, precision, scale),
        Variant::Double(f) => f64_to_decimal128(*f, precision, scale),
        Variant::String(s) => parse_string_to_decimal128(s, precision, scale),
        Variant::ShortString(s) => parse_string_to_decimal128(s.as_ref(), precision, scale),
        _ => None,
    }
}

/// Scale an integer up to a Decimal128 unscaled value, bounds-checked against
/// the target precision. Returns None on overflow.
fn scale_int_to_decimal(n: i128, precision: u8, scale: i8) -> Option<i128> {
    if scale < 0 {
        return None;
    }
    let factor = 10_i128.checked_pow(scale as u32)?;
    let unscaled = n.checked_mul(factor)?;
    decimal_fits_precision(unscaled, precision).then_some(unscaled)
}

/// Rescale `(integer, src_scale)` to `target_scale`, returning the unscaled
/// i128 value bounds-checked against `precision`. When narrowing
/// (target_scale < src_scale), truncates toward zero — matching Arrow's
/// `cast_decimal_to_decimal_same_type` (`div_checked`). When widening, exact.
fn rescale_decimal_to_decimal(
    integer: i128,
    src_scale: i8,
    precision: u8,
    target_scale: i8,
) -> Option<i128> {
    let unscaled = if target_scale >= src_scale {
        let diff = (target_scale - src_scale) as u32;
        let factor = 10_i128.checked_pow(diff)?;
        integer.checked_mul(factor)?
    } else {
        let diff = (src_scale - target_scale) as u32;
        let divisor = 10_i128.checked_pow(diff)?;
        integer / divisor
    };
    decimal_fits_precision(unscaled, precision).then_some(unscaled)
}

/// Convert an f64 to a Decimal128 unscaled value with rounding. NaN / ±Inf and
/// values that overflow the target precision return None.
fn f64_to_decimal128(f: f64, precision: u8, scale: i8) -> Option<i128> {
    if !f.is_finite() {
        return None;
    }
    if scale < 0 {
        return None;
    }
    let scaled = f * 10f64.powi(scale as i32);
    let rounded = scaled.round();
    if !rounded.is_finite() || rounded < i128::MIN as f64 || rounded > i128::MAX as f64 {
        return None;
    }
    let unscaled = rounded as i128;
    decimal_fits_precision(unscaled, precision).then_some(unscaled)
}

/// Parse a string into a Decimal128 unscaled value via Arrow's
/// `parse_decimal::<Decimal128Type>`, which handles arbitrary precision and
/// matches Hamelins native `string AS decimal` semantics exactly.
fn parse_string_to_decimal128(s: &str, precision: u8, scale: i8) -> Option<i128> {
    use datafusion::arrow::compute::kernels::cast_utils::parse_decimal;
    use datafusion::arrow::datatypes::Decimal128Type;
    parse_decimal::<Decimal128Type>(s.trim(), precision, scale).ok()
}

/// Check that an unscaled Decimal128 value fits within `precision` digits.
fn decimal_fits_precision(unscaled: i128, precision: u8) -> bool {
    let max = 10_i128
        .checked_pow(precision as u32)
        .map(|p| p - 1)
        .unwrap_or(i128::MAX);
    unscaled.unsigned_abs() <= max as u128
}

/// Write the bare display form of a variant directly into a StringViewBuilder.
/// Object/array/null/binary variants and non-finite floats (NaN / ±Inf) produce
/// null. Strings copy unwrapped; other scalars format on-the-fly via
/// `write_variant_display`.
fn append_variant_to_stringview(builder: &mut StringViewBuilder, variant: &Variant) -> Result<()> {
    match variant {
        Variant::String(s) => builder.append_value(s),
        Variant::ShortString(s) => builder.append_value(s.as_ref()),
        Variant::Null | Variant::Binary(_) | Variant::List(_) | Variant::Object(_) => {
            builder.append_null()
        }
        Variant::Float(f) if !f.is_finite() => builder.append_null(),
        Variant::Double(f) if !f.is_finite() => builder.append_null(),
        _ => {
            let mut scratch = String::new();
            write_variant_display(&mut scratch, variant).map_err(|e| {
                DataFusionError::Execution(format!("failed to format variant: {e}"))
            })?;
            builder.append_value(&scratch);
        }
    }
    Ok(())
}

/// Append a variant value to a (Large)ListBuilder. Generic over offset width so
/// one helper serves both `List` (i32) and `LargeList` (i64) targets.
fn append_variant_to_list_builder<O: OffsetSizeTrait>(
    builder: &mut dyn ArrayBuilder,
    element_type: &DataType,
    variant: &Variant,
) -> Result<()> {
    let list_builder = builder
        .as_any_mut()
        .downcast_mut::<GenericListBuilder<O, BoxedArrayBuilder>>()
        .ok_or_else(|| DataFusionError::Execution("Type mismatch for list".to_string()))?;
    match variant {
        Variant::List(list) => {
            for elem in list.iter() {
                append_value_to_builder(list_builder.values().as_mut(), element_type, &elem)?;
            }
            list_builder.append(true);
        }
        _ => list_builder.append(false),
    }
    Ok(())
}

/// Append a null value to a builder of the given type.
fn append_null_to_builder(builder: &mut dyn ArrayBuilder, data_type: &DataType) -> Result<()> {
    // Variant target: handled before the match since a variant is structurally
    // `DataType::Struct(...)` with a specific field shape that the generic
    // Struct arm wouldn't distinguish.
    if super::is_variant_data_type(data_type) {
        builder
            .as_any_mut()
            .downcast_mut::<VariantBuilder>()
            .ok_or_else(|| DataFusionError::Execution("Type mismatch".to_string()))?
            .append_null();
        return Ok(());
    }
    match data_type {
        DataType::Boolean => builder
            .as_any_mut()
            .downcast_mut::<BooleanBuilder>()
            .ok_or_else(|| DataFusionError::Execution("Type mismatch".to_string()))?
            .append_null(),
        DataType::Int8 => builder
            .as_any_mut()
            .downcast_mut::<Int8Builder>()
            .ok_or_else(|| DataFusionError::Execution("Type mismatch".to_string()))?
            .append_null(),
        DataType::Int16 => builder
            .as_any_mut()
            .downcast_mut::<Int16Builder>()
            .ok_or_else(|| DataFusionError::Execution("Type mismatch".to_string()))?
            .append_null(),
        DataType::Int32 => builder
            .as_any_mut()
            .downcast_mut::<Int32Builder>()
            .ok_or_else(|| DataFusionError::Execution("Type mismatch".to_string()))?
            .append_null(),
        DataType::Int64 => builder
            .as_any_mut()
            .downcast_mut::<Int64Builder>()
            .ok_or_else(|| DataFusionError::Execution("Type mismatch".to_string()))?
            .append_null(),
        DataType::Float32 => builder
            .as_any_mut()
            .downcast_mut::<Float32Builder>()
            .ok_or_else(|| DataFusionError::Execution("Type mismatch".to_string()))?
            .append_null(),
        DataType::Float64 => builder
            .as_any_mut()
            .downcast_mut::<Float64Builder>()
            .ok_or_else(|| DataFusionError::Execution("Type mismatch".to_string()))?
            .append_null(),
        DataType::Utf8View => builder
            .as_any_mut()
            .downcast_mut::<StringViewBuilder>()
            .ok_or_else(|| DataFusionError::Execution("Type mismatch".to_string()))?
            .append_null(),
        DataType::Timestamp(TimeUnit::Microsecond, _) => builder
            .as_any_mut()
            .downcast_mut::<TimestampMicrosecondBuilder>()
            .ok_or_else(|| DataFusionError::Execution("Type mismatch".to_string()))?
            .append_null(),
        DataType::Decimal128(_, _) => builder
            .as_any_mut()
            .downcast_mut::<Decimal128Builder>()
            .ok_or_else(|| DataFusionError::Execution("Type mismatch".to_string()))?
            .append_null(),
        DataType::List(_) => builder
            .as_any_mut()
            .downcast_mut::<ListBuilder<BoxedArrayBuilder>>()
            .ok_or_else(|| DataFusionError::Execution("Type mismatch".to_string()))?
            .append(false),
        DataType::LargeList(_) => builder
            .as_any_mut()
            .downcast_mut::<LargeListBuilder<BoxedArrayBuilder>>()
            .ok_or_else(|| DataFusionError::Execution("Type mismatch".to_string()))?
            .append(false),
        DataType::Struct(fields) => {
            let struct_builder = builder
                .as_any_mut()
                .downcast_mut::<StructBuilder>()
                .ok_or_else(|| DataFusionError::Execution("Type mismatch".to_string()))?;
            // Must advance all child field builders before calling append_null
            let field_builders = struct_builder.field_builders_mut();
            for (i, field) in fields.iter().enumerate() {
                append_null_to_builder(field_builders[i].as_mut(), field.data_type())?;
            }
            struct_builder.append_null();
        }
        DataType::Map(_, _) => builder
            .as_any_mut()
            .downcast_mut::<MapBuilder<BoxedArrayBuilder, BoxedArrayBuilder>>()
            .ok_or_else(|| DataFusionError::Execution("Type mismatch".to_string()))?
            .append(false)?,
        _ => {
            return Err(DataFusionError::Execution(format!(
                "Cannot append null to unsupported type: {data_type}"
            )));
        }
    }
    Ok(())
}
