hamelin_duckdb 0.3.6

use std::fmt::{Debug, Formatter};
use std::path::PathBuf;
use std::sync::Arc;

use anyhow::{anyhow, bail};
use async_trait::async_trait;
use base64::prelude::*;
use duckdb::arrow::array::{Array, AsArray};
use duckdb::arrow::datatypes::{
    ArrowDictionaryKeyType, ArrowNativeType, ArrowPrimitiveType, RunEndIndexType,
};
use duckdb::arrow::datatypes::{
    DataType, Date32Type, Date64Type, Decimal128Type, Decimal256Type, DecimalType,
    DurationMicrosecondType, DurationMillisecondType, DurationNanosecondType, DurationSecondType,
    Float16Type, Float32Type, Float64Type, Int16Type, Int32Type, Int64Type, Int8Type,
    IntervalDayTimeType, IntervalYearMonthType, TimeUnit, TimestampMicrosecondType,
    TimestampMillisecondType, TimestampNanosecondType, TimestampSecondType, UInt16Type, UInt32Type,
    UInt64Type, UInt8Type,
};
use duckdb::Connection;

use hamelin_executor::executor::{Executor, ExecutorError};
use hamelin_executor::results::ResultSet;
use hamelin_lib::translation::{DMLTranslation, QueryTranslation, StatementTranslation};
use hamelin_lib::TimeRange;
use serde_json::Value;

use hamelin_lib::catalog::{Catalog, HamelinType};
use hamelin_lib::provider::EnvironmentProvider;
use hamelin_lib::sql::expression::identifier::Identifier;
use hamelin_lib::sql::query::TableReference;
use hamelin_lib::translation::env::Environment;

pub const VERSION: &str = env!("CARGO_PKG_VERSION");

pub struct DuckDB {
    /// Path to the duckdb database. If not specified, will use memory database.
    path: Option<PathBuf>,
}

#[derive(Debug)]
pub struct InformationSchemaColumns {
    pub table_catalog: String,
    pub table_schema: String,
    pub table_name: String,
    pub column_name: String,
    pub ordinal_position: i32,
    pub data_type: String,
}

#[derive(Debug)]
pub struct InformationSchemaDatasets {
    pub table_schema: String,
    pub table_name: String,
}

#[derive(Debug)]
pub struct Describe {
    pub column_name: String,
    pub column_type: String,
    pub null_: String,
    pub key: String,
    pub default: String,
    pub extra: String,
}

impl DuckDB {
    pub fn from_path<P: Into<PathBuf>>(path: P) -> Self {
        Self {
            path: Some(path.into()),
        }
    }

    pub fn in_memory() -> Self {
        Self { path: None }
    }

    pub fn connection(&self) -> Result<Connection, ExecutorError> {
        match &self.path {
            Some(p) => Connection::open(p),
            None => Connection::open_in_memory(),
        }
        .map_err(|e| ExecutorError::ConnectionError(e.into()))
    }
}

fn map_err(err: duckdb::Error) -> ExecutorError {
    match err {
        e @ duckdb::Error::FromSqlConversionFailure(_, _, _)
        | e @ duckdb::Error::Utf8Error(_)
        | e @ duckdb::Error::NulError(_)
        | e @ duckdb::Error::InvalidPath(_)
        | e @ duckdb::Error::ExecuteReturnedResults
        | e @ duckdb::Error::QueryReturnedNoRows
        | e @ duckdb::Error::ArrowTypeToDuckdbType(_, _)
        | e @ duckdb::Error::StatementChangedRows(_)
        | e @ duckdb::Error::InvalidColumnIndex(_)
        | e @ duckdb::Error::InvalidColumnName(_)
        | e @ duckdb::Error::InvalidColumnType(_, _, _)
        | e @ duckdb::Error::ToSqlConversionFailure(_)
        | e @ duckdb::Error::IntegralValueOutOfRange(_, _) => {
            ExecutorError::UnexpectedResultSet(e.into())
        }
        e @ duckdb::Error::InvalidQuery
        | e @ duckdb::Error::InvalidParameterCount(_, _)
        | e @ duckdb::Error::MultipleStatement
        | e @ duckdb::Error::InvalidParameterName(_) => ExecutorError::QueryError(e.into()),
        e @ _ => ExecutorError::ServerError(e.into()),
    }
}

