//! DuckDB-file DuckLake catalog backend.

use std::collections::HashMap;
use std::fs;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex, OnceLock};

#[cfg(test)]
use crate::{
    AddColumnCommit, CleanupOldFilesOptions, ColumnDefaultCommit, ColumnTypeCommit,
    ColumnTypeTreeCommit, CreateSchemaCommit, CreateTableCommit, CreateViewCommit,
    DeleteFileCommit, DropColumnCommit, DropSchemaCommit, DropTableCommit, DropViewCommit,
    ExpireSnapshotsOptions, FlushedInlinedDataCommit, FlushedInlinedFileDeletionCommit,
    InlinedDataAppendCommit, InlinedDataDeletionCommit, InlinedFileDeletionCommit,
    MergeAdjacentCompactionCommit, RenameColumnCommit, RenameTableCommit, RewriteDataFilesCommit,
    SortedByCommit, TableChangeCommit, TruncateTableCommit,
};
use arrow::array::Array;
#[cfg(test)]
use arrow::array::{
    BinaryArray, Date32Array, Decimal128Array, Decimal256Array, FixedSizeBinaryArray, Float32Array,
    Float64Array, Int8Array, Int16Array, Int32Array, IntervalMonthDayNanoArray, StringArray,
    Time64MicrosecondArray, Time64NanosecondArray, TimestampMicrosecondArray,
    TimestampMillisecondArray, TimestampNanosecondArray, TimestampSecondArray, UInt8Array,
    UInt16Array, UInt32Array, UInt64Array,
};
use arrow::datatypes::IntervalMonthDayNano;
#[cfg(test)]
use arrow::datatypes::{IntervalUnit, i256};
#[cfg(test)]
use arrow::record_batch::RecordBatch;
use arrow_schema::TimeUnit;
#[cfg(test)]
use arrow_schema::{DataType, Field, Schema};
use async_trait::async_trait;
use duckdb::types::{TimeUnit as DuckDbTimeUnit, Value as DuckDbValue, ValueRef};
use duckdb::{Connection, OptionalExt, Row, params, params_from_iter};
#[cfg(test)]
use ducklake_catalog::InlinedDataTableRef;
#[cfg(test)]
use ducklake_catalog::PartitionTransform;
use ducklake_catalog::SchemaField;
#[cfg(test)]
use ducklake_catalog::ViewId;
#[cfg(test)]
use ducklake_catalog::{
    DataFileId, FileColumnStats, MacroId, MacroImplId, PartitionId, SchemaId, SnapshotId, SortId,
    TableId, TableSchema,
};

use crate::embedded_reads::quote_identifier;
use crate::inlined_values::{
    InlinedSqlDialect, InlinedSqlScalarRef, decode_inlined_scalar_string,
    duckdb_inlined_value_needs_cast, format_inlined_date32, inlined_column_type,
    parse_inlined_date32, parse_inlined_time64, parse_inlined_timestamp, parse_inlined_uuid,
};
use crate::{
    CatalogBackend, DuckLakeCatalogConnectionOptions, Result, StorageError, duckdb_literal_to_json,
    parse_interval_month_day_nano,
};

#[cfg(test)]
use arrow::array::{ListArray, StructArray};

static DUCKDB_CATALOG_WRITE_LOCKS: OnceLock<Mutex<HashMap<PathBuf, Arc<Mutex<()>>>>> =
    OnceLock::new();

#[derive(Debug, Clone)]
pub struct DuckDbCatalog {
    path: PathBuf,
    metadata_schema: Option<String>,
    connection_options: DuckLakeCatalogConnectionOptions,
}

impl DuckDbCatalog {
    pub fn new(path: impl Into<PathBuf>) -> Self {
        Self {
            path: path.into(),
            metadata_schema: None,
            connection_options: DuckLakeCatalogConnectionOptions::default(),
        }
    }

    pub fn with_metadata_schema(
        path: impl Into<PathBuf>,
        metadata_schema: impl Into<String>,
    ) -> Self {
        Self {
            path: path.into(),
            metadata_schema: Some(metadata_schema.into()),
            connection_options: DuckLakeCatalogConnectionOptions::default(),
        }
    }

    impl_connection_options_builder_method!();

    impl_path_accessor_method!();

    impl_metadata_schema_accessor_method!(Option<&str>, |this| this.metadata_schema.as_deref());

    async fn with_conn<T, F>(&self, f: F) -> Result<T>
    where
        T: Send + 'static,
        F: FnOnce(Connection) -> Result<T> + Send + 'static,
    {
        let path = self.path.clone();
        let metadata_schema = self.metadata_schema.clone();
        tokio::task::spawn_blocking(move || {
            let conn = open_duckdb_catalog_connection(&path, metadata_schema.as_deref())?;
            f(conn)
        })
        .await
        .map_err(|err| StorageError::Backend(err.to_string()))?
    }

    async fn with_write_conn<T, F>(&self, f: F) -> Result<T>
    where
        T: Send + 'static,
        F: FnOnce(Connection) -> Result<T> + Send + 'static,
    {
        let path = self.path.clone();
        let metadata_schema = self.metadata_schema.clone();
        tokio::task::spawn_blocking(move || {
            let lock = duckdb_catalog_write_lock(&path)?;
            let _guard = lock
                .lock()
                .map_err(|_| StorageError::Backend("DuckDB catalog write lock poisoned".into()))?;
            let conn = open_duckdb_catalog_connection(&path, metadata_schema.as_deref())?;
            f(conn)
        })
        .await
        .map_err(|err| StorageError::Backend(err.to_string()))?
    }