#[async_trait]
impl Executor for DuckDB {
    async fn resolve_time_range(
        &self,
        _query: QueryTranslation,
    ) -> Result<TimeRange, ExecutorError> {
        todo!()
    }

    async fn execute_query(
        &self,
        translation: QueryTranslation,
    ) -> Result<ResultSet, ExecutorError> {
        let conn = self.connection()?;
        let mut stmt = conn
            .prepare(&translation.translation.sql)
            .map_err(map_err)?;

        let mut all_rows = vec![];
        for batch in stmt.query_arrow([]).map_err(map_err)? {
            let mut result_columns = Vec::with_capacity(batch.num_columns());
            for (i, col) in translation.translation.columns.iter().enumerate() {
                result_columns.push(
                    translate_arrow_array_value(batch.column(i), &col.typ)
                        .map_err(|e| ExecutorError::UnexpectedResultSet(e.into()))?,
                );
            }
            let mut rows = transpose(result_columns);

            all_rows.append(&mut rows);
        }

        Ok(ResultSet::new(translation.translation.columns, all_rows))
    }

    async fn execute_dml(&self, dml: DMLTranslation) -> Result<usize, ExecutorError> {
        let conn = self.connection()?;
        conn.execute(&dml.translation.sql, []).map_err(map_err)
    }

    async fn execute(&self, statement: StatementTranslation) -> Result<(), ExecutorError> {
        let statement = match statement {
            StatementTranslation::Query(query_translation) => query_translation.translation.sql,
            StatementTranslation::DML(dmltranslation) => dmltranslation.translation.sql,
        };

        let conn = self.connection()?;
        conn.execute(&statement, []).map_err(map_err).map(|_| ())
    }

    async fn reflect_catalog(&self) -> Result<Catalog, ExecutorError> {
        todo!()
    }
}

impl Debug for DuckDB {
    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
        match &self.path {
            Some(p) => write!(f, "DuckDB({:?})", p.to_str().unwrap_or_default()),
            None => write!(f, "DuckDB(memory)"),
        }
    }
}

impl EnvironmentProvider for DuckDB {
    fn reflect_columns(&self, _table_reference: TableReference) -> anyhow::Result<Environment> {
        todo!()
    }

    fn reflect_datasets(&self) -> anyhow::Result<Vec<Identifier>> {
        todo!()
    }
}

fn transpose(columnar: Vec<Vec<Value>>) -> Vec<Vec<Value>> {
    if columnar.is_empty() {
        return columnar;
    }

    let mut transposed = vec![Vec::with_capacity(columnar.len()); columnar[0].len()];

    for col in columnar {
        for (i, value) in col.into_iter().enumerate() {
            transposed[i].push(value);
        }
    }

    transposed
}

// Helper function to handle common iteration pattern with null checking
fn process_array_with_null_check<T, F>(
    array: &Arc<dyn Array>,
    value_extractor: F,
) -> anyhow::Result<Vec<Value>>
where
    F: Fn(usize) -> anyhow::Result<T>,
    T: Into<Value>,
{
    let mut results = Vec::new();
    for i in 0..array.len() {
        if array.is_null(i) {
            results.push(Value::Null);
        } else {
            results.push(value_extractor(i)?.into());
        }
    }
    Ok(results)
}

// Helper function for primitive integer arrays
fn process_primitive_int_array<T>(array: &Arc<dyn Array>) -> anyhow::Result<Vec<Value>>
where
    T: ArrowPrimitiveType,
    T::Native: Into<serde_json::Number>,
{
    let primitive_array = array.as_primitive::<T>();
    process_array_with_null_check(array, |i| {
        Ok(Value::Number(primitive_array.value(i).into()))
    })
}

// Helper function for primitive float arrays
fn process_primitive_float_array<T>(array: &Arc<dyn Array>) -> anyhow::Result<Vec<Value>>
where
    T: ArrowPrimitiveType,
    T::Native: Into<f64>,
{
    let primitive_array = array.as_primitive::<T>();
    process_array_with_null_check(array, |i| {
        let float_val: f64 = primitive_array.value(i).into();
        let json_number = serde_json::Number::from_f64(float_val)
            .ok_or_else(|| anyhow!("Invalid float value"))?;
        Ok(Value::Number(json_number))
    })
}

// Helper function to handle dictionary arrays with a specific key type
fn handle_dictionary<K: ArrowDictionaryKeyType>(
    array: &Arc<dyn Array>,
    hamelin_type: &HamelinType,
) -> Option<anyhow::Result<Vec<Value>>> {
    let dict_array = array.as_dictionary_opt::<K>()?;
    let keys = dict_array.keys();
    let values = dict_array.values();
    let translated_values = match translate_arrow_array_value(values, hamelin_type) {
        Ok(v) => v,
        Err(e) => return Some(Err(e)),
    };

    Some(process_array_with_null_check(array, |i| {
        let key_value = keys.value(i).to_usize().unwrap_or(usize::MAX);
        if key_value < translated_values.len() {
            Ok(translated_values[key_value].clone())
        } else {
            Ok(Value::Null)
        }
    }))
}

// Helper function to process dictionary arrays with different key types
fn process_dictionary_array(
    array: &Arc<dyn Array>,
    hamelin_type: &HamelinType,
) -> anyhow::Result<Vec<Value>> {
    // Try different unsigned integer key types
    if let Some(result) = handle_dictionary::<UInt8Type>(array, hamelin_type) {
        return result;
    }
    if let Some(result) = handle_dictionary::<UInt16Type>(array, hamelin_type) {
        return result;
    }
    if let Some(result) = handle_dictionary::<UInt32Type>(array, hamelin_type) {
        return result;
    }

    bail!("Unsupported dictionary key type: {:?}", array.data_type())
}

// Helper function to handle run-end encoded arrays with a specific run-end index type
fn handle_run_end_encoded<R: RunEndIndexType>(
    array: &Arc<dyn Array>,
    hamelin_type: &HamelinType,
) -> Option<anyhow::Result<Vec<Value>>> {
    let run_array = array.as_run_opt::<R>()?;
    let run_ends = run_array.run_ends();
    let values = run_array.values();

    let translated_values = match translate_arrow_array_value(values, hamelin_type) {
        Ok(v) => v,
        Err(e) => return Some(Err(e)),
    };

    // Precompute mapping from logical index to physical index - O(n) instead of O(n²)
    let mut logical_to_physical = vec![0; array.len()];
    let run_ends_values = run_ends.values();
    let mut current_run = 0;
    let mut current_run_end = if run_ends_values.is_empty() {
        0
    } else {
        run_ends_values[0].to_usize().unwrap_or(0)
    };

    for logical_index in 0..array.len() {
        // Move to the next run if we've reached the end of the current run
        while logical_index >= current_run_end && current_run + 1 < run_ends_values.len() {
            current_run += 1;
            current_run_end = run_ends_values[current_run].to_usize().unwrap_or(0);
        }
        logical_to_physical[logical_index] = current_run;
    }

    Some(process_array_with_null_check(array, |i| {
        let physical_index = logical_to_physical[i];

        // Check if the physical index is valid and if the value is not null
        if physical_index < translated_values.len() {
            // Check if the values array has a null at this physical index
            if values.is_null(physical_index) {
                Ok(Value::Null)
            } else {
                Ok(translated_values[physical_index].clone())
            }
        } else {
            Ok(Value::Null)
        }
    }))
}

// Helper function to process run-end encoded arrays with different run-end index types
fn process_run_end_encoded_array(
    array: &Arc<dyn Array>,
    hamelin_type: &HamelinType,
) -> anyhow::Result<Vec<Value>> {
    // Try different run-end index types
    if let Some(result) = handle_run_end_encoded::<Int32Type>(array, hamelin_type) {
        return result;
    }
    if let Some(result) = handle_run_end_encoded::<Int64Type>(array, hamelin_type) {
        return result;
    }
    if let Some(result) = handle_run_end_encoded::<Int16Type>(array, hamelin_type) {
        return result;
    }

    bail!(
        "Unsupported run-end encoded index type: {:?}",
        array.data_type()
    )
}

// Helper function for timestamp arrays
fn process_timestamp_array<T>(
    array: &Arc<dyn Array>,
    time_converter: fn(T::Native) -> anyhow::Result<chrono::DateTime<chrono::Utc>>,
) -> anyhow::Result<Vec<Value>>
where
    T: ArrowPrimitiveType,
{
    let ts_array = array.as_primitive::<T>();
    process_array_with_null_check(array, |i| {
        let ts_value = ts_array.value(i);
        let datetime = time_converter(ts_value)?;
        Ok(Value::String(datetime.to_rfc3339()))
    })
}