    async fn with_initialization_conn<T, F>(&self, f: F) -> Result<T>
    where
        T: Send + 'static,
        F: FnOnce(Connection) -> Result<T> + Send + 'static,
    {
        let path = self.path.clone();
        tokio::task::spawn_blocking(move || {
            let lock = duckdb_catalog_write_lock(&path)?;
            let _guard = lock
                .lock()
                .map_err(|_| StorageError::Backend("DuckDB catalog write lock poisoned".into()))?;
            let conn = Connection::open(&path).map_err(sql_err)?;
            f(conn)
        })
        .await
        .map_err(|err| StorageError::Backend(err.to_string()))?
    }

    impl_embedded_run_commit_request_method!(DuckDbCommitOps, with_write_conn, |conn| conn
        .transaction()
        .map_err(sql_err));
}

fn duckdb_catalog_write_lock(path: &Path) -> Result<Arc<Mutex<()>>> {
    let key = fs::canonicalize(path).unwrap_or_else(|_| path.to_path_buf());
    let locks = DUCKDB_CATALOG_WRITE_LOCKS.get_or_init(|| Mutex::new(HashMap::new()));
    let mut locks = locks
        .lock()
        .map_err(|_| StorageError::Backend("DuckDB catalog write lock registry poisoned".into()))?;
    Ok(locks
        .entry(key)
        .or_insert_with(|| Arc::new(Mutex::new(())))
        .clone())
}

fn open_duckdb_catalog_connection(
    path: &Path,
    metadata_schema: Option<&str>,
) -> Result<Connection> {
    let conn = Connection::open(path).map_err(sql_err)?;
    use_ducklake_metadata_schema(&conn, metadata_schema)?;
    Ok(conn)
}

fn use_ducklake_metadata_schema(conn: &Connection, metadata_schema: Option<&str>) -> Result<()> {
    if let Some(schema) = metadata_schema {
        let exists = conn
            .query_row(
                "SELECT 1 \
                 FROM information_schema.tables \
                 WHERE table_schema = ?1 AND table_name = 'ducklake_metadata' \
                 LIMIT 1",
                params![schema],
                |row| row.get::<_, i64>(0),
            )
            .optional()
            .map_err(sql_err)?
            .is_some();
        if !exists {
            return Err(StorageError::Backend(format!(
                "DuckDB catalog metadata schema {schema:?} does not contain ducklake_metadata"
            )));
        }
        conn.execute_batch(&format!("USE {}", quote_identifier(schema)))
            .map_err(sql_err)?;
        return Ok(());
    }

    let mut stmt = conn
        .prepare(
            "SELECT table_schema \
             FROM information_schema.tables \
             WHERE table_name = 'ducklake_metadata' \
             ORDER BY CASE WHEN table_schema = 'main' THEN 0 ELSE 1 END, table_schema",
        )
        .map_err(sql_err)?;
    let schemas = collect_rows(&mut stmt, [], |row| row.get::<_, String>(0))?;
    match schemas.as_slice() {
        [] => Ok(()),
        [schema] => {
            if schema != "main" {
                conn.execute_batch(&format!("USE {}", quote_identifier(schema)))
                    .map_err(sql_err)?;
            }
            Ok(())
        }
        _ => Err(StorageError::Backend(format!(
            "DuckDB catalog contains multiple ducklake_metadata tables: {}",
            schemas.join(", ")
        ))),
    }
}

fn duckdb_metadata_schema_exists_and_use(
    conn: &Connection,
    metadata_schema: Option<&str>,
) -> Result<bool> {
    if let Some(schema) = metadata_schema {
        let exists = conn
            .query_row(
                "SELECT 1 \
                 FROM information_schema.tables \
                 WHERE table_schema = ?1 AND table_name = 'ducklake_metadata' \
                 LIMIT 1",
                params![schema],
                |row| row.get::<_, i64>(0),
            )
            .optional()
            .map_err(sql_err)?
            .is_some();
        if exists {
            conn.execute_batch(&format!("USE {}", quote_identifier(schema)))
                .map_err(sql_err)?;
        }
        return Ok(exists);
    }

    let mut stmt = conn
        .prepare(
            "SELECT table_schema \
             FROM information_schema.tables \
             WHERE table_name = 'ducklake_metadata' \
             ORDER BY CASE WHEN table_schema = 'main' THEN 0 ELSE 1 END, table_schema",
        )
        .map_err(sql_err)?;
    let schemas = collect_rows(&mut stmt, [], |row| row.get::<_, String>(0))?;
    match schemas.as_slice() {
        [] => Ok(false),
        [schema] => {
            if schema != "main" {
                conn.execute_batch(&format!("USE {}", quote_identifier(schema)))
                    .map_err(sql_err)?;
            }
            Ok(true)
        }
        _ => Err(StorageError::Backend(format!(
            "DuckDB catalog contains multiple ducklake_metadata tables: {}",
            schemas.join(", ")
        ))),
    }
}

fn duckdb_initialize_ducklake(
    conn: &Connection,
    metadata_schema: Option<&str>,
    bootstrap: &crate::catalog_ddl::CatalogBootstrap,
) -> Result<()> {
    if let Some(schema) = metadata_schema {
        conn.execute_batch(&format!(
            "CREATE SCHEMA IF NOT EXISTS {}",
            quote_identifier(schema)
        ))
        .map_err(sql_err)?;
    }
    let ddl = crate::catalog_ddl::render_initialization_batch(metadata_schema, bootstrap);
    conn.execute_batch(&ddl).map_err(sql_err)?;
    if let Some(schema) = metadata_schema {
        conn.execute_batch(&format!("USE {}", quote_identifier(schema)))
            .map_err(sql_err)?;
    }
    Ok(())
}