// Helper function for decimal arrays
fn process_decimal_array<T>(
    array: &Arc<dyn Array>,
    precision: u8,
    scale: i8,
) -> anyhow::Result<Vec<Value>>
where
    T: DecimalType,
{
    let decimal_array = array.as_primitive::<T>();
    process_array_with_null_check(array, |i| {
        let decimal_value = decimal_array.value(i);
        let decimal_str = T::format_decimal(decimal_value, precision, scale);
        Ok(Value::String(decimal_str))
    })
}

fn translate_arrow_array_value(
    array: &Arc<dyn Array>,
    element_type: &HamelinType,
) -> anyhow::Result<Vec<Value>> {
    let mut results = Vec::with_capacity(array.len());
    // Do the downcast once outside the loop based on element type
    match element_type {
        HamelinType::Unknown => {
            for _ in 0..array.len() {
                results.push(Value::Null);
            }
        }
        HamelinType::Int => match array.data_type() {
            DataType::Int8 => return process_primitive_int_array::<Int8Type>(array),
            DataType::Int16 => return process_primitive_int_array::<Int16Type>(array),
            DataType::Int32 => return process_primitive_int_array::<Int32Type>(array),
            DataType::Int64 => return process_primitive_int_array::<Int64Type>(array),
            DataType::UInt8 => return process_primitive_int_array::<UInt8Type>(array),
            DataType::UInt16 => return process_primitive_int_array::<UInt16Type>(array),
            DataType::UInt32 => return process_primitive_int_array::<UInt32Type>(array),
            DataType::UInt64 => return process_primitive_int_array::<UInt64Type>(array),
            DataType::Dictionary(_, _) => {
                return process_dictionary_array(array, &HamelinType::Int);
            }
            DataType::RunEndEncoded(_, _) => {
                return process_run_end_encoded_array(array, &HamelinType::Int);
            }
            _ => bail!("Unsupported integer array type"),
        },
        HamelinType::Boolean => match array.data_type() {
            DataType::Boolean => {
                let bool_array = array.as_boolean();

                for i in 0..array.len() {
                    if array.is_null(i) {
                        results.push(Value::Null);
                    } else {
                        results.push(Value::Bool(bool_array.value(i)));
                    }
                }
            }
            DataType::Dictionary(_, _) => {
                return process_dictionary_array(array, &HamelinType::Boolean);
            }
            DataType::RunEndEncoded(_, _) => {
                return process_run_end_encoded_array(array, &HamelinType::Boolean);
            }
            _ => {
                bail!("Unexpected data type for boolean array")
            }
        },

        HamelinType::Double => match array.data_type() {
            DataType::Float16 => return process_primitive_float_array::<Float16Type>(array),
            DataType::Float32 => return process_primitive_float_array::<Float32Type>(array),
            DataType::Float64 => return process_primitive_float_array::<Float64Type>(array),
            DataType::Dictionary(_, _) => {
                return process_dictionary_array(array, &HamelinType::Double);
            }
            DataType::RunEndEncoded(_, _) => {
                return process_run_end_encoded_array(array, &HamelinType::Double);
            }

            _ => bail!("Unsupported float array type"),
        },
        HamelinType::Interval => match array.data_type() {
            DataType::Interval(arrow_interval_unit) => match arrow_interval_unit {
                duckdb::arrow::datatypes::IntervalUnit::DayTime => {
                    let interval_array = array.as_primitive::<IntervalDayTimeType>();

                    for i in 0..array.len() {
                        if array.is_null(i) {
                            results.push(Value::Null);
                        } else {
                            let interval_value = interval_array.value(i);
                            // DayTime interval: extract days and milliseconds
                            let days = interval_value.days;
                            let millis = interval_value.milliseconds;

                            let td = chrono::TimeDelta::days(days as i64)
                                + chrono::TimeDelta::milliseconds(millis as i64);
                            let interval_str = format!(
                                "{} {:02}:{:02}:{:02}.{:03}",
                                td.num_days(),
                                td.num_hours(),
                                td.num_minutes(),
                                td.num_seconds(),
                                td.num_milliseconds()
                            );
                            results.push(Value::String(interval_str));
                        }
                    }
                }
                _ => {
                    bail!("Unsupported calendar interval array type")
                }
            },
            DataType::Duration(time_unit) => match time_unit {
                TimeUnit::Second => {
                    let duration_array = array.as_primitive::<DurationSecondType>();
                    return process_array_with_null_check(array, |i| {
                        let duration_value = duration_array.value(i);
                        let td = chrono::TimeDelta::seconds(duration_value);
                        let duration_str = format!(
                            "{} {:02}:{:02}:{:02}.{:03}",
                            td.num_days(),
                            td.num_hours() % 24,
                            td.num_minutes() % 60,
                            td.num_seconds() % 60,
                            td.num_milliseconds() % 1000
                        );
                        Ok(Value::String(duration_str))
                    });
                }
                TimeUnit::Millisecond => {
                    let duration_array = array.as_primitive::<DurationMillisecondType>();
                    return process_array_with_null_check(array, |i| {
                        let duration_value = duration_array.value(i);
                        let td = chrono::TimeDelta::milliseconds(duration_value);
                        let duration_str = format!(
                            "{} {:02}:{:02}:{:02}.{:03}",
                            td.num_days(),
                            td.num_hours() % 24,
                            td.num_minutes() % 60,
                            td.num_seconds() % 60,
                            td.num_milliseconds() % 1000
                        );
                        Ok(Value::String(duration_str))
                    });
                }
                TimeUnit::Microsecond => {
                    let duration_array = array.as_primitive::<DurationMicrosecondType>();
                    return process_array_with_null_check(array, |i| {
                        let duration_value = duration_array.value(i);
                        let td = chrono::TimeDelta::microseconds(duration_value);
                        let duration_str = format!(
                            "{} {:02}:{:02}:{:02}.{:03}",
                            td.num_days(),
                            td.num_hours() % 24,
                            td.num_minutes() % 60,
                            td.num_seconds() % 60,
                            td.num_milliseconds() % 1000
                        );
                        Ok(Value::String(duration_str))
                    });
                }
                TimeUnit::Nanosecond => {
                    let duration_array = array.as_primitive::<DurationNanosecondType>();
                    return process_array_with_null_check(array, |i| {
                        let duration_value = duration_array.value(i);
                        let td = chrono::TimeDelta::nanoseconds(duration_value);
                        let duration_str = format!(
                            "{} {:02}:{:02}:{:02}.{:03}",
                            td.num_days(),
                            td.num_hours() % 24,
                            td.num_minutes() % 60,
                            td.num_seconds() % 60,
                            td.num_milliseconds() % 1000
                        );
                        Ok(Value::String(duration_str))
                    });
                }
            },

            _ => {
                bail!("Unsupported interval array type")
            }
        },
        HamelinType::CalendarInterval => match array.data_type() {
            DataType::Interval(arrow_interval_unit) => match arrow_interval_unit {
                duckdb::arrow::datatypes::IntervalUnit::YearMonth => {
                    let interval_array = array.as_primitive::<IntervalYearMonthType>();

                    for i in 0..array.len() {
                        if array.is_null(i) {
                            results.push(Value::Null);
                        } else {
                            let months = interval_array.value(i);
                            let interval_str = format!("{}-{}", months / 12, months % 12);
                            results.push(Value::String(interval_str));
                        }
                    }
                }
                _ => {
                    bail!("Unsupported calendar interval array type")
                }
            },
            _ => {
                bail!("Unsupported calendar interval array type")
            }
        },
        HamelinType::Binary => match array.data_type() {
            DataType::Binary => {
                let binary_array = array.as_binary::<i32>();
                return process_array_with_null_check(array, |i| {
                    let binary_data = binary_array.value(i);
                    Ok(Value::String(BASE64_STANDARD.encode(binary_data)))
                });
            }
            DataType::LargeBinary => {
                let binary_array = array.as_binary::<i64>();
                return process_array_with_null_check(array, |i| {
                    let binary_data = binary_array.value(i);
                    Ok(Value::String(BASE64_STANDARD.encode(binary_data)))
                });
            }
            DataType::FixedSizeBinary(_) => {
                let binary_array = array.as_fixed_size_binary();
                return process_array_with_null_check(array, |i| {
                    let binary_data = binary_array.value(i);
                    Ok(Value::String(BASE64_STANDARD.encode(binary_data)))
                });
            }
            DataType::BinaryView => {
                let binary_view_array = array.as_binary_view();
                return process_array_with_null_check(array, |i| {
                    let binary_data = binary_view_array.value(i);
                    Ok(Value::String(BASE64_STANDARD.encode(binary_data)))
                });
            }
            DataType::Dictionary(_, _) => {
                return process_dictionary_array(array, &HamelinType::Binary);
            }
            DataType::RunEndEncoded(_, _) => {
                return process_run_end_encoded_array(array, &HamelinType::Binary);
            }
            _ => bail!("Unsupported binary array type"),
        },
        HamelinType::String | HamelinType::Variant => match array.data_type() {
            DataType::Utf8 => {
                let string_array = array.as_string::<i32>();
                return process_array_with_null_check(array, |i| {
                    Ok(Value::String(string_array.value(i).to_string()))
                });
            }
            DataType::LargeUtf8 => {
                let string_array = array.as_string::<i64>();
                return process_array_with_null_check(array, |i| {
                    Ok(Value::String(string_array.value(i).to_string()))
                });
            }
            DataType::Utf8View => {
                let string_view_array = array.as_string_view();
                return process_array_with_null_check(array, |i| {
                    Ok(Value::String(string_view_array.value(i).to_string()))
                });
            }
            DataType::Dictionary(_, _) => {
                return process_dictionary_array(array, &HamelinType::String);
            }
            DataType::RunEndEncoded(_, _) => {
                return process_run_end_encoded_array(array, &HamelinType::String);
            }
            _ => bail!("Unsupported string array type"),
        },
        HamelinType::Timestamp => match array.data_type() {
            DataType::Date32 => {
                let date_array = array.as_primitive::<Date32Type>();
                return process_array_with_null_check(array, |i| {
                    let date_value = date_array.value(i);
                    let datetime =
                        chrono::DateTime::UNIX_EPOCH + chrono::Duration::days(date_value as i64);
                    Ok(Value::String(datetime.to_rfc3339()))
                });
            }
            DataType::Date64 => {
                let date_array = array.as_primitive::<Date64Type>();
                return process_array_with_null_check(array, |i| {
                    let date_value = date_array.value(i);
                    let datetime = chrono::DateTime::from_timestamp_millis(date_value)
                        .ok_or_else(|| anyhow!("Invalid date64 value"))?;
                    Ok(Value::String(datetime.to_rfc3339()))
                });
            }
            DataType::Timestamp(time_unit, _) => match time_unit {
                TimeUnit::Nanosecond => {
                    return process_timestamp_array::<TimestampNanosecondType>(array, |ts_value| {
                        Ok(chrono::DateTime::from_timestamp_nanos(ts_value))
                    });
                }
                TimeUnit::Microsecond => {
                    return process_timestamp_array::<TimestampMicrosecondType>(
                        array,
                        |ts_value| {
                            chrono::DateTime::from_timestamp_micros(ts_value)
                                .ok_or_else(|| anyhow!("Invalid timestamp value"))
                        },
                    );
                }
                TimeUnit::Millisecond => {
                    return process_timestamp_array::<TimestampMillisecondType>(
                        array,
                        |ts_value| {
                            chrono::DateTime::from_timestamp_millis(ts_value)
                                .ok_or_else(|| anyhow!("Invalid timestamp value"))
                        },
                    );
                }
                TimeUnit::Second => {
                    return process_timestamp_array::<TimestampSecondType>(array, |ts_value| {
                        chrono::DateTime::from_timestamp(ts_value, 0)
                            .ok_or_else(|| anyhow!("Invalid timestamp value"))
                    });
                }
            },
            _ => bail!("Unsupported timestamp array type"),
        },
        HamelinType::Array {
            element_type: inner_element_type,
        } => match array.data_type() {
            DataType::List(_) => {
                let list_array = array.as_list::<i32>();
                return process_array_with_null_check(array, |i| {
                    let nested_array = list_array.value(i);
                    let nested_values =
                        translate_arrow_array_value(&nested_array, inner_element_type)?;
                    Ok(Value::Array(nested_values))
                });
            }
            DataType::LargeList(_) => {
                let list_array = array.as_list::<i64>();
                return process_array_with_null_check(array, |i| {
                    let nested_array = list_array.value(i);
                    let nested_values =
                        translate_arrow_array_value(&nested_array, inner_element_type)?;
                    Ok(Value::Array(nested_values))
                });
            }
            DataType::FixedSizeList(_, _) => {
                let list_array = array.as_fixed_size_list();
                return process_array_with_null_check(array, |i| {
                    let nested_array = list_array.value(i);
                    let nested_values =
                        translate_arrow_array_value(&nested_array, inner_element_type)?;
                    Ok(Value::Array(nested_values))
                });
            }
            DataType::ListView(_) => {
                let list_view_array = array.as_list_view::<i32>();
                return process_array_with_null_check(array, |i| {
                    let nested_array = list_view_array.value(i);
                    let nested_values =
                        translate_arrow_array_value(&nested_array, inner_element_type)?;
                    Ok(Value::Array(nested_values))
                });
            }
            DataType::LargeListView(_) => {
                let list_view_array = array.as_list_view::<i64>();
                return process_array_with_null_check(array, |i| {
                    let nested_array = list_view_array.value(i);
                    let nested_values =
                        translate_arrow_array_value(&nested_array, inner_element_type)?;
                    Ok(Value::Array(nested_values))
                });
            }
            _ => bail!("Unsupported list array type"),
        },
        // Map types in Arrow are represented as List<Struct<key, value>>
        // We convert them to JSON objects by extracting keys and values separately
        // and then creating a map where keys are converted to strings for JSON compatibility
        HamelinType::Map {
            key_type,
            value_type,
        } => match array.data_type() {
            DataType::Map(_, _) => {
                let map_array = array.as_map();

                for i in 0..array.len() {
                    if array.is_null(i) {
                        results.push(Value::Null);
                    } else {
                        let entries = map_array.value(i);
                        // entries is a struct array with key and value fields

                        // Get the key and value columns
                        let key_column = entries.column(0); // keys
                        let value_column = entries.column(1); // values

                        // Translate keys and values
                        let keys = translate_arrow_array_value(key_column, key_type)?;
                        let values = translate_arrow_array_value(value_column, value_type)?;

                        // Create JSON object
                        let mut map_obj = serde_json::Map::new();
                        for (key, value) in keys.iter().zip(values.iter()) {
                            // Convert key to string for JSON object
                            let key_str = match key {
                                Value::String(s) => s.clone(),
                                _ => key.to_string(),
                            };
                            map_obj.insert(key_str, value.clone());
                        }

                        results.push(Value::Object(map_obj));
                    }
                }
            }
            _ => {
                bail!("Unsupported map array type")
            }
        },
        // Struct and Tuple types in Arrow are both represented as Structs.
        // Arrow stores these in columnar format - each field is a separate column array.
        // We process each field column entirely, then transpose to get row-wise representation.
        // Final output: JSON arrays containing field values in field order (compact representation).
        // This avoids repeating field names for each row since the schema already defines them.
        HamelinType::Struct(_) | HamelinType::Tuple { .. } | HamelinType::Range { .. } => {
            let types: Vec<HamelinType> = match element_type.clone() {
                HamelinType::Struct(fields) => fields.into_iter().map(|f| f.typ).collect(),
                HamelinType::Tuple { elements } => elements,
                HamelinType::Range { of } => vec![*of.clone(), *of.clone()],
                _ => unreachable!(),
            };

            match array.data_type() {
                DataType::Struct(_) => {
                    let struct_array = array.as_struct();
                    let mut column_results = Vec::new();

                    for (i, typ) in types.iter().enumerate() {
                        let column = struct_array.column(i);
                        let values = translate_arrow_array_value(column, typ)?;
                        column_results.push(values);
                    }

                    let row_results = transpose(column_results);

                    for (i, row) in row_results.into_iter().enumerate() {
                        if array.is_null(i) {
                            results.push(Value::Null);
                        } else {
                            results.push(Value::Array(row));
                        }
                    }
                }
                _ => {
                    bail!("Unsupported struct/tuple array type")
                }
            }
        }
        HamelinType::Decimal {
            precision: _,
            scale: _,
        } => match array.data_type() {
            DataType::Decimal128(precision, scale) => {
                return process_decimal_array::<Decimal128Type>(array, *precision, *scale);
            }
            DataType::Decimal256(precision, scale) => {
                return process_decimal_array::<Decimal256Type>(array, *precision, *scale);
            }
            _ => bail!("Unsupported decimal array type"),
        },
        _ => {
            bail!("Unsupported array element type: {:?}", element_type)
        }
    }

    Ok(results)
}