struct DuckDbCommitOps;

impl_embedded_sync_commit_ops!(
    DuckDbCommitOps,
    tx_type = duckdb::Transaction<'tx>,
    tx = duckdb::Transaction<'_>,
    table_exists = duckdb_table_exists
);

impl_embedded_sync_table_data_commit_ops!(
    DuckDbCommitOps,
    duckdb::Transaction<'_>,
    dialect = crate::inlined_values::InlinedSqlDialect::DuckDbFile,
    next_row_id_for_append =
        crate::stats_engine::next_row_id_for_append_from_table_stats_sync_query,
    insert_inlined_batches = duckdb_insert_inlined_batches,
    overwritten_cleanup_data_path =
        |tx| { global_metadata_value(tx, "data_path").map(|value| value.unwrap_or_default()) },
    write_inlined_file_deletions = self::write_inlined_file_deletions,
    apply_inlined_data_deletions = duckdb_apply_inlined_data_deletions,
    delete_flushed_inlined_data = duckdb_delete_flushed_inlined_data,
    delete_flushed_inlined_file_deletions = duckdb_delete_flushed_inlined_file_deletions
);

impl_embedded_sync_maintenance_commit_ops!(DuckDbCommitOps, duckdb::Transaction<'_>);

impl_embedded_sync_catalog_object_commit_ops!(
    DuckDbCommitOps,
    duckdb::Transaction<'_>,
    uuid_style = crate::catalog_queries::UuidParamStyle::CastAsUuid,
    data_path = duckdb_transaction_data_path,
    close_inlined = close_duckdb_inlined_data_rows_for_table
);

impl crate::commit::SyncCreateTableWithFilesCommitOps for DuckDbCommitOps {}

impl_embedded_sync_metadata_commit_ops!(
    DuckDbCommitOps,
    duckdb::Transaction<'_>,
    rewrite_sort_expressions_for_rename = rewrite_duckdb_sort_expressions_for_rename
);

#[async_trait]
impl crate::catalog_initialization_methods::CatalogInitializationExecutor for DuckDbCatalog {
    impl_catalog_initialization_info_methods!(
        catalog_type = "duckdb",
        default_data_path = |this| Some(format!("{}.files", this.path.display())),
        snapshot_time_sql = |_this| Ok("NOW()".into())
    );

    impl_sync_catalog_initialization_request!(
        with_session = with_initialization_conn,
        capture = |this| {
            let metadata_schema = this.metadata_schema.clone()
        },
        exists = |conn| duckdb_metadata_schema_exists_and_use(conn, metadata_schema.as_deref()),
        load = |conn| {
            crate::catalog_metadata::load_catalog_metadata_sync(|query, mapper| {
                run_catalog_query(conn, query, mapper)
            })
        },
        create = |conn, bootstrap| {
            duckdb_initialize_ducklake(conn, metadata_schema.as_deref(), &bootstrap)
        },
    );
}

#[async_trait]
impl CatalogBackend for DuckDbCatalog {
    impl_catalog_type_method!("duckdb");

    impl_embedded_metadata_method!(with_conn);

    impl_embedded_set_metadata_option_method!(with_write_conn, |conn| conn
        .transaction()
        .map_err(sql_err));

    impl_catalog_initialization_method!();

    impl_catalog_backend_read_methods_direct!();

    impl_embedded_inlined_read_methods!(
        file_deletion_table_exists_sql = crate::catalog_table_exists::DUCKDB_TABLE_PRESENT_SQL
    );

    impl_catalog_backend_commit_methods!();

    impl_embedded_catalog_maintenance_methods!(with_write_conn, |conn| conn
        .transaction()
        .map_err(sql_err));
}

impl_embedded_inlined_mutation_helpers!(
    tx = duckdb::Transaction<'_>,
    table_exists = duckdb_table_exists,
    apply_inlined_data_deletions = duckdb_apply_inlined_data_deletions,
    delete_flushed_inlined_data = duckdb_delete_flushed_inlined_data,
    delete_flushed_inlined_file_deletions = duckdb_delete_flushed_inlined_file_deletions,
    write_inlined_file_deletions = write_inlined_file_deletions,
    close_inlined_data_rows_for_table = close_duckdb_inlined_data_rows_for_table,
);

impl_embedded_insert_inlined_batches_helper!(
    fn duckdb_insert_inlined_batches,
    tx = duckdb::Transaction<'_>,
    params_from_iter = params_from_iter,
    metadata_i64_value = |value| DuckDbValue::BigInt(value),
    metadata_null_value = DuckDbValue::Null,
    placeholder = |_idx, field| duckdb_inlined_insert_placeholder(field),
    value = duckdb_inlined_value,
);

fn duckdb_inlined_insert_placeholder(field: &SchemaField) -> Result<String> {
    if duckdb_inlined_value_needs_cast(field) {
        Ok(format!(
            "CAST(? AS {})",
            inlined_column_type(field, InlinedSqlDialect::DuckDbFile)?
        ))
    } else {
        Ok("?".into())
    }
}

fn duckdb_inlined_value(field: &SchemaField, array: &dyn Array, row: usize) -> Result<DuckDbValue> {
    if matches!(
        field.ducklake_type,
        ducklake_catalog::DuckLakeType::Struct
            | ducklake_catalog::DuckLakeType::List
            | ducklake_catalog::DuckLakeType::Map
    ) {
        if array.is_null(row) {
            return Ok(DuckDbValue::Null);
        }
        return crate::inlined_values::inlined_duckdb_literal_value(
            field,
            array,
            row,
            "DuckDB inlined append",
            crate::inlined_values::InlinedIntervalFormat::MicrosecondsOnly("DuckDB"),
        )
        .map(DuckDbValue::Text);
    }
    crate::inlined_values::encode_inlined_scalar_value(
        field,
        array,
        row,
        "DuckDB inlined append",
        crate::inlined_values::InlinedIntervalFormat::MicrosecondsOnly("DuckDB"),
        &DuckDbInlinedValueEncoder,
    )
}

struct DuckDbInlinedValueEncoder;

impl_inlined_scalar_value_encoder!(
    impl DuckDbInlinedValueEncoder,
    Value = DuckDbValue,
    {
        null(_field) => Ok(DuckDbValue::Null),
        boolean(value) => Ok(DuckDbValue::Boolean(value)),
        int8(value) => Ok(DuckDbValue::TinyInt(value)),
        int16(value) => Ok(DuckDbValue::SmallInt(value)),
        int32(value) => Ok(DuckDbValue::Int(value)),
        int64(value) => Ok(DuckDbValue::BigInt(value)),
        uint8(value) => Ok(DuckDbValue::UTinyInt(value)),
        uint16(value) => Ok(DuckDbValue::USmallInt(value)),
        uint32(value) => Ok(DuckDbValue::UInt(value)),
        uint64(_field, value) => Ok(DuckDbValue::Text(value.to_string())),
        float32(value) => Ok(DuckDbValue::Float(value)),
        float64(value) => Ok(DuckDbValue::Double(value)),
        text(_field, value) => Ok(DuckDbValue::Text(value)),
        bytes(_field, value) => Ok(DuckDbValue::Blob(value)),
    }
);

fn duckdb_transaction_data_path(conn: &Connection) -> Result<String> {
    global_metadata_value(conn, "data_path").map(|value| value.unwrap_or_default())
}

fn duckdb_table_exists(conn: &Connection, table_name: &str) -> Result<bool> {
    conn.query_row(
        crate::catalog_table_exists::DUCKDB_TABLE_EXISTS_SQL,
        params![table_name],
        |row| row.get(0),
    )
    .map_err(sql_err)
}

impl_embedded_sort_rename_rewrite_helper!(
    fn rewrite_duckdb_sort_expressions_for_rename,
    tx = Connection,
    table_exists = duckdb_table_exists,
);

impl_embedded_query_helpers!(
    statement = duckdb::Statement<'_>,
    params_trait = duckdb::Params,
    row = Row<'_>,
    error = duckdb::Error,
    params_from_iter = params_from_iter,
    catalog_params = duckdb_catalog_params
);

impl_embedded_async_read_query_executor!(impl DuckDbCatalog);

impl_embedded_catalog_params!(
    fn duckdb_catalog_params -> DuckDbValue {
        null = DuckDbValue::Null,
        i64 = |value| DuckDbValue::BigInt(value),
        string = |value| DuckDbValue::Text(value),
        bool = |value| DuckDbValue::Boolean(value),
    }
);

impl_embedded_maintenance_conn!(
    table_exists = duckdb_table_exists,
    catalog_params = duckdb_catalog_params,
    scheduled_cleanup_timestamp = current_timestamp,
);

struct DuckDbInlinedValueDecoder;

impl_embedded_inlined_data_reader!(
    dialect = crate::inlined_values::InlinedSqlDialect::DuckDbFile,
    projection = |field| {
        crate::inlined_values::inlined_read_projection_expr(
            field,
            crate::inlined_values::InlinedSqlDialect::DuckDbFile,
            quote_identifier,
        )
    },
    decoder = DuckDbInlinedValueDecoder,
);

impl crate::inlined_values::InlinedSqlValueDecoder<ValueRef<'_>> for DuckDbInlinedValueDecoder {
    impl_inlined_scalar_decoder_methods!(
        value = ValueRef<'_>,
        backend = "DuckDB",
        scalar = duckdb_scalar,
    );

    fn decode_string(&self, value: ValueRef<'_>) -> Result<Option<String>> {
        match value {
            ValueRef::Date32(value) => format_inlined_date32(value, "DuckDB").map(Some),
            ValueRef::Time64(source_unit, value) => {
                format_duckdb_time64(source_unit, value, arrow_unit_from_duckdb(source_unit))
                    .map(Some)
            }
            ValueRef::Timestamp(source_unit, value) => format_duckdb_timestamp(
                source_unit,
                value,
                arrow_unit_from_duckdb(source_unit),
                false,
            )
            .map(Some),
            value => decode_inlined_scalar_string(duckdb_scalar(value), "DuckDB"),
        }
    }

    fn decode_date32(&self, value: ValueRef<'_>) -> Result<Option<i32>> {
        sql_date32(value)
    }

    fn decode_time64(&self, value: ValueRef<'_>, unit: TimeUnit) -> Result<Option<i64>> {
        sql_time64(value, unit)
    }

    fn decode_timestamp(&self, value: ValueRef<'_>, unit: TimeUnit) -> Result<Option<i64>> {
        sql_timestamp(value, unit)
    }

    fn decode_interval(&self, value: ValueRef<'_>) -> Result<Option<IntervalMonthDayNano>> {
        sql_interval(value)
    }

    fn decode_nested(&self, field: &SchemaField, value: ValueRef<'_>) -> Result<Option<String>> {
        sql_nested_json(field, value)
    }
}

fn duckdb_scalar(value: ValueRef<'_>) -> InlinedSqlScalarRef<'_> {
    match value {
        ValueRef::Null => InlinedSqlScalarRef::Null,
        ValueRef::Boolean(value) => InlinedSqlScalarRef::Boolean(value),
        ValueRef::TinyInt(value) => InlinedSqlScalarRef::Signed(i128::from(value)),
        ValueRef::SmallInt(value) => InlinedSqlScalarRef::Signed(i128::from(value)),
        ValueRef::Int(value) => InlinedSqlScalarRef::Signed(i128::from(value)),
        ValueRef::BigInt(value) => InlinedSqlScalarRef::Signed(i128::from(value)),
        ValueRef::HugeInt(value) => InlinedSqlScalarRef::Signed(value),
        ValueRef::UTinyInt(value) => InlinedSqlScalarRef::Unsigned(u64::from(value)),
        ValueRef::USmallInt(value) => InlinedSqlScalarRef::Unsigned(u64::from(value)),
        ValueRef::UInt(value) => InlinedSqlScalarRef::Unsigned(u64::from(value)),
        ValueRef::UBigInt(value) => InlinedSqlScalarRef::Unsigned(value),
        ValueRef::Float(value) => InlinedSqlScalarRef::Float(f64::from(value)),
        ValueRef::Double(value) => InlinedSqlScalarRef::Float(value),
        ValueRef::Decimal(value) => InlinedSqlScalarRef::DecimalText(value.to_string()),
        ValueRef::Text(value) => InlinedSqlScalarRef::Text(value),
        ValueRef::Blob(value) => InlinedSqlScalarRef::Bytes(value),
        other => InlinedSqlScalarRef::Unsupported(format!("{other:?}")),
    }
}

fn sql_nested_json(field: &SchemaField, value: ValueRef<'_>) -> Result<Option<String>> {
    match value {
        ValueRef::Null => Ok(None),
        ValueRef::Text(_) => decode_inlined_scalar_string(duckdb_scalar(value), "DuckDB")?
            .map(|value| duckdb_literal_to_json(field, &value))
            .transpose(),
        value => duckdb_value_to_json(field, &value.to_owned()).map(Some),
    }
}

fn duckdb_value_to_json(field: &SchemaField, value: &DuckDbValue) -> Result<String> {
    match value {
        DuckDbValue::Null => Ok("null".into()),
        DuckDbValue::Struct(values) => {
            let mut parts = Vec::new();
            for (name, value) in values.iter() {
                let child = field
                    .children
                    .iter()
                    .find(|child| child.name == *name)
                    .ok_or_else(|| {
                        StorageError::Backend(format!(
                            "DuckDB inlined struct {} has unknown child {name:?}",
                            field.name
                        ))
                    })?;
                parts.push(format!(
                    "{}:{}",
                    crate::json_string(name),
                    duckdb_value_to_json(child, value)?
                ));
            }
            Ok(format!("{{{}}}", parts.join(", ")))
        }
        DuckDbValue::List(values) | DuckDbValue::Array(values) => {
            let child = field.children.first().ok_or_else(|| {
                StorageError::Backend(format!("DuckLake list field {} has no child", field.name))
            })?;
            let parts = values
                .iter()
                .map(|value| duckdb_value_to_json(child, value))
                .collect::<Result<Vec<_>>>()?;
            Ok(format!("[{}]", parts.join(", ")))
        }
        DuckDbValue::Map(values) => {
            if field.children.len() != 2 {
                return Err(StorageError::Backend(format!(
                    "DuckLake map field {} must have key and value children",
                    field.name
                )));
            }
            let key_field = &field.children[0];
            let value_field = &field.children[1];
            let parts = values
                .iter()
                .map(|(key, value)| {
                    Ok(format!(
                        "{}:{}",
                        duckdb_map_key_to_json(key_field, key)?,
                        duckdb_value_to_json(value_field, value)?
                    ))
                })
                .collect::<Result<Vec<_>>>()?;
            Ok(format!("{{{}}}", parts.join(", ")))
        }
        value => duckdb_scalar_value_to_json(field, value),
    }
}

fn duckdb_scalar_value_to_json(field: &SchemaField, value: &DuckDbValue) -> Result<String> {
    match (&field.ducklake_type, value) {
        (_, DuckDbValue::Null) => Ok("null".into()),
        (_, DuckDbValue::Struct(_))
        | (_, DuckDbValue::List(_))
        | (_, DuckDbValue::Array(_))
        | (_, DuckDbValue::Map(_)) => duckdb_value_to_json(field, value),
        (ducklake_catalog::DuckLakeType::Boolean, DuckDbValue::Boolean(value)) => {
            Ok(value.to_string())
        }
        (
            ducklake_catalog::DuckLakeType::Int8
            | ducklake_catalog::DuckLakeType::Int16
            | ducklake_catalog::DuckLakeType::Int32
            | ducklake_catalog::DuckLakeType::Int64
            | ducklake_catalog::DuckLakeType::UInt8
            | ducklake_catalog::DuckLakeType::UInt16
            | ducklake_catalog::DuckLakeType::UInt32
            | ducklake_catalog::DuckLakeType::UInt64
            | ducklake_catalog::DuckLakeType::Float32
            | ducklake_catalog::DuckLakeType::Float64,
            DuckDbValue::Boolean(value),
        ) => Ok(i32::from(*value).to_string()),
        (
            ducklake_catalog::DuckLakeType::Int8
            | ducklake_catalog::DuckLakeType::Int16
            | ducklake_catalog::DuckLakeType::Int32
            | ducklake_catalog::DuckLakeType::Int64
            | ducklake_catalog::DuckLakeType::UInt8
            | ducklake_catalog::DuckLakeType::UInt16
            | ducklake_catalog::DuckLakeType::UInt32
            | ducklake_catalog::DuckLakeType::UInt64
            | ducklake_catalog::DuckLakeType::Float32
            | ducklake_catalog::DuckLakeType::Float64,
            DuckDbValue::TinyInt(value),
        ) => Ok(value.to_string()),
        (
            ducklake_catalog::DuckLakeType::Int8
            | ducklake_catalog::DuckLakeType::Int16
            | ducklake_catalog::DuckLakeType::Int32
            | ducklake_catalog::DuckLakeType::Int64
            | ducklake_catalog::DuckLakeType::UInt8
            | ducklake_catalog::DuckLakeType::UInt16
            | ducklake_catalog::DuckLakeType::UInt32
            | ducklake_catalog::DuckLakeType::UInt64
            | ducklake_catalog::DuckLakeType::Float32
            | ducklake_catalog::DuckLakeType::Float64,
            DuckDbValue::SmallInt(value),
        ) => Ok(value.to_string()),
        (
            ducklake_catalog::DuckLakeType::Int8
            | ducklake_catalog::DuckLakeType::Int16
            | ducklake_catalog::DuckLakeType::Int32
            | ducklake_catalog::DuckLakeType::Int64
            | ducklake_catalog::DuckLakeType::UInt8
            | ducklake_catalog::DuckLakeType::UInt16
            | ducklake_catalog::DuckLakeType::UInt32
            | ducklake_catalog::DuckLakeType::UInt64
            | ducklake_catalog::DuckLakeType::Float32
            | ducklake_catalog::DuckLakeType::Float64,
            DuckDbValue::Int(value),
        ) => Ok(value.to_string()),
        (
            ducklake_catalog::DuckLakeType::Int8
            | ducklake_catalog::DuckLakeType::Int16
            | ducklake_catalog::DuckLakeType::Int32
            | ducklake_catalog::DuckLakeType::Int64
            | ducklake_catalog::DuckLakeType::UInt8
            | ducklake_catalog::DuckLakeType::UInt16
            | ducklake_catalog::DuckLakeType::UInt32
            | ducklake_catalog::DuckLakeType::UInt64
            | ducklake_catalog::DuckLakeType::Float32
            | ducklake_catalog::DuckLakeType::Float64,
            DuckDbValue::BigInt(value),
        ) => Ok(value.to_string()),
        (
            ducklake_catalog::DuckLakeType::Int8
            | ducklake_catalog::DuckLakeType::Int16
            | ducklake_catalog::DuckLakeType::Int32
            | ducklake_catalog::DuckLakeType::Int64
            | ducklake_catalog::DuckLakeType::UInt8
            | ducklake_catalog::DuckLakeType::UInt16
            | ducklake_catalog::DuckLakeType::UInt32
            | ducklake_catalog::DuckLakeType::UInt64
            | ducklake_catalog::DuckLakeType::Float32
            | ducklake_catalog::DuckLakeType::Float64,
            DuckDbValue::HugeInt(value),
        ) => Ok(value.to_string()),
        (
            ducklake_catalog::DuckLakeType::Int8
            | ducklake_catalog::DuckLakeType::Int16
            | ducklake_catalog::DuckLakeType::Int32
            | ducklake_catalog::DuckLakeType::Int64
            | ducklake_catalog::DuckLakeType::UInt8
            | ducklake_catalog::DuckLakeType::UInt16
            | ducklake_catalog::DuckLakeType::UInt32
            | ducklake_catalog::DuckLakeType::UInt64
            | ducklake_catalog::DuckLakeType::Float32
            | ducklake_catalog::DuckLakeType::Float64,
            DuckDbValue::UTinyInt(value),
        ) => Ok(value.to_string()),
        (
            ducklake_catalog::DuckLakeType::Int8
            | ducklake_catalog::DuckLakeType::Int16
            | ducklake_catalog::DuckLakeType::Int32
            | ducklake_catalog::DuckLakeType::Int64
            | ducklake_catalog::DuckLakeType::UInt8
            | ducklake_catalog::DuckLakeType::UInt16
            | ducklake_catalog::DuckLakeType::UInt32
            | ducklake_catalog::DuckLakeType::UInt64
            | ducklake_catalog::DuckLakeType::Float32
            | ducklake_catalog::DuckLakeType::Float64,
            DuckDbValue::USmallInt(value),
        ) => Ok(value.to_string()),
        (
            ducklake_catalog::DuckLakeType::Int8
            | ducklake_catalog::DuckLakeType::Int16
            | ducklake_catalog::DuckLakeType::Int32
            | ducklake_catalog::DuckLakeType::Int64
            | ducklake_catalog::DuckLakeType::UInt8
            | ducklake_catalog::DuckLakeType::UInt16
            | ducklake_catalog::DuckLakeType::UInt32
            | ducklake_catalog::DuckLakeType::UInt64
            | ducklake_catalog::DuckLakeType::Float32
            | ducklake_catalog::DuckLakeType::Float64,
            DuckDbValue::UInt(value),
        ) => Ok(value.to_string()),
        (
            ducklake_catalog::DuckLakeType::Int8
            | ducklake_catalog::DuckLakeType::Int16
            | ducklake_catalog::DuckLakeType::Int32
            | ducklake_catalog::DuckLakeType::Int64
            | ducklake_catalog::DuckLakeType::UInt8
            | ducklake_catalog::DuckLakeType::UInt16
            | ducklake_catalog::DuckLakeType::UInt32
            | ducklake_catalog::DuckLakeType::UInt64
            | ducklake_catalog::DuckLakeType::Float32
            | ducklake_catalog::DuckLakeType::Float64,
            DuckDbValue::UBigInt(value),
        ) => Ok(value.to_string()),
        (
            ducklake_catalog::DuckLakeType::Int8
            | ducklake_catalog::DuckLakeType::Int16
            | ducklake_catalog::DuckLakeType::Int32
            | ducklake_catalog::DuckLakeType::Int64
            | ducklake_catalog::DuckLakeType::UInt8
            | ducklake_catalog::DuckLakeType::UInt16
            | ducklake_catalog::DuckLakeType::UInt32
            | ducklake_catalog::DuckLakeType::UInt64
            | ducklake_catalog::DuckLakeType::Float32
            | ducklake_catalog::DuckLakeType::Float64,
            DuckDbValue::Float(value),
        ) => Ok(value.to_string()),
        (
            ducklake_catalog::DuckLakeType::Int8
            | ducklake_catalog::DuckLakeType::Int16
            | ducklake_catalog::DuckLakeType::Int32
            | ducklake_catalog::DuckLakeType::Int64
            | ducklake_catalog::DuckLakeType::UInt8
            | ducklake_catalog::DuckLakeType::UInt16
            | ducklake_catalog::DuckLakeType::UInt32
            | ducklake_catalog::DuckLakeType::UInt64
            | ducklake_catalog::DuckLakeType::Float32
            | ducklake_catalog::DuckLakeType::Float64,
            DuckDbValue::Double(value),
        ) => Ok(value.to_string()),
        (
            ducklake_catalog::DuckLakeType::Decimal { .. }
            | ducklake_catalog::DuckLakeType::Int128
            | ducklake_catalog::DuckLakeType::UInt128,
            DuckDbValue::Decimal(value),
        ) => Ok(crate::json_string(&value.to_string())),
        (
            ducklake_catalog::DuckLakeType::Varchar | ducklake_catalog::DuckLakeType::Json,
            DuckDbValue::Text(value) | DuckDbValue::Enum(value),
        ) => Ok(crate::json_string(value)),
        (ducklake_catalog::DuckLakeType::Date, DuckDbValue::Date32(value)) => {
            format_inlined_date32(*value, "DuckDB").map(|value| crate::json_string(&value))
        }
        (ducklake_catalog::DuckLakeType::Time, DuckDbValue::Time64(source_unit, value)) => {
            format_duckdb_time64(*source_unit, *value, TimeUnit::Microsecond)
                .map(|value| crate::json_string(&value))
        }
        (ducklake_catalog::DuckLakeType::TimeNs, DuckDbValue::Time64(source_unit, value)) => {
            format_duckdb_time64(*source_unit, *value, TimeUnit::Nanosecond)
                .map(|value| crate::json_string(&value))
        }
        (ducklake_catalog::DuckLakeType::TimeTz, DuckDbValue::Time64(source_unit, value)) => {
            format_duckdb_time64(*source_unit, *value, arrow_unit_from_duckdb(*source_unit))
                .map(|value| crate::json_string(&value))
        }
        (
            ducklake_catalog::DuckLakeType::TimestampS,
            DuckDbValue::Timestamp(source_unit, value),
        ) => format_duckdb_timestamp(*source_unit, *value, TimeUnit::Second, false)
            .map(|value| crate::json_string(&value)),
        (
            ducklake_catalog::DuckLakeType::TimestampMs,
            DuckDbValue::Timestamp(source_unit, value),
        ) => format_duckdb_timestamp(*source_unit, *value, TimeUnit::Millisecond, false)
            .map(|value| crate::json_string(&value)),
        (
            ducklake_catalog::DuckLakeType::Timestamp | ducklake_catalog::DuckLakeType::TimestampUs,
            DuckDbValue::Timestamp(source_unit, value),
        ) => format_duckdb_timestamp(*source_unit, *value, TimeUnit::Microsecond, false)
            .map(|value| crate::json_string(&value)),
        (
            ducklake_catalog::DuckLakeType::TimestampTz,
            DuckDbValue::Timestamp(source_unit, value),
        ) => format_duckdb_timestamp(*source_unit, *value, TimeUnit::Microsecond, true)
            .map(|value| crate::json_string(&value)),
        (
            ducklake_catalog::DuckLakeType::TimestampNs,
            DuckDbValue::Timestamp(source_unit, value),
        ) => format_duckdb_timestamp(*source_unit, *value, TimeUnit::Nanosecond, false)
            .map(|value| crate::json_string(&value)),
        (
            ducklake_catalog::DuckLakeType::Interval,
            DuckDbValue::Interval {
                months,
                days,
                nanos,
            },
        ) => Ok(crate::json_string(&format!(
            "{months} months {days} days {nanos} nanoseconds"
        ))),
        (
            ducklake_catalog::DuckLakeType::Blob
            | ducklake_catalog::DuckLakeType::Variant
            | ducklake_catalog::DuckLakeType::Geometry,
            DuckDbValue::Blob(value),
        ) => Ok(crate::json_string(&crate::hex_encode_bytes(value))),
        (ducklake_catalog::DuckLakeType::Uuid, DuckDbValue::Text(value)) => Ok(crate::json_string(
            &crate::hex_encode_bytes(&parse_inlined_uuid(value, "DuckDB")?),
        )),
        (_, DuckDbValue::Union(value)) => duckdb_value_to_json(field, value),
        _ => Err(StorageError::Backend(format!(
            "cannot decode DuckDB nested value {value:?} as DuckLake type {}",
            field.ducklake_type
        ))),
    }
}

fn duckdb_map_key_to_json(field: &SchemaField, value: &DuckDbValue) -> Result<String> {
    let value = duckdb_scalar_value_to_json(field, value)?;
    match &field.ducklake_type {
        ducklake_catalog::DuckLakeType::Varchar
        | ducklake_catalog::DuckLakeType::Json
        | ducklake_catalog::DuckLakeType::Uuid => Ok(value),
        _ => Ok(crate::json_string(value.trim_matches('"'))),
    }
}

fn sql_date32(value: ValueRef<'_>) -> Result<Option<i32>> {
    match value {
        ValueRef::Null => Ok(None),
        ValueRef::Date32(value) => Ok(Some(value)),
        value => decode_inlined_scalar_string(duckdb_scalar(value), "DuckDB")?
            .map(|value| parse_inlined_date32(&value, "DuckDB", true))
            .transpose(),
    }
}

fn sql_time64(value: ValueRef<'_>, target_unit: TimeUnit) -> Result<Option<i64>> {
    match value {
        ValueRef::Null => Ok(None),
        ValueRef::Time64(source_unit, value) => {
            duckdb_temporal_to_arrow_unit(value, source_unit, target_unit, "time").map(Some)
        }
        value => decode_inlined_scalar_string(duckdb_scalar(value), "DuckDB")?
            .map(|value| parse_inlined_time64(&value, target_unit, "DuckDB"))
            .transpose(),
    }
}

fn sql_timestamp(value: ValueRef<'_>, target_unit: TimeUnit) -> Result<Option<i64>> {
    match value {
        ValueRef::Null => Ok(None),
        ValueRef::Timestamp(source_unit, value) => {
            duckdb_temporal_to_arrow_unit(value, source_unit, target_unit, "timestamp").map(Some)
        }
        value => decode_inlined_scalar_string(duckdb_scalar(value), "DuckDB")?
            .map(|value| parse_inlined_timestamp(&value, target_unit, "DuckDB", true))
            .transpose(),
    }
}

fn duckdb_temporal_to_arrow_unit(
    value: i64,
    source_unit: DuckDbTimeUnit,
    target_unit: TimeUnit,
    context: &str,
) -> Result<i64> {
    let nanos = i128::from(value)
        .checked_mul(duckdb_time_unit_nanos(source_unit))
        .ok_or_else(|| {
            StorageError::Backend(format!("DuckDB {context} {value} is out of range"))
        })?;
    let target_nanos = crate::arrow_time_unit_nanos(target_unit);
    if nanos % target_nanos != 0 {
        return Err(StorageError::Backend(format!(
            "DuckDB {context} {value} cannot be represented as {target_unit:?} without precision loss"
        )));
    }
    i64::try_from(nanos / target_nanos)
        .map_err(|_| StorageError::Backend(format!("DuckDB {context} {value} is out of range")))
}

fn duckdb_time_unit_nanos(unit: DuckDbTimeUnit) -> i128 {
    match unit {
        DuckDbTimeUnit::Second => 1_000_000_000,
        DuckDbTimeUnit::Millisecond => 1_000_000,
        DuckDbTimeUnit::Microsecond => 1_000,
        DuckDbTimeUnit::Nanosecond => 1,
    }
}

fn arrow_unit_from_duckdb(unit: DuckDbTimeUnit) -> TimeUnit {
    match unit {
        DuckDbTimeUnit::Second => TimeUnit::Second,
        DuckDbTimeUnit::Millisecond => TimeUnit::Millisecond,
        DuckDbTimeUnit::Microsecond => TimeUnit::Microsecond,
        DuckDbTimeUnit::Nanosecond => TimeUnit::Nanosecond,
    }
}

fn format_duckdb_time64(
    source_unit: DuckDbTimeUnit,
    value: i64,
    target_unit: TimeUnit,
) -> Result<String> {
    let value = duckdb_temporal_to_arrow_unit(value, source_unit, target_unit, "time")?;
    crate::format_time64(value, target_unit, "DuckDB")
}

fn format_duckdb_timestamp(
    source_unit: DuckDbTimeUnit,
    value: i64,
    target_unit: TimeUnit,
    utc_suffix: bool,
) -> Result<String> {
    let value = duckdb_temporal_to_arrow_unit(value, source_unit, target_unit, "timestamp")?;
    crate::format_timestamp(value, target_unit, utc_suffix, "DuckDB")
}

fn sql_interval(value: ValueRef<'_>) -> Result<Option<IntervalMonthDayNano>> {
    match value {
        ValueRef::Null => Ok(None),
        ValueRef::Interval {
            months,
            days,
            nanos,
        } => Ok(Some(IntervalMonthDayNano::new(months, days, nanos))),
        value => decode_inlined_scalar_string(duckdb_scalar(value), "DuckDB")?
            .map(|value| parse_interval_month_day_nano(&value, "DuckDB"))
            .transpose(),
    }
}

impl_embedded_sql_err!(fn sql_err, error = duckdb::Error);

#[cfg(test)]
#[path = "duckdb_file/tests.rs"]
mod tests;
