#![allow(dead_code)]
// Inlined-data conversion is deliberately centralized for all catalog
// backends. Each feature profile uses a different subset of the encoders,
// decoders, and SQL builders in this module.

use std::future::Future;
use std::io::Cursor;
use std::sync::Arc;

use arrow::array::{
    Array, ArrayRef, BinaryArray, BooleanArray, Date32Array, Decimal128Array, Decimal256Array,
    FixedSizeBinaryArray, Float32Array, Float64Array, Int8Array, Int16Array, Int32Array,
    Int64Array, IntervalMonthDayNanoArray, LargeListArray, ListArray, MapArray, StringArray,
    StructArray, Time64MicrosecondArray, Time64NanosecondArray, TimestampMicrosecondArray,
    TimestampMillisecondArray, TimestampNanosecondArray, TimestampSecondArray, UInt8Array,
    UInt16Array, UInt32Array, UInt64Array, new_empty_array,
};
use arrow::datatypes::{
    DataType, Field, IntervalMonthDayNano, IntervalUnit, Schema, TimeUnit, i256,
};
use arrow::json::ReaderBuilder;
use arrow::record_batch::RecordBatch;
use chrono::{DateTime, Duration, NaiveDate, NaiveDateTime, NaiveTime, Timelike, Utc};
use ducklake_catalog::{DuckLakeType, SchemaField, TableId};

pub(crate) use crate::ducklake_sql_string_literal as sql_string_literal;
use crate::{
    FlushedInlinedDataCommit, FlushedInlinedFileDeletion, FlushedInlinedFileDeletionCommit,
    InlinedDataBatch, InlinedDataDeletionCommit, InlinedFileDeletionCommit, Result, StorageError,
    TableSchema,
};

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum InlinedIntervalFormat {
    PreserveNanoseconds,
    MicrosecondsOnly(&'static str),
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum InlinedSqlDialect {
    Sqlite,
    DuckDbFile,
    Postgres,
    Quack,
}

pub(crate) enum InlinedScalarValue {
    Null,
    Boolean(bool),
    Int8(i8),
    Int16(i16),
    Int32(i32),
    Int64(i64),
    UInt8(u8),
    UInt16(u16),
    UInt32(u32),
    UInt64(u64),
    Float32(f32),
    Float64(f64),
    Text(String),
    Bytes(Vec<u8>),
}

pub(crate) trait InlinedScalarValueEncoder {
    type Value;

    fn null_value(&self, field: &SchemaField) -> Result<Self::Value>;

    fn boolean_value(&self, value: bool) -> Result<Self::Value>;

    fn int8_value(&self, value: i8) -> Result<Self::Value>;

    fn int16_value(&self, value: i16) -> Result<Self::Value>;

    fn int32_value(&self, value: i32) -> Result<Self::Value>;

    fn int64_value(&self, value: i64) -> Result<Self::Value>;

    fn uint8_value(&self, value: u8) -> Result<Self::Value>;

    fn uint16_value(&self, value: u16) -> Result<Self::Value>;

    fn uint32_value(&self, value: u32) -> Result<Self::Value>;

    fn uint64_value(&self, field: &SchemaField, value: u64) -> Result<Self::Value>;

    fn float32_value(&self, value: f32) -> Result<Self::Value>;

    fn float64_value(&self, value: f64) -> Result<Self::Value>;

    fn text_value(&self, field: &SchemaField, value: String) -> Result<Self::Value>;

    fn bytes_value(&self, field: &SchemaField, value: Vec<u8>) -> Result<Self::Value>;
}

macro_rules! impl_inlined_scalar_value_encoder {
    (
        impl $encoder:ty,
        Value = $value:ty,
        {
            null($null_field:ident) => $null_expr:expr,
            boolean($bool_value:ident) => $bool_expr:expr,
            int8($int8_value:ident) => $int8_expr:expr,
            int16($int16_value:ident) => $int16_expr:expr,
            int32($int32_value:ident) => $int32_expr:expr,
            int64($int64_value:ident) => $int64_expr:expr,
            uint8($uint8_value:ident) => $uint8_expr:expr,
            uint16($uint16_value:ident) => $uint16_expr:expr,
            uint32($uint32_value:ident) => $uint32_expr:expr,
            uint64($uint64_field:ident, $uint64_value:ident) => $uint64_expr:expr,
            float32($float32_value:ident) => $float32_expr:expr,
            float64($float64_value:ident) => $float64_expr:expr,
            text($text_field:ident, $text_value:ident) => $text_expr:expr,
            bytes($bytes_field:ident, $bytes_value:ident) => $bytes_expr:expr $(,)?
        }
    ) => {
        impl $crate::inlined_values::InlinedScalarValueEncoder for $encoder {
            type Value = $value;

            fn null_value(
                &self,
                field: &ducklake_catalog::SchemaField,
            ) -> $crate::Result<Self::Value> {
                let $null_field = field;
                $null_expr
            }

            fn boolean_value(&self, value: bool) -> $crate::Result<Self::Value> {
                let $bool_value = value;
                $bool_expr
            }

            fn int8_value(&self, value: i8) -> $crate::Result<Self::Value> {
                let $int8_value = value;
                $int8_expr
            }

            fn int16_value(&self, value: i16) -> $crate::Result<Self::Value> {
                let $int16_value = value;
                $int16_expr
            }

            fn int32_value(&self, value: i32) -> $crate::Result<Self::Value> {
                let $int32_value = value;
                $int32_expr
            }

            fn int64_value(&self, value: i64) -> $crate::Result<Self::Value> {
                let $int64_value = value;
                $int64_expr
            }

            fn uint8_value(&self, value: u8) -> $crate::Result<Self::Value> {
                let $uint8_value = value;
                $uint8_expr
            }

            fn uint16_value(&self, value: u16) -> $crate::Result<Self::Value> {
                let $uint16_value = value;
                $uint16_expr
            }

            fn uint32_value(&self, value: u32) -> $crate::Result<Self::Value> {
                let $uint32_value = value;
                $uint32_expr
            }

            fn uint64_value(
                &self,
                field: &ducklake_catalog::SchemaField,
                value: u64,
            ) -> $crate::Result<Self::Value> {
                let $uint64_field = field;
                let $uint64_value = value;
                $uint64_expr
            }

            fn float32_value(&self, value: f32) -> $crate::Result<Self::Value> {
                let $float32_value = value;
                $float32_expr
            }

            fn float64_value(&self, value: f64) -> $crate::Result<Self::Value> {
                let $float64_value = value;
                $float64_expr
            }

            fn text_value(
                &self,
                field: &ducklake_catalog::SchemaField,
                value: String,
            ) -> $crate::Result<Self::Value> {
                let $text_field = field;
                let $text_value = value;
                $text_expr
            }

            fn bytes_value(
                &self,
                field: &ducklake_catalog::SchemaField,
                value: Vec<u8>,
            ) -> $crate::Result<Self::Value> {
                let $bytes_field = field;
                let $bytes_value = value;
                $bytes_expr
            }
        }
    };
}

pub(crate) fn encode_inlined_scalar_value<E>(
    field: &SchemaField,
    array: &dyn Array,
    row: usize,
    context: &str,
    interval_format: InlinedIntervalFormat,
    encoder: &E,
) -> Result<E::Value>
where
    E: InlinedScalarValueEncoder + ?Sized,
{
    let value = inlined_scalar_value(field, array, row, context, interval_format)?;
    encode_inlined_scalar(field, value, encoder)
}

pub(crate) fn encode_inlined_scalar<E>(
    field: &SchemaField,
    value: InlinedScalarValue,
    encoder: &E,
) -> Result<E::Value>
where
    E: InlinedScalarValueEncoder + ?Sized,
{
    match value {
        InlinedScalarValue::Null => encoder.null_value(field),
        InlinedScalarValue::Boolean(value) => encoder.boolean_value(value),
        InlinedScalarValue::Int8(value) => encoder.int8_value(value),
        InlinedScalarValue::Int16(value) => encoder.int16_value(value),
        InlinedScalarValue::Int32(value) => encoder.int32_value(value),
        InlinedScalarValue::Int64(value) => encoder.int64_value(value),
        InlinedScalarValue::UInt8(value) => encoder.uint8_value(value),
        InlinedScalarValue::UInt16(value) => encoder.uint16_value(value),
        InlinedScalarValue::UInt32(value) => encoder.uint32_value(value),
        InlinedScalarValue::UInt64(value) => encoder.uint64_value(field, value),
        InlinedScalarValue::Float32(value) => encoder.float32_value(value),
        InlinedScalarValue::Float64(value) => encoder.float64_value(value),
        InlinedScalarValue::Text(value) => encoder.text_value(field, value),
        InlinedScalarValue::Bytes(value) => encoder.bytes_value(field, value),
    }
}

pub(crate) fn inlined_duckdb_literal_value(
    field: &SchemaField,
    array: &dyn Array,
    row: usize,
    context: &str,
    interval_format: InlinedIntervalFormat,
) -> Result<String> {
    if array.is_null(row) {
        return Ok("NULL".into());
    }

    match field.ducklake_type {
        DuckLakeType::Struct => inlined_struct_duckdb_literal(
            field,
            typed_array::<StructArray>(field, array)?,
            row,
            context,
            interval_format,
        ),
        DuckLakeType::List => {
            inlined_list_duckdb_literal(field, array, row, context, interval_format)
        }
        DuckLakeType::Map => inlined_map_duckdb_literal(
            field,
            typed_array::<MapArray>(field, array)?,
            row,
            context,
            interval_format,
        ),
        _ => {
            let value = inlined_scalar_value(field, array, row, context, interval_format)?;
            inlined_scalar_duckdb_literal(field, value)
        }
    }
}

fn inlined_struct_duckdb_literal(
    field: &SchemaField,
    array: &StructArray,
    row: usize,
    context: &str,
    interval_format: InlinedIntervalFormat,
) -> Result<String> {
    if array.columns().len() != field.children.len() {
        return Err(StorageError::Backend(format!(
            "DuckLake struct field {} has {} Arrow children but {} schema children",
            field.name,
            array.columns().len(),
            field.children.len()
        )));
    }
    let mut values = Vec::with_capacity(field.children.len());
    for (child, child_array) in field.children.iter().zip(array.columns()) {
        values.push(format!(
            "{}: {}",
            sql_string_literal(&child.name),
            inlined_duckdb_literal_value(
                child,
                child_array.as_ref(),
                row,
                context,
                interval_format,
            )?
        ));
    }
    Ok(format!("{{{}}}", values.join(", ")))
}

fn inlined_list_duckdb_literal(
    field: &SchemaField,
    array: &dyn Array,
    row: usize,
    context: &str,
    interval_format: InlinedIntervalFormat,
) -> Result<String> {
    let child = field.children.first().ok_or_else(|| {
        StorageError::Backend(format!("DuckLake list field {} has no child", field.name))
    })?;
    if let Some(array) = array.as_any().downcast_ref::<ListArray>() {
        let offsets = array.value_offsets();
        let start = usize::try_from(offsets[row]).map_err(|_| {
            StorageError::Backend(format!(
                "DuckLake list field {} has negative offset",
                field.name
            ))
        })?;
        let end = usize::try_from(offsets[row + 1]).map_err(|_| {
            StorageError::Backend(format!(
                "DuckLake list field {} has negative offset",
                field.name
            ))
        })?;
        return inlined_list_range_duckdb_literal(
            child,
            array.values().as_ref(),
            start,
            end,
            context,
            interval_format,
        );
    }
    if let Some(array) = array.as_any().downcast_ref::<LargeListArray>() {
        let offsets = array.value_offsets();
        let start = usize::try_from(offsets[row]).map_err(|_| {
            StorageError::Backend(format!(
                "DuckLake list field {} has negative offset",
                field.name
            ))
        })?;
        let end = usize::try_from(offsets[row + 1]).map_err(|_| {
            StorageError::Backend(format!(
                "DuckLake list field {} has negative offset",
                field.name
            ))
        })?;
        return inlined_list_range_duckdb_literal(
            child,
            array.values().as_ref(),
            start,
            end,
            context,
            interval_format,
        );
    }
    Err(inlined_append_type_error(field, array.data_type()))
}

fn inlined_list_range_duckdb_literal(
    child: &SchemaField,
    values: &dyn Array,
    start: usize,
    end: usize,
    context: &str,
    interval_format: InlinedIntervalFormat,
) -> Result<String> {
    let mut rendered = Vec::with_capacity(end.saturating_sub(start));
    for idx in start..end {
        rendered.push(inlined_duckdb_literal_value(
            child,
            values,
            idx,
            context,
            interval_format,
        )?);
    }
    Ok(format!("[{}]", rendered.join(", ")))
}

fn inlined_map_duckdb_literal(
    field: &SchemaField,
    array: &MapArray,
    row: usize,
    context: &str,
    interval_format: InlinedIntervalFormat,
) -> Result<String> {
    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 offsets = array.value_offsets();
    let start = usize::try_from(offsets[row]).map_err(|_| {
        StorageError::Backend(format!(
            "DuckLake map field {} has negative offset",
            field.name
        ))
    })?;
    let end = usize::try_from(offsets[row + 1]).map_err(|_| {
        StorageError::Backend(format!(
            "DuckLake map field {} has negative offset",
            field.name
        ))
    })?;
    let keys = array.keys();
    let values = array.values();
    let mut rendered = Vec::with_capacity(end.saturating_sub(start));
    for idx in start..end {
        let key =
            inlined_duckdb_literal_value(key_field, keys.as_ref(), idx, context, interval_format)?;
        if key.eq_ignore_ascii_case("null") {
            return Err(StorageError::Backend(format!(
                "DuckLake map field {} contains NULL key",
                field.name
            )));
        }
        let value = inlined_duckdb_literal_value(
            value_field,
            values.as_ref(),
            idx,
            context,
            interval_format,
        )?;
        rendered.push(format!("{key}={value}"));
    }
    Ok(format!("{{{}}}", rendered.join(", ")))
}

fn inlined_scalar_duckdb_literal(field: &SchemaField, value: InlinedScalarValue) -> Result<String> {
    Ok(match value {
        InlinedScalarValue::Null => "NULL".into(),
        InlinedScalarValue::Boolean(value) => value.to_string(),
        InlinedScalarValue::Int8(value) => value.to_string(),
        InlinedScalarValue::Int16(value) => value.to_string(),
        InlinedScalarValue::Int32(value) => value.to_string(),
        InlinedScalarValue::Int64(value) => value.to_string(),
        InlinedScalarValue::UInt8(value) => value.to_string(),
        InlinedScalarValue::UInt16(value) => value.to_string(),
        InlinedScalarValue::UInt32(value) => value.to_string(),
        InlinedScalarValue::UInt64(value) => value.to_string(),
        InlinedScalarValue::Float32(value) if value.is_finite() => value.to_string(),
        InlinedScalarValue::Float64(value) if value.is_finite() => value.to_string(),
        InlinedScalarValue::Float32(value) => sql_cast_string_literal(&value.to_string(), "FLOAT"),
        InlinedScalarValue::Float64(value) => sql_cast_string_literal(&value.to_string(), "DOUBLE"),
        InlinedScalarValue::Text(value) => match field.ducklake_type {
            DuckLakeType::Decimal { .. }
            | DuckLakeType::Int128
            | DuckLakeType::UInt128
            | DuckLakeType::Date
            | DuckLakeType::Time
            | DuckLakeType::TimeNs
            | DuckLakeType::Timestamp
            | DuckLakeType::TimestampS
            | DuckLakeType::TimestampMs
            | DuckLakeType::TimestampUs
            | DuckLakeType::TimestampNs
            | DuckLakeType::TimestampTz
            | DuckLakeType::TimeTz
            | DuckLakeType::Interval
            | DuckLakeType::Varchar
            | DuckLakeType::Json
            | DuckLakeType::Uuid => sql_string_literal(&value),
            ref other => {
                return Err(StorageError::Unsupported(format!(
                    "DuckLake inlined literal cannot encode type {other} as text"
                )));
            }
        },
        InlinedScalarValue::Bytes(value) => sql_string_literal(&hex_encode_bytes(&value)),
    })
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum InlinedNestedValuesFormat {
    DuckDbLiteral,
    Json,
}

pub(crate) enum InlinedColumnBuffer {
    Boolean(Vec<Option<bool>>),
    Int8(Vec<Option<i8>>),
    Int16(Vec<Option<i16>>),
    Int32(Vec<Option<i32>>),
    Int64(Vec<Option<i64>>),
    UInt8(Vec<Option<u8>>),
    UInt16(Vec<Option<u16>>),
    UInt32(Vec<Option<u32>>),
    UInt64(Vec<Option<u64>>),
    Float32(Vec<Option<f32>>),
    Float64(Vec<Option<f64>>),
    Decimal128 {
        values: Vec<Option<i128>>,
        precision: u8,
        scale: u8,
    },
    Decimal256 {
        values: Vec<Option<i256>>,
        precision: u8,
        scale: u8,
    },
    Date32(Vec<Option<i32>>),
    TimeMicros(Vec<Option<i64>>),
    TimeNanos(Vec<Option<i64>>),
    TimestampSeconds(Vec<Option<i64>>),
    TimestampMillis(Vec<Option<i64>>),
    TimestampMicros {
        values: Vec<Option<i64>>,
        timezone: Option<&'static str>,
    },
    TimestampNanos(Vec<Option<i64>>),
    Interval(Vec<Option<arrow::datatypes::IntervalMonthDayNano>>),
    Binary(Vec<Option<Vec<u8>>>),
    Uuid(Vec<Option<[u8; 16]>>),
    Utf8(Vec<Option<String>>),
    Utf8HexBytes(Vec<Option<String>>),
    Nested {
        field: SchemaField,
        values: Vec<Option<String>>,
        format: InlinedNestedValuesFormat,
    },
}

pub(crate) trait InlinedSqlValueDecoder<V> {
    fn backend_name(&self) -> &'static str;

    fn decode_bool(&self, value: V) -> Result<Option<bool>>;

    fn decode_i64(&self, value: V) -> Result<Option<i64>>;

    fn decode_u64(&self, value: V) -> Result<Option<u64>>;

    fn decode_f64(&self, value: V) -> Result<Option<f64>>;

    fn decode_string(&self, value: V) -> Result<Option<String>>;

    fn decode_bytes(&self, value: V) -> Result<Option<Vec<u8>>>;

    fn decode_decimal128(&self, value: V, scale: u8) -> Result<Option<i128>> {
        self.decode_string(value)?
            .map(|value| parse_inlined_decimal128(&value, scale, self.backend_name()))
            .transpose()
    }

    fn decode_decimal256(&self, value: V, scale: u8) -> Result<Option<i256>> {
        self.decode_string(value)?
            .map(|value| parse_inlined_decimal256(&value, scale, self.backend_name()))
            .transpose()
    }

    fn decode_uuid(&self, value: V) -> Result<Option<[u8; 16]>> {
        self.decode_string(value)?
            .map(|value| parse_inlined_uuid(&value, self.backend_name()))
            .transpose()
    }

    fn decode_utf8_hex_string(&self, value: V) -> Result<Option<String>> {
        self.decode_string(value)
    }

    fn decode_date32(&self, value: V) -> Result<Option<i32>> {
        self.decode_string(value)?
            .map(|value| parse_inlined_date32(&value, self.backend_name(), true))
            .transpose()
    }

    fn decode_time64(&self, value: V, unit: TimeUnit) -> Result<Option<i64>> {
        self.decode_string(value)?
            .map(|value| parse_inlined_time64(&value, unit, self.backend_name()))
            .transpose()
    }

    fn decode_timestamp(&self, value: V, unit: TimeUnit) -> Result<Option<i64>> {
        self.decode_string(value)?
            .map(|value| parse_inlined_timestamp(&value, unit, self.backend_name(), true))
            .transpose()
    }

    fn decode_interval(&self, value: V) -> Result<Option<IntervalMonthDayNano>> {
        self.decode_string(value)?
            .map(|value| crate::parse_interval_month_day_nano(&value, self.backend_name()))
            .transpose()
    }

    fn decode_nested(&self, _field: &SchemaField, value: V) -> Result<Option<String>> {
        self.decode_string(value)
    }
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
macro_rules! impl_inlined_scalar_decoder_methods {
    (
        value = $value:ty,
        backend = $backend:expr,
        scalar = $scalar:path $(,)?
    ) => {
        fn backend_name(&self) -> &'static str {
            $backend
        }

        fn decode_bool(&self, value: $value) -> $crate::Result<Option<bool>> {
            $crate::inlined_values::decode_inlined_scalar_bool($scalar(value), $backend)
        }

        fn decode_i64(&self, value: $value) -> $crate::Result<Option<i64>> {
            $crate::inlined_values::decode_inlined_scalar_i64($scalar(value), $backend)
        }

        fn decode_u64(&self, value: $value) -> $crate::Result<Option<u64>> {
            $crate::inlined_values::decode_inlined_scalar_u64($scalar(value), $backend)
        }

        fn decode_f64(&self, value: $value) -> $crate::Result<Option<f64>> {
            $crate::inlined_values::decode_inlined_scalar_f64($scalar(value), $backend)
        }

        fn decode_bytes(&self, value: $value) -> $crate::Result<Option<Vec<u8>>> {
            $crate::inlined_values::decode_inlined_scalar_bytes($scalar(value), $backend)
        }
    };
}

#[cfg(any(feature = "sqlite", test))]
macro_rules! impl_inlined_scalar_string_decoder_method {
    (
        value = $value:ty,
        backend = $backend:expr,
        scalar = $scalar:path $(,)?
    ) => {
        fn decode_string(&self, value: $value) -> $crate::Result<Option<String>> {
            $crate::inlined_values::decode_inlined_scalar_string($scalar(value), $backend)
        }
    };
}

#[derive(Debug, Clone, Copy)]
pub(crate) struct TextInlinedValueDecoder {
    backend_name: &'static str,
    parse_numeric_date: bool,
    parse_timestamp_offsets: bool,
}

impl TextInlinedValueDecoder {
    pub(crate) const fn new(
        backend_name: &'static str,
        parse_numeric_date: bool,
        parse_timestamp_offsets: bool,
    ) -> Self {
        Self {
            backend_name,
            parse_numeric_date,
            parse_timestamp_offsets,
        }
    }
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) enum InlinedSqlScalarRef<'a> {
    Null,
    Boolean(bool),
    Signed(i128),
    Unsigned(u64),
    Float(f64),
    DecimalText(String),
    Text(&'a [u8]),
    Bytes(&'a [u8]),
    Unsupported(String),
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn decode_inlined_scalar_bool(
    value: InlinedSqlScalarRef<'_>,
    backend: &str,
) -> Result<Option<bool>> {
    match value {
        InlinedSqlScalarRef::Null => Ok(None),
        InlinedSqlScalarRef::Boolean(value) => Ok(Some(value)),
        InlinedSqlScalarRef::Signed(value) => Ok(Some(value != 0)),
        InlinedSqlScalarRef::Unsigned(value) => Ok(Some(value != 0)),
        InlinedSqlScalarRef::Text(value) => parse_inlined_utf8_text(value, backend)
            .and_then(|value| parse_inlined_bool_text(value, backend))
            .map(Some),
        value => Err(inlined_scalar_type_error(value, backend, "boolean")),
    }
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn decode_inlined_scalar_i64(
    value: InlinedSqlScalarRef<'_>,
    backend: &str,
) -> Result<Option<i64>> {
    match value {
        InlinedSqlScalarRef::Null => Ok(None),
        InlinedSqlScalarRef::Boolean(value) => Ok(Some(i64::from(value))),
        InlinedSqlScalarRef::Signed(value) => i64::try_from(value).map(Some).map_err(|_| {
            StorageError::Backend(format!("{backend} integer {value} is out of range"))
        }),
        InlinedSqlScalarRef::Unsigned(value) => i64::try_from(value).map(Some).map_err(|_| {
            StorageError::Backend(format!("{backend} integer {value} is out of range"))
        }),
        InlinedSqlScalarRef::Text(value) => parse_inlined_utf8_text(value, backend)
            .and_then(|value| parse_inlined_i64_text(value, backend))
            .map(Some),
        value => Err(inlined_scalar_type_error(value, backend, "integer")),
    }
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn decode_inlined_scalar_u64(
    value: InlinedSqlScalarRef<'_>,
    backend: &str,
) -> Result<Option<u64>> {
    match value {
        InlinedSqlScalarRef::Null => Ok(None),
        InlinedSqlScalarRef::Boolean(value) => Ok(Some(u64::from(value))),
        InlinedSqlScalarRef::Signed(value) => u64::try_from(value).map(Some).map_err(|_| {
            StorageError::Backend(format!(
                "{backend} integer {value} is out of range for UInt64"
            ))
        }),
        InlinedSqlScalarRef::Unsigned(value) => Ok(Some(value)),
        InlinedSqlScalarRef::Text(value) => parse_inlined_utf8_text(value, backend)
            .and_then(|value| parse_inlined_u64_text(value, backend))
            .map(Some),
        value => Err(inlined_scalar_type_error(
            value,
            backend,
            "unsigned integer",
        )),
    }
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn decode_inlined_scalar_f64(
    value: InlinedSqlScalarRef<'_>,
    backend: &str,
) -> Result<Option<f64>> {
    match value {
        InlinedSqlScalarRef::Null => Ok(None),
        InlinedSqlScalarRef::Boolean(value) => Ok(Some(if value { 1.0 } else { 0.0 })),
        InlinedSqlScalarRef::Signed(value) => Ok(Some(value as f64)),
        InlinedSqlScalarRef::Unsigned(value) => Ok(Some(value as f64)),
        InlinedSqlScalarRef::Float(value) => Ok(Some(value)),
        InlinedSqlScalarRef::DecimalText(value) => value.parse::<f64>().map(Some).map_err(|_| {
            StorageError::Backend(format!("cannot decode {backend} decimal as float"))
        }),
        InlinedSqlScalarRef::Text(value) => parse_inlined_utf8_text(value, backend)
            .and_then(|value| parse_inlined_f64_text(value, backend))
            .map(Some),
        value => Err(inlined_scalar_type_error(value, backend, "float")),
    }
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn decode_inlined_scalar_string(
    value: InlinedSqlScalarRef<'_>,
    backend: &str,
) -> Result<Option<String>> {
    match value {
        InlinedSqlScalarRef::Null => Ok(None),
        InlinedSqlScalarRef::Boolean(value) => Ok(Some(value.to_string())),
        InlinedSqlScalarRef::Signed(value) => Ok(Some(value.to_string())),
        InlinedSqlScalarRef::Unsigned(value) => Ok(Some(value.to_string())),
        InlinedSqlScalarRef::Float(value) => Ok(Some(value.to_string())),
        InlinedSqlScalarRef::DecimalText(value) => Ok(Some(value)),
        InlinedSqlScalarRef::Text(value) => {
            Ok(Some(parse_inlined_utf8_text(value, backend)?.into()))
        }
        value => Err(inlined_scalar_type_error(value, backend, "string")),
    }
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn decode_inlined_scalar_bytes(
    value: InlinedSqlScalarRef<'_>,
    backend: &str,
) -> Result<Option<Vec<u8>>> {
    match value {
        InlinedSqlScalarRef::Null => Ok(None),
        InlinedSqlScalarRef::Bytes(value) => Ok(Some(value.to_vec())),
        InlinedSqlScalarRef::Text(value) => parse_inlined_utf8_text(value, backend)
            .and_then(|value| parse_inlined_hex_bytes(value, backend))
            .map(Some),
        value => Err(inlined_scalar_type_error(value, backend, "bytes")),
    }
}

impl InlinedSqlValueDecoder<Option<String>> for TextInlinedValueDecoder {
    fn backend_name(&self) -> &'static str {
        self.backend_name
    }

    fn decode_bool(&self, value: Option<String>) -> Result<Option<bool>> {
        value
            .map(|value| parse_inlined_bool_text(&value, self.backend_name))
            .transpose()
    }

    fn decode_i64(&self, value: Option<String>) -> Result<Option<i64>> {
        value
            .map(|value| parse_inlined_i64_text(&value, self.backend_name))
            .transpose()
    }

    fn decode_u64(&self, value: Option<String>) -> Result<Option<u64>> {
        value
            .map(|value| parse_inlined_u64_text(&value, self.backend_name))
            .transpose()
    }

    fn decode_f64(&self, value: Option<String>) -> Result<Option<f64>> {
        value
            .map(|value| parse_inlined_f64_text(&value, self.backend_name))
            .transpose()
    }

    fn decode_string(&self, value: Option<String>) -> Result<Option<String>> {
        Ok(value)
    }

    fn decode_bytes(&self, value: Option<String>) -> Result<Option<Vec<u8>>> {
        value
            .map(|value| parse_inlined_hex_bytes(&value, self.backend_name))
            .transpose()
    }

    fn decode_utf8_hex_string(&self, value: Option<String>) -> Result<Option<String>> {
        value
            .map(|value| parse_inlined_utf8_hex_bytes(&value, self.backend_name))
            .transpose()
    }

    fn decode_date32(&self, value: Option<String>) -> Result<Option<i32>> {
        value
            .map(|value| parse_inlined_date32(&value, self.backend_name, self.parse_numeric_date))
            .transpose()
    }

    fn decode_timestamp(&self, value: Option<String>, unit: TimeUnit) -> Result<Option<i64>> {
        value
            .map(|value| {
                parse_inlined_timestamp(
                    &value,
                    unit,
                    self.backend_name,
                    self.parse_timestamp_offsets,
                )
            })
            .transpose()
    }
}

pub(crate) fn push_decoded_inlined_column<V, D>(
    column: &mut InlinedColumnBuffer,
    value: V,
    decoder: &D,
) -> Result<()>
where
    D: InlinedSqlValueDecoder<V> + ?Sized,
{
    match column {
        InlinedColumnBuffer::Boolean(values) => values.push(decoder.decode_bool(value)?),
        InlinedColumnBuffer::Int8(values) => values.push(
            decoder
                .decode_i64(value)?
                .map(|value| cast_i64_inlined::<i8>(value, decoder.backend_name(), "Int8"))
                .transpose()?,
        ),
        InlinedColumnBuffer::Int16(values) => values.push(
            decoder
                .decode_i64(value)?
                .map(|value| cast_i64_inlined::<i16>(value, decoder.backend_name(), "Int16"))
                .transpose()?,
        ),
        InlinedColumnBuffer::Int32(values) => values.push(
            decoder
                .decode_i64(value)?
                .map(|value| cast_i64_inlined::<i32>(value, decoder.backend_name(), "Int32"))
                .transpose()?,
        ),
        InlinedColumnBuffer::Int64(values) => values.push(decoder.decode_i64(value)?),
        InlinedColumnBuffer::UInt8(values) => values.push(
            decoder
                .decode_u64(value)?
                .map(|value| cast_u64_inlined::<u8>(value, decoder.backend_name(), "UInt8"))
                .transpose()?,
        ),
        InlinedColumnBuffer::UInt16(values) => values.push(
            decoder
                .decode_u64(value)?
                .map(|value| cast_u64_inlined::<u16>(value, decoder.backend_name(), "UInt16"))
                .transpose()?,
        ),
        InlinedColumnBuffer::UInt32(values) => values.push(
            decoder
                .decode_u64(value)?
                .map(|value| cast_u64_inlined::<u32>(value, decoder.backend_name(), "UInt32"))
                .transpose()?,
        ),
        InlinedColumnBuffer::UInt64(values) => values.push(decoder.decode_u64(value)?),
        InlinedColumnBuffer::Float32(values) => {
            values.push(decoder.decode_f64(value)?.map(|value| value as f32))
        }
        InlinedColumnBuffer::Float64(values) => values.push(decoder.decode_f64(value)?),
        InlinedColumnBuffer::Decimal128 { values, scale, .. } => {
            values.push(decoder.decode_decimal128(value, *scale)?);
        }
        InlinedColumnBuffer::Decimal256 { values, scale, .. } => {
            values.push(decoder.decode_decimal256(value, *scale)?);
        }
        InlinedColumnBuffer::Date32(values) => values.push(decoder.decode_date32(value)?),
        InlinedColumnBuffer::TimeMicros(values) => {
            values.push(decoder.decode_time64(value, TimeUnit::Microsecond)?);
        }
        InlinedColumnBuffer::TimeNanos(values) => {
            values.push(decoder.decode_time64(value, TimeUnit::Nanosecond)?);
        }
        InlinedColumnBuffer::TimestampSeconds(values) => {
            values.push(decoder.decode_timestamp(value, TimeUnit::Second)?);
        }
        InlinedColumnBuffer::TimestampMillis(values) => {
            values.push(decoder.decode_timestamp(value, TimeUnit::Millisecond)?);
        }
        InlinedColumnBuffer::TimestampMicros { values, .. } => {
            values.push(decoder.decode_timestamp(value, TimeUnit::Microsecond)?);
        }
        InlinedColumnBuffer::TimestampNanos(values) => {
            values.push(decoder.decode_timestamp(value, TimeUnit::Nanosecond)?);
        }
        InlinedColumnBuffer::Interval(values) => values.push(decoder.decode_interval(value)?),
        InlinedColumnBuffer::Binary(values) => values.push(decoder.decode_bytes(value)?),
        InlinedColumnBuffer::Uuid(values) => {
            values.push(decoder.decode_uuid(value)?);
        }
        InlinedColumnBuffer::Utf8(values) => values.push(decoder.decode_string(value)?),
        InlinedColumnBuffer::Utf8HexBytes(values) => {
            values.push(decoder.decode_utf8_hex_string(value)?);
        }
        InlinedColumnBuffer::Nested { field, values, .. } => {
            values.push(decoder.decode_nested(field, value)?);
        }
    }
    Ok(())
}

fn cast_i64_inlined<T>(value: i64, backend: &str, target: &str) -> Result<T>
where
    T: TryFrom<i64>,
{
    T::try_from(value).map_err(|_| {
        StorageError::Backend(format!(
            "{backend} integer {value} is out of range for {target}"
        ))
    })
}

fn cast_u64_inlined<T>(value: u64, backend: &str, target: &str) -> Result<T>
where
    T: TryFrom<u64>,
{
    T::try_from(value).map_err(|_| {
        StorageError::Backend(format!(
            "{backend} unsigned integer {value} is out of range for {target}"
        ))
    })
}

impl InlinedColumnBuffer {
    pub(crate) fn new(field: &SchemaField, dialect: InlinedSqlDialect) -> Result<Self> {
        if !field.children.is_empty() {
            let format = match dialect {
                InlinedSqlDialect::DuckDbFile | InlinedSqlDialect::Quack => {
                    InlinedNestedValuesFormat::Json
                }
                InlinedSqlDialect::Sqlite | InlinedSqlDialect::Postgres => {
                    InlinedNestedValuesFormat::DuckDbLiteral
                }
            };
            return Ok(Self::Nested {
                field: field.clone(),
                values: Vec::new(),
                format,
            });
        }
        match &field.ducklake_type {
            DuckLakeType::Boolean => Ok(Self::Boolean(Vec::new())),
            DuckLakeType::Int8 => Ok(Self::Int8(Vec::new())),
            DuckLakeType::Int16 => Ok(Self::Int16(Vec::new())),
            DuckLakeType::Int32 => Ok(Self::Int32(Vec::new())),
            DuckLakeType::Int64 => Ok(Self::Int64(Vec::new())),
            DuckLakeType::UInt8 => Ok(Self::UInt8(Vec::new())),
            DuckLakeType::UInt16 => Ok(Self::UInt16(Vec::new())),
            DuckLakeType::UInt32 => Ok(Self::UInt32(Vec::new())),
            DuckLakeType::UInt64 => Ok(Self::UInt64(Vec::new())),
            DuckLakeType::Int128 | DuckLakeType::UInt128 => Ok(Self::Decimal256 {
                values: Vec::new(),
                precision: 39,
                scale: 0,
            }),
            DuckLakeType::Float32 => Ok(Self::Float32(Vec::new())),
            DuckLakeType::Float64 => Ok(Self::Float64(Vec::new())),
            DuckLakeType::Decimal { width, scale } if *width <= 38 => Ok(Self::Decimal128 {
                values: Vec::new(),
                precision: *width,
                scale: *scale,
            }),
            DuckLakeType::Decimal { width, scale } => Ok(Self::Decimal256 {
                values: Vec::new(),
                precision: *width,
                scale: *scale,
            }),
            DuckLakeType::Date => Ok(Self::Date32(Vec::new())),
            DuckLakeType::Time => Ok(Self::TimeMicros(Vec::new())),
            DuckLakeType::TimeNs => Ok(Self::TimeNanos(Vec::new())),
            DuckLakeType::TimestampS => Ok(Self::TimestampSeconds(Vec::new())),
            DuckLakeType::TimestampMs => Ok(Self::TimestampMillis(Vec::new())),
            DuckLakeType::Timestamp | DuckLakeType::TimestampUs => Ok(Self::TimestampMicros {
                values: Vec::new(),
                timezone: None,
            }),
            DuckLakeType::TimestampTz => Ok(Self::TimestampMicros {
                values: Vec::new(),
                timezone: Some("UTC"),
            }),
            DuckLakeType::TimestampNs => Ok(Self::TimestampNanos(Vec::new())),
            DuckLakeType::TimeTz => match dialect {
                InlinedSqlDialect::Postgres => Ok(Self::Utf8HexBytes(Vec::new())),
                _ => Ok(Self::Utf8(Vec::new())),
            },
            DuckLakeType::Interval => Ok(Self::Interval(Vec::new())),
            DuckLakeType::Varchar | DuckLakeType::Json => match dialect {
                InlinedSqlDialect::Postgres => Ok(Self::Utf8HexBytes(Vec::new())),
                _ => Ok(Self::Utf8(Vec::new())),
            },
            DuckLakeType::Blob | DuckLakeType::Variant | DuckLakeType::Geometry => {
                Ok(Self::Binary(Vec::new()))
            }
            DuckLakeType::Uuid => Ok(Self::Uuid(Vec::new())),
            other => Err(StorageError::Unsupported(format!(
                "Inlined DuckLake type {other} is not supported"
            ))),
        }
    }

    pub(crate) fn into_array(self) -> Result<ArrayRef> {
        let array: ArrayRef = match self {
            Self::Boolean(values) => std::sync::Arc::new(BooleanArray::from(values)),
            Self::Int8(values) => std::sync::Arc::new(Int8Array::from(values)),
            Self::Int16(values) => std::sync::Arc::new(Int16Array::from(values)),
            Self::Int32(values) => std::sync::Arc::new(Int32Array::from(values)),
            Self::Int64(values) => std::sync::Arc::new(Int64Array::from(values)),
            Self::UInt8(values) => std::sync::Arc::new(UInt8Array::from(values)),
            Self::UInt16(values) => std::sync::Arc::new(UInt16Array::from(values)),
            Self::UInt32(values) => std::sync::Arc::new(UInt32Array::from(values)),
            Self::UInt64(values) => std::sync::Arc::new(UInt64Array::from(values)),
            Self::Float32(values) => std::sync::Arc::new(Float32Array::from(values)),
            Self::Float64(values) => std::sync::Arc::new(Float64Array::from(values)),
            Self::Decimal128 {
                values,
                precision,
                scale,
            } => std::sync::Arc::new(
                Decimal128Array::from(values)
                    .with_precision_and_scale(precision, scale as i8)
                    .map_err(|err| StorageError::Backend(err.to_string()))?,
            ),
            Self::Decimal256 {
                values,
                precision,
                scale,
            } => std::sync::Arc::new(
                Decimal256Array::from(values)
                    .with_precision_and_scale(precision, scale as i8)
                    .map_err(|err| StorageError::Backend(err.to_string()))?,
            ),
            Self::Date32(values) => std::sync::Arc::new(Date32Array::from(values)),
            Self::TimeMicros(values) => std::sync::Arc::new(Time64MicrosecondArray::from(values)),
            Self::TimeNanos(values) => std::sync::Arc::new(Time64NanosecondArray::from(values)),
            Self::TimestampSeconds(values) => {
                std::sync::Arc::new(TimestampSecondArray::from(values))
            }
            Self::TimestampMillis(values) => {
                std::sync::Arc::new(TimestampMillisecondArray::from(values))
            }
            Self::TimestampMicros { values, timezone } => {
                let array = TimestampMicrosecondArray::from(values);
                match timezone {
                    Some(timezone) => std::sync::Arc::new(array.with_timezone(timezone)),
                    None => std::sync::Arc::new(array),
                }
            }
            Self::TimestampNanos(values) => {
                std::sync::Arc::new(TimestampNanosecondArray::from(values))
            }
            Self::Interval(values) => std::sync::Arc::new(IntervalMonthDayNanoArray::from(values)),
            Self::Binary(values) => {
                let value_refs = values
                    .iter()
                    .map(|value| value.as_deref())
                    .collect::<Vec<_>>();
                std::sync::Arc::new(BinaryArray::from(value_refs))
            }
            Self::Uuid(values) => std::sync::Arc::new(
                FixedSizeBinaryArray::try_from_sparse_iter_with_size(
                    values.into_iter().map(|value| value.map(Vec::from)),
                    16,
                )
                .map_err(|err| StorageError::Backend(err.to_string()))?,
            ),
            Self::Utf8(values) | Self::Utf8HexBytes(values) => {
                std::sync::Arc::new(StringArray::from(values))
            }
            Self::Nested {
                field,
                values,
                format,
            } => match format {
                InlinedNestedValuesFormat::DuckDbLiteral => {
                    crate::inlined_nested_values_to_array(&field, values)?
                }
                InlinedNestedValuesFormat::Json => {
                    crate::inlined_nested_json_values_to_array(&field, values)?
                }
            },
        };
        Ok(array)
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct InlinedBatchRow {
    pub(crate) batch_index: usize,
    pub(crate) row_index: usize,
    pub(crate) row_id: i64,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct InlinedDataDeletionPlan {
    pub(crate) table_name: String,
    pub(crate) row_ids: Vec<i64>,
    pub(crate) rows_deleted: i64,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct FlushedInlinedDataPlan {
    pub(crate) table_name: String,
    pub(crate) row_ids: Vec<i64>,
    pub(crate) rows_flushed: i64,
    pub(crate) expected_rows: usize,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct InlinedFileDeletionRow {
    pub(crate) data_file_id: i64,
    pub(crate) row_id: i64,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct InlinedFileDeletionPlan {
    pub(crate) table_name: String,
    pub(crate) rows: Vec<InlinedFileDeletionRow>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct FlushedInlinedFileDeletionPlan {
    pub(crate) table_name: String,
    pub(crate) rows: Vec<FlushedInlinedFileDeletion>,
}

pub(crate) fn inlined_data_table_name(table_id: TableId, schema_version: i64) -> String {
    format!("ducklake_inlined_data_{}_{}", table_id.0, schema_version)
}

pub(crate) fn inlined_file_deletion_table_name(table_id: TableId) -> String {
    format!("ducklake_inlined_delete_{}", table_id.0)
}

pub(crate) fn inlined_column_definitions<Q, T>(
    schema: &TableSchema,
    mut quote_identifier: Q,
    mut column_type: T,
) -> Result<String>
where
    Q: FnMut(&str) -> String,
    T: FnMut(&SchemaField) -> Result<String>,
{
    let mut column_defs = Vec::with_capacity(schema.columns.len());
    for field in &schema.columns {
        column_defs.push(format!(
            "{} {}",
            quote_identifier(&field.name),
            column_type(field)?
        ));
    }
    if column_defs.is_empty() {
        Ok(String::new())
    } else {
        Ok(format!(", {}", column_defs.join(", ")))
    }
}

pub(crate) fn create_inlined_data_table_sql(table: &str, column_defs: &str) -> String {
    format!(
        "CREATE TABLE IF NOT EXISTS {table}(\
         row_id BIGINT, begin_snapshot BIGINT, end_snapshot BIGINT{column_defs}\
         )"
    )
}

pub(crate) fn inlined_record_batch_from_arrays(
    schema: &TableSchema,
    arrays: Vec<ArrayRef>,
) -> Result<Option<RecordBatch>> {
    if arrays.is_empty() {
        return Ok(None);
    }
    let batch = RecordBatch::try_new(inlined_arrow_schema(schema)?, arrays)
        .map_err(|err| StorageError::Backend(err.to_string()))?;
    Ok((batch.num_rows() > 0).then_some(batch))
}

pub(crate) fn inlined_data_batch_from_arrays(
    schema: &TableSchema,
    arrays: Vec<ArrayRef>,
    row_ids: Vec<i64>,
    begin_snapshots: Vec<ducklake_catalog::SnapshotId>,
) -> Result<Option<InlinedDataBatch>> {
    Ok(inlined_record_batch_from_arrays(schema, arrays)?
        .map(|batch| InlinedDataBatch::with_metadata(batch, row_ids, begin_snapshots)))
}

pub(crate) fn inlined_column_buffers(
    schema: &TableSchema,
    dialect: InlinedSqlDialect,
) -> Result<Vec<InlinedColumnBuffer>> {
    schema
        .columns
        .iter()
        .map(|field| InlinedColumnBuffer::new(field, dialect))
        .collect()
}

pub(crate) fn inlined_record_batches_from_columns(
    schema: &TableSchema,
    columns: Vec<InlinedColumnBuffer>,
) -> Result<Vec<RecordBatch>> {
    if columns.is_empty() {
        return Ok(Vec::new());
    }
    let arrays = columns
        .into_iter()
        .map(InlinedColumnBuffer::into_array)
        .collect::<Result<Vec<_>>>()?;
    Ok(inlined_record_batch_from_arrays(schema, arrays)?
        .into_iter()
        .collect())
}

pub(crate) fn inlined_data_batches_from_columns(
    schema: &TableSchema,
    columns: Vec<InlinedColumnBuffer>,
    row_ids: Vec<i64>,
    begin_snapshots: Vec<ducklake_catalog::SnapshotId>,
) -> Result<Vec<InlinedDataBatch>> {
    if columns.is_empty() {
        return Ok(Vec::new());
    }
    let arrays = columns
        .into_iter()
        .map(InlinedColumnBuffer::into_array)
        .collect::<Result<Vec<_>>>()?;
    Ok(
        inlined_data_batch_from_arrays(schema, arrays, row_ids, begin_snapshots)?
            .into_iter()
            .collect(),
    )
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn read_inlined_record_batches<Scan>(
    schema: &TableSchema,
    dialect: InlinedSqlDialect,
    scan: Scan,
) -> Result<Vec<RecordBatch>>
where
    Scan: FnOnce(&mut [InlinedColumnBuffer]) -> Result<()>,
{
    let mut columns = inlined_column_buffers(schema, dialect)?;
    if columns.is_empty() {
        return Ok(Vec::new());
    }
    scan(&mut columns)?;
    inlined_record_batches_from_columns(schema, columns)
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn read_inlined_data_batches<Scan>(
    schema: &TableSchema,
    dialect: InlinedSqlDialect,
    scan: Scan,
) -> Result<Vec<InlinedDataBatch>>
where
    Scan: FnOnce(
        &mut [InlinedColumnBuffer],
        &mut Vec<i64>,
        &mut Vec<ducklake_catalog::SnapshotId>,
    ) -> Result<()>,
{
    let mut columns = inlined_column_buffers(schema, dialect)?;
    let mut row_ids = Vec::new();
    let mut begin_snapshots = Vec::new();
    scan(&mut columns, &mut row_ids, &mut begin_snapshots)?;
    inlined_data_batches_from_columns(schema, columns, row_ids, begin_snapshots)
}

pub(crate) async fn read_inlined_record_batches_async<Scan, ScanFuture>(
    schema: &TableSchema,
    dialect: InlinedSqlDialect,
    scan: Scan,
) -> Result<Vec<RecordBatch>>
where
    Scan: FnOnce(Vec<InlinedColumnBuffer>) -> ScanFuture,
    ScanFuture: Future<Output = Result<Vec<InlinedColumnBuffer>>>,
{
    let columns = inlined_column_buffers(schema, dialect)?;
    if columns.is_empty() {
        return Ok(Vec::new());
    }
    let columns = scan(columns).await?;
    inlined_record_batches_from_columns(schema, columns)
}

type InlinedDataBatchScanResult = (
    Vec<InlinedColumnBuffer>,
    Vec<i64>,
    Vec<ducklake_catalog::SnapshotId>,
);

pub(crate) async fn read_inlined_data_batches_async<Scan, ScanFuture>(
    schema: &TableSchema,
    dialect: InlinedSqlDialect,
    scan: Scan,
) -> Result<Vec<InlinedDataBatch>>
where
    Scan:
        FnOnce(Vec<InlinedColumnBuffer>, Vec<i64>, Vec<ducklake_catalog::SnapshotId>) -> ScanFuture,
    ScanFuture: Future<Output = Result<InlinedDataBatchScanResult>>,
{
    let columns = inlined_column_buffers(schema, dialect)?;
    let (columns, row_ids, begin_snapshots) = scan(columns, Vec::new(), Vec::new()).await?;
    inlined_data_batches_from_columns(schema, columns, row_ids, begin_snapshots)
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn inlined_arrow_field_from_schema(field: &SchemaField) -> Result<Field> {
    let data_type = match &field.ducklake_type {
        DuckLakeType::Boolean => DataType::Boolean,
        DuckLakeType::Int8 => DataType::Int8,
        DuckLakeType::Int16 => DataType::Int16,
        DuckLakeType::Int32 => DataType::Int32,
        DuckLakeType::Int64 => DataType::Int64,
        DuckLakeType::UInt8 => DataType::UInt8,
        DuckLakeType::UInt16 => DataType::UInt16,
        DuckLakeType::UInt32 => DataType::UInt32,
        DuckLakeType::UInt64 => DataType::UInt64,
        // i128 spans 39 decimal digits, so int128 needs Decimal256 like uint128.
        DuckLakeType::Int128 => DataType::Decimal256(39, 0),
        DuckLakeType::UInt128 => DataType::Decimal256(39, 0),
        DuckLakeType::Float32 => DataType::Float32,
        DuckLakeType::Float64 => DataType::Float64,
        DuckLakeType::Decimal { width, scale } if *width <= 38 => {
            DataType::Decimal128(*width, *scale as i8)
        }
        DuckLakeType::Decimal { width, scale } => DataType::Decimal256(*width, *scale as i8),
        DuckLakeType::Date => DataType::Date32,
        DuckLakeType::Time => DataType::Time64(TimeUnit::Microsecond),
        DuckLakeType::TimeNs => DataType::Time64(TimeUnit::Nanosecond),
        DuckLakeType::TimestampS => DataType::Timestamp(TimeUnit::Second, None),
        DuckLakeType::TimestampMs => DataType::Timestamp(TimeUnit::Millisecond, None),
        DuckLakeType::Timestamp | DuckLakeType::TimestampUs => {
            DataType::Timestamp(TimeUnit::Microsecond, None)
        }
        DuckLakeType::TimestampTz => DataType::Timestamp(TimeUnit::Microsecond, Some("UTC".into())),
        DuckLakeType::TimestampNs => DataType::Timestamp(TimeUnit::Nanosecond, None),
        DuckLakeType::TimeTz => DataType::Utf8,
        DuckLakeType::Interval => DataType::Interval(IntervalUnit::MonthDayNano),
        DuckLakeType::Varchar | DuckLakeType::Json => DataType::Utf8,
        DuckLakeType::Blob | DuckLakeType::Variant | DuckLakeType::Geometry => DataType::Binary,
        DuckLakeType::Uuid => DataType::FixedSizeBinary(16),
        DuckLakeType::Struct => {
            let children = field
                .children
                .iter()
                .map(inlined_arrow_field_from_schema)
                .collect::<Result<Vec<_>>>()?;
            DataType::Struct(children.into())
        }
        DuckLakeType::List => {
            let child = field.children.first().ok_or_else(|| {
                StorageError::Backend(format!("DuckLake list field {} has no child", field.name))
            })?;
            DataType::List(Arc::new(inlined_arrow_field_from_schema(child)?))
        }
        DuckLakeType::Map => {
            if field.children.len() != 2 {
                return Err(StorageError::Backend(format!(
                    "DuckLake map field {} must have key and value children",
                    field.name
                )));
            }
            let entries = Field::new(
                "entries",
                DataType::Struct(
                    vec![
                        inlined_arrow_field_from_schema(&field.children[0])?,
                        inlined_arrow_field_from_schema(&field.children[1])?,
                    ]
                    .into(),
                ),
                false,
            );
            DataType::Map(Arc::new(entries), false)
        }
        other => {
            return Err(StorageError::Unsupported(format!(
                "Inlined DuckLake type {other} is not supported"
            )));
        }
    };
    Ok(Field::new(
        field.name.clone(),
        data_type,
        field.nulls_allowed,
    ))
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[allow(dead_code)]
pub(crate) fn inlined_nested_values_to_array(
    field: &SchemaField,
    values: Vec<Option<String>>,
) -> Result<ArrayRef> {
    let values = values
        .into_iter()
        .map(|value| {
            value
                .map(|value| duckdb_literal_to_json(field, &value))
                .transpose()
        })
        .collect::<Result<Vec<_>>>()?;
    inlined_nested_json_values_to_array(field, values)
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn inlined_nested_json_values_to_array(
    field: &SchemaField,
    values: Vec<Option<String>>,
) -> Result<ArrayRef> {
    let arrow_field = inlined_arrow_field_from_schema(field)?;
    if values.is_empty() {
        return Ok(new_empty_array(arrow_field.data_type()));
    }

    let mut json = String::new();
    for value in values {
        json.push('{');
        json.push_str(&json_string(&field.name));
        json.push(':');
        match value {
            Some(value) => json.push_str(&value),
            None => json.push_str("null"),
        }
        json.push_str("}\n");
    }

    let schema = Arc::new(Schema::new(vec![arrow_field]));
    let mut reader = ReaderBuilder::new(schema)
        .with_batch_size(json.lines().count().max(1))
        .build(Cursor::new(json.into_bytes()))
        .map_err(|err| StorageError::Backend(err.to_string()))?;
    let Some(batch) = reader.next() else {
        return Err(StorageError::Backend(
            "Arrow JSON reader produced no batch for inlined nested values".into(),
        ));
    };
    let batch = batch.map_err(|err| StorageError::Backend(err.to_string()))?;
    Ok(Arc::clone(batch.column(0)))
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn duckdb_literal_to_json(field: &SchemaField, raw: &str) -> Result<String> {
    let raw = raw.trim();
    if raw.eq_ignore_ascii_case("null") {
        return Ok("null".into());
    }

    match &field.ducklake_type {
        DuckLakeType::Struct => duckdb_struct_literal_to_json(field, raw),
        DuckLakeType::List => duckdb_list_literal_to_json(field, raw),
        DuckLakeType::Map => duckdb_map_literal_to_json(field, raw),
        DuckLakeType::Boolean => {
            let lower = raw.to_ascii_lowercase();
            match lower.as_str() {
                "true" | "false" => Ok(lower),
                _ => Err(StorageError::Backend(format!(
                    "cannot decode DuckDB inlined literal {raw:?} as boolean"
                ))),
            }
        }
        DuckLakeType::Int8
        | DuckLakeType::Int16
        | DuckLakeType::Int32
        | DuckLakeType::Int64
        | DuckLakeType::UInt8
        | DuckLakeType::UInt16
        | DuckLakeType::UInt32
        | DuckLakeType::UInt64
        | DuckLakeType::Float32
        | DuckLakeType::Float64 => Ok(raw.to_string()),
        DuckLakeType::Decimal { .. } | DuckLakeType::Int128 | DuckLakeType::UInt128 => {
            Ok(json_string(&duckdb_unquote(raw)))
        }
        DuckLakeType::Varchar
        | DuckLakeType::Json
        | DuckLakeType::Date
        | DuckLakeType::Time
        | DuckLakeType::TimeNs
        | DuckLakeType::TimeTz
        | DuckLakeType::TimestampS
        | DuckLakeType::TimestampMs
        | DuckLakeType::Timestamp
        | DuckLakeType::TimestampUs
        | DuckLakeType::TimestampNs
        | DuckLakeType::TimestampTz
        | DuckLakeType::Interval => Ok(json_string(&duckdb_unquote(raw))),
        DuckLakeType::Uuid => Ok(json_string(&hex_encode_bytes(&parse_uuid_bytes(
            &duckdb_unquote(raw),
            "DuckDB inlined literal",
        )?))),
        DuckLakeType::Blob | DuckLakeType::Variant | DuckLakeType::Geometry => {
            Ok(json_string(&hex_encode_bytes(&parse_hex_bytes(
                &duckdb_unquote(raw),
                "DuckDB inlined literal",
            )?)))
        }
        DuckLakeType::Unknown => Err(StorageError::Unsupported(format!(
            "Inlined DuckLake type {} is not supported",
            field.type_name
        ))),
    }
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
fn duckdb_struct_literal_to_json(field: &SchemaField, raw: &str) -> Result<String> {
    let inner = delimited_inner(raw, '{', '}', field)?;
    if inner.trim().is_empty() {
        return Ok("{}".into());
    }

    let mut json = String::from("{");
    for (idx, part) in split_top_level(inner, ',')?.iter().enumerate() {
        let (name, value) = split_top_level_once(part, ':')?;
        let name = duckdb_unquote(name);
        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
                ))
            })?;
        if idx > 0 {
            json.push(',');
        }
        json.push_str(&json_string(&name));
        json.push(':');
        json.push_str(&duckdb_literal_to_json(child, value)?);
    }
    json.push('}');
    Ok(json)
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
fn duckdb_list_literal_to_json(field: &SchemaField, raw: &str) -> Result<String> {
    let child = field.children.first().ok_or_else(|| {
        StorageError::Backend(format!("DuckLake list field {} has no child", field.name))
    })?;
    let inner = delimited_inner(raw, '[', ']', field)?;
    if inner.trim().is_empty() {
        return Ok("[]".into());
    }

    let mut json = String::from("[");
    for (idx, part) in split_top_level(inner, ',')?.iter().enumerate() {
        if idx > 0 {
            json.push(',');
        }
        json.push_str(&duckdb_literal_to_json(child, part)?);
    }
    json.push(']');
    Ok(json)
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
fn duckdb_map_literal_to_json(field: &SchemaField, raw: &str) -> Result<String> {
    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 inner = delimited_inner(raw, '{', '}', field)?;
    if inner.trim().is_empty() {
        return Ok("{}".into());
    }

    let mut json = String::from("{");
    for (idx, part) in split_top_level(inner, ',')?.iter().enumerate() {
        let (key, value) = split_top_level_once(part, '=')?;
        if idx > 0 {
            json.push(',');
        }
        json.push_str(&duckdb_map_key_to_json(key_field, key));
        json.push(':');
        json.push_str(&duckdb_literal_to_json(value_field, value)?);
    }
    json.push('}');
    Ok(json)
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
fn duckdb_map_key_to_json(field: &SchemaField, raw: &str) -> String {
    match field.ducklake_type {
        DuckLakeType::Varchar | DuckLakeType::Json | DuckLakeType::Uuid => {
            json_string(&duckdb_unquote(raw))
        }
        _ => json_string(raw.trim()),
    }
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
fn delimited_inner<'a>(
    raw: &'a str,
    open: char,
    close: char,
    field: &SchemaField,
) -> Result<&'a str> {
    let raw = raw.trim();
    if raw.starts_with(open) && raw.ends_with(close) {
        Ok(&raw[open.len_utf8()..raw.len() - close.len_utf8()])
    } else {
        Err(StorageError::Backend(format!(
            "cannot decode DuckDB inlined literal {raw:?} as {}",
            field.ducklake_type
        )))
    }
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
fn split_top_level(input: &str, delimiter: char) -> Result<Vec<&str>> {
    let mut parts = Vec::new();
    let mut start = 0;
    let mut depth = 0_i32;
    let mut in_string = false;
    let mut chars = input.char_indices().peekable();

    while let Some((idx, ch)) = chars.next() {
        if ch == '\'' {
            if in_string && matches!(chars.peek(), Some((_, '\''))) {
                chars.next();
                continue;
            }
            in_string = !in_string;
            continue;
        }
        if in_string {
            continue;
        }

        match ch {
            '{' | '[' | '(' => depth += 1,
            '}' | ']' | ')' => depth -= 1,
            _ if ch == delimiter && depth == 0 => {
                parts.push(input[start..idx].trim());
                start = idx + ch.len_utf8();
            }
            _ => {}
        }
        if depth < 0 {
            return Err(StorageError::Backend(format!(
                "unbalanced DuckDB inlined literal {input:?}"
            )));
        }
    }

    if in_string || depth != 0 {
        return Err(StorageError::Backend(format!(
            "unbalanced DuckDB inlined literal {input:?}"
        )));
    }
    parts.push(input[start..].trim());
    Ok(parts)
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
fn split_top_level_once(input: &str, delimiter: char) -> Result<(&str, &str)> {
    let parts = split_top_level(input, delimiter)?;
    if parts.len() == 2 {
        Ok((parts[0], parts[1]))
    } else {
        Err(StorageError::Backend(format!(
            "cannot split DuckDB inlined literal {input:?} on {delimiter:?}"
        )))
    }
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
fn duckdb_unquote(raw: &str) -> String {
    let raw = raw.trim();
    if raw.len() >= 2 && raw.starts_with('\'') && raw.ends_with('\'') {
        raw[1..raw.len() - 1].replace("''", "'")
    } else {
        raw.to_string()
    }
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn parse_uuid_bytes(raw: &str, source: &str) -> Result<[u8; 16]> {
    let compact = raw.trim().replace('-', "");
    if compact.len() != 32 {
        return Err(StorageError::Backend(format!(
            "cannot decode {source} text {raw:?} as UUID"
        )));
    }

    let mut out = [0_u8; 16];
    for (idx, byte) in out.iter_mut().enumerate() {
        let start = idx * 2;
        *byte = u8::from_str_radix(&compact[start..start + 2], 16).map_err(|_| {
            StorageError::Backend(format!("cannot decode {source} text {raw:?} as UUID"))
        })?;
    }
    Ok(out)
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn format_uuid_bytes(bytes: &[u8], source: &str) -> Result<String> {
    if bytes.len() != 16 {
        return Err(StorageError::Backend(format!(
            "cannot encode {source} bytes of length {} as UUID",
            bytes.len()
        )));
    }

    use std::fmt::Write as _;
    let mut out = String::with_capacity(36);
    for (idx, byte) in bytes.iter().enumerate() {
        if matches!(idx, 4 | 6 | 8 | 10) {
            out.push('-');
        }
        let _ = write!(out, "{byte:02x}");
    }
    Ok(out)
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn parse_hex_bytes(raw: &str, source: &str) -> Result<Vec<u8>> {
    let value = normalize_hex_literal(raw);
    if !value.len().is_multiple_of(2) {
        return Err(StorageError::Backend(format!(
            "cannot decode {source} text {raw:?} as hex bytes"
        )));
    }

    let mut out = Vec::with_capacity(value.len() / 2);
    for idx in (0..value.len()).step_by(2) {
        out.push(u8::from_str_radix(&value[idx..idx + 2], 16).map_err(|_| {
            StorageError::Backend(format!("cannot decode {source} text {raw:?} as hex bytes"))
        })?);
    }
    Ok(out)
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
fn normalize_hex_literal(raw: &str) -> String {
    let value = raw.trim();
    let value =
        if value.len() >= 3 && value[..2].eq_ignore_ascii_case("x'") && value.ends_with('\'') {
            value[2..value.len() - 1].replace("''", "'")
        } else {
            duckdb_unquote(value)
        };
    value
        .strip_prefix("\\x")
        .or_else(|| value.strip_prefix("\\X"))
        .or_else(|| value.strip_prefix("0x"))
        .or_else(|| value.strip_prefix("0X"))
        .unwrap_or(&value)
        .to_string()
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn hex_encode_bytes(bytes: &[u8]) -> String {
    const HEX: &[u8; 16] = b"0123456789abcdef";
    let mut out = String::with_capacity(bytes.len() * 2);
    for byte in bytes {
        out.push(HEX[(byte >> 4) as usize] as char);
        out.push(HEX[(byte & 0x0f) as usize] as char);
    }
    out
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn format_decimal_i128(unscaled: i128, scale: u8) -> String {
    format_decimal_digits(&unscaled.to_string(), scale)
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn format_decimal_i256(unscaled: i256, scale: u8) -> String {
    format_decimal_digits(&unscaled.to_string(), scale)
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
fn format_decimal_digits(unscaled: &str, scale: u8) -> String {
    if scale == 0 {
        return unscaled.to_string();
    }

    let (sign, digits) = unscaled
        .strip_prefix('-')
        .map(|digits| ("-", digits))
        .unwrap_or(("", unscaled));
    let scale = usize::from(scale);
    if digits.len() <= scale {
        let mut out = String::with_capacity(sign.len() + 2 + scale);
        out.push_str(sign);
        out.push_str("0.");
        for _ in 0..(scale - digits.len()) {
            out.push('0');
        }
        out.push_str(digits);
        return out;
    }

    let split_at = digits.len() - scale;
    format!("{sign}{}.{}", &digits[..split_at], &digits[split_at..])
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn format_date32(days: i32, source: &str) -> Result<String> {
    match days {
        i32::MAX => return Ok("infinity".into()),
        value if value == -i32::MAX => return Ok("-infinity".into()),
        _ => {}
    }
    let epoch = NaiveDate::from_ymd_opt(1970, 1, 1).expect("valid epoch date");
    let date = epoch
        .checked_add_signed(Duration::days(i64::from(days)))
        .ok_or_else(|| StorageError::Backend(format!("{source} date {days} is out of range")))?;
    Ok(date.format("%Y-%m-%d").to_string())
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn format_time64(value: i64, unit: TimeUnit, source: &str) -> Result<String> {
    let total_nanos = i128::from(value)
        .checked_mul(arrow_time_unit_nanos(unit))
        .ok_or_else(|| StorageError::Backend(format!("{source} time {value} is out of range")))?;
    let nanos_per_day = 86_400_i128 * 1_000_000_000;
    if total_nanos < 0 || total_nanos >= nanos_per_day {
        return Err(StorageError::Backend(format!(
            "{source} time {value} is out of range"
        )));
    }
    let seconds = u32::try_from(total_nanos / 1_000_000_000)
        .map_err(|_| StorageError::Backend(format!("{source} time {value} is out of range")))?;
    let nanos = u32::try_from(total_nanos % 1_000_000_000)
        .map_err(|_| StorageError::Backend(format!("{source} time {value} is out of range")))?;
    let time = NaiveTime::from_num_seconds_from_midnight_opt(seconds, nanos)
        .ok_or_else(|| StorageError::Backend(format!("{source} time {value} is out of range")))?;
    Ok(format!(
        "{}{}",
        time.format("%H:%M:%S"),
        format_subsecond(nanos, unit)
    ))
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn format_timestamp(
    value: i64,
    unit: TimeUnit,
    utc_suffix: bool,
    source: &str,
) -> Result<String> {
    match value {
        i64::MAX => return Ok("infinity".into()),
        value if value == -i64::MAX => return Ok("-infinity".into()),
        _ => {}
    }

    let total_nanos = i128::from(value)
        .checked_mul(arrow_time_unit_nanos(unit))
        .ok_or_else(|| {
            StorageError::Backend(format!("{source} timestamp {value} is out of range"))
        })?;
    let seconds = i64::try_from(total_nanos.div_euclid(1_000_000_000)).map_err(|_| {
        StorageError::Backend(format!("{source} timestamp {value} is out of range"))
    })?;
    let nanos = u32::try_from(total_nanos.rem_euclid(1_000_000_000)).map_err(|_| {
        StorageError::Backend(format!("{source} timestamp {value} is out of range"))
    })?;
    let timestamp = DateTime::<Utc>::from_timestamp(seconds, nanos).ok_or_else(|| {
        StorageError::Backend(format!("{source} timestamp {value} is out of range"))
    })?;
    let suffix = if utc_suffix { "+00" } else { "" };
    Ok(format!(
        "{}{}{}",
        timestamp.format("%Y-%m-%d %H:%M:%S"),
        format_subsecond(nanos, unit),
        suffix
    ))
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn format_interval_month_day_nano(value: IntervalMonthDayNano) -> String {
    if value.nanoseconds % 1_000 == 0 {
        format!(
            "{} months {} days {} microseconds",
            value.months,
            value.days,
            value.nanoseconds / 1_000
        )
    } else {
        format!(
            "{} months {} days {} nanoseconds",
            value.months, value.days, value.nanoseconds
        )
    }
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn format_microsecond_interval_month_day_nano(
    value: IntervalMonthDayNano,
    source: &str,
) -> Result<String> {
    if value.nanoseconds % 1_000 != 0 {
        return Err(StorageError::Unsupported(format!(
            "{source} inlined interval values require microsecond precision"
        )));
    }
    Ok(format_interval_month_day_nano(value))
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
fn format_subsecond(nanos: u32, unit: TimeUnit) -> String {
    let digits = match unit {
        TimeUnit::Second => return String::new(),
        TimeUnit::Millisecond => 3,
        TimeUnit::Microsecond => 6,
        TimeUnit::Nanosecond => 9,
    };
    let divisor = 10_u32.pow(9 - digits);
    format!(".{:0width$}", nanos / divisor, width = digits as usize)
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn arrow_time_unit_nanos(unit: TimeUnit) -> i128 {
    match unit {
        TimeUnit::Second => 1_000_000_000,
        TimeUnit::Millisecond => 1_000_000,
        TimeUnit::Microsecond => 1_000,
        TimeUnit::Nanosecond => 1,
    }
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn json_string(value: &str) -> String {
    let mut out = String::with_capacity(value.len() + 2);
    out.push('"');
    for ch in value.chars() {
        match ch {
            '"' => out.push_str("\\\""),
            '\\' => out.push_str("\\\\"),
            '\n' => out.push_str("\\n"),
            '\r' => out.push_str("\\r"),
            '\t' => out.push_str("\\t"),
            '\u{08}' => out.push_str("\\b"),
            '\u{0C}' => out.push_str("\\f"),
            ch if ch.is_control() => out.push_str(&format!("\\u{:04x}", ch as u32)),
            ch => out.push(ch),
        }
    }
    out.push('"');
    out
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn parse_interval_month_day_nano(
    raw: &str,
    source: &str,
) -> Result<arrow::datatypes::IntervalMonthDayNano> {
    let mut months = 0_i64;
    let mut days = 0_i64;
    let mut nanos = 0_i128;
    let tokens = raw.split_whitespace().collect::<Vec<_>>();
    let mut idx = 0;

    while idx < tokens.len() {
        let token = tokens[idx].trim_matches(',');
        if token.contains(':') {
            nanos = checked_i128_add(
                nanos,
                parse_interval_clock_nanos(token, raw, source)?,
                raw,
                source,
            )?;
            idx += 1;
            continue;
        }

        let amount = token.parse::<i64>().map_err(|_| {
            StorageError::Backend(format!("cannot decode {source} text {raw:?} as interval"))
        })?;
        let Some(unit) = tokens.get(idx + 1) else {
            return Err(StorageError::Backend(format!(
                "cannot decode {source} text {raw:?} as interval"
            )));
        };
        match unit.trim_matches(',').to_ascii_lowercase().as_str() {
            "year" | "years" | "yr" | "yrs" => {
                months = checked_i64_add(
                    months,
                    amount
                        .checked_mul(12)
                        .ok_or_else(|| interval_out_of_range(raw, source))?,
                    raw,
                    source,
                )?;
            }
            "month" | "months" | "mon" | "mons" => {
                months = checked_i64_add(months, amount, raw, source)?;
            }
            "week" | "weeks" => {
                days = checked_i64_add(
                    days,
                    amount
                        .checked_mul(7)
                        .ok_or_else(|| interval_out_of_range(raw, source))?,
                    raw,
                    source,
                )?;
            }
            "day" | "days" => days = checked_i64_add(days, amount, raw, source)?,
            "hour" | "hours" | "hr" | "hrs" => {
                nanos = checked_i128_add(
                    nanos,
                    interval_unit_nanos(amount, 3_600_000_000_000, raw, source)?,
                    raw,
                    source,
                )?;
            }
            "minute" | "minutes" | "min" | "mins" => {
                nanos = checked_i128_add(
                    nanos,
                    interval_unit_nanos(amount, 60_000_000_000, raw, source)?,
                    raw,
                    source,
                )?;
            }
            "second" | "seconds" | "sec" | "secs" => {
                nanos = checked_i128_add(
                    nanos,
                    interval_unit_nanos(amount, 1_000_000_000, raw, source)?,
                    raw,
                    source,
                )?;
            }
            "millisecond" | "milliseconds" | "msec" | "msecs" | "ms" => {
                nanos = checked_i128_add(
                    nanos,
                    interval_unit_nanos(amount, 1_000_000, raw, source)?,
                    raw,
                    source,
                )?;
            }
            "microsecond" | "microseconds" | "usec" | "usecs" | "us" => {
                nanos = checked_i128_add(
                    nanos,
                    interval_unit_nanos(amount, 1_000, raw, source)?,
                    raw,
                    source,
                )?;
            }
            "nanosecond" | "nanoseconds" | "nsec" | "nsecs" | "ns" => {
                nanos = checked_i128_add(nanos, i128::from(amount), raw, source)?;
            }
            _ => {
                return Err(StorageError::Backend(format!(
                    "cannot decode {source} text {raw:?} as interval"
                )));
            }
        }
        idx += 2;
    }

    let months = i32::try_from(months).map_err(|_| interval_out_of_range(raw, source))?;
    let days = i32::try_from(days).map_err(|_| interval_out_of_range(raw, source))?;
    let nanoseconds = i64::try_from(nanos).map_err(|_| interval_out_of_range(raw, source))?;
    Ok(arrow::datatypes::IntervalMonthDayNano::new(
        months,
        days,
        nanoseconds,
    ))
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
fn parse_interval_clock_nanos(raw_token: &str, raw: &str, source: &str) -> Result<i128> {
    let negative = raw_token.starts_with('-');
    let token = raw_token
        .strip_prefix('-')
        .or_else(|| raw_token.strip_prefix('+'))
        .unwrap_or(raw_token);
    let parts = token.split(':').collect::<Vec<_>>();
    if parts.len() != 3 {
        return Err(StorageError::Backend(format!(
            "cannot decode {source} text {raw:?} as interval"
        )));
    }

    let hours = parts[0].parse::<i128>().map_err(|_| {
        StorageError::Backend(format!("cannot decode {source} text {raw:?} as interval"))
    })?;
    let minutes = parts[1].parse::<i128>().map_err(|_| {
        StorageError::Backend(format!("cannot decode {source} text {raw:?} as interval"))
    })?;
    let (seconds, nanos) = parse_interval_seconds_nanos(parts[2], raw, source)?;
    let total = hours
        .checked_mul(3_600_000_000_000)
        .and_then(|value| value.checked_add(minutes.checked_mul(60_000_000_000)?))
        .and_then(|value| value.checked_add(seconds.checked_mul(1_000_000_000)?))
        .and_then(|value| value.checked_add(nanos))
        .ok_or_else(|| interval_out_of_range(raw, source))?;
    Ok(if negative { -total } else { total })
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
fn parse_interval_seconds_nanos(raw: &str, interval: &str, source: &str) -> Result<(i128, i128)> {
    let mut parts = raw.split('.');
    let seconds = parts.next().unwrap_or("").parse::<i128>().map_err(|_| {
        StorageError::Backend(format!(
            "cannot decode {source} text {interval:?} as interval"
        ))
    })?;
    let fraction = parts.next().unwrap_or("");
    if parts.next().is_some() || fraction.len() > 9 || !fraction.bytes().all(|b| b.is_ascii_digit())
    {
        return Err(StorageError::Backend(format!(
            "cannot decode {source} text {interval:?} as interval"
        )));
    }
    let nanos = if fraction.is_empty() {
        0
    } else {
        format!("{fraction:0<9}").parse::<i128>().map_err(|_| {
            StorageError::Backend(format!(
                "cannot decode {source} text {interval:?} as interval"
            ))
        })?
    };
    Ok((seconds, nanos))
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
fn checked_i64_add(lhs: i64, rhs: i64, raw: &str, source: &str) -> Result<i64> {
    lhs.checked_add(rhs)
        .ok_or_else(|| interval_out_of_range(raw, source))
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
fn checked_i128_add(lhs: i128, rhs: i128, raw: &str, source: &str) -> Result<i128> {
    lhs.checked_add(rhs)
        .ok_or_else(|| interval_out_of_range(raw, source))
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
fn interval_unit_nanos(amount: i64, nanos_per_unit: i128, raw: &str, source: &str) -> Result<i128> {
    i128::from(amount)
        .checked_mul(nanos_per_unit)
        .ok_or_else(|| interval_out_of_range(raw, source))
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
fn interval_out_of_range(raw: &str, source: &str) -> StorageError {
    StorageError::Backend(format!("{source} interval {raw:?} is out of range"))
}

fn inlined_arrow_schema(schema: &TableSchema) -> Result<std::sync::Arc<Schema>> {
    let fields = schema
        .columns
        .iter()
        .map(crate::inlined_arrow_field_from_schema)
        .collect::<Result<Vec<_>>>()?;
    Ok(std::sync::Arc::new(Schema::new(fields)))
}

pub(crate) fn inlined_column_type(
    field: &SchemaField,
    dialect: InlinedSqlDialect,
) -> Result<String> {
    if matches!(
        field.ducklake_type,
        DuckLakeType::Struct | DuckLakeType::List | DuckLakeType::Map
    ) {
        return nested_inlined_column_type(field, dialect);
    }
    match dialect {
        InlinedSqlDialect::Sqlite => sqlite_inlined_column_type(field),
        InlinedSqlDialect::DuckDbFile => duckdb_inlined_column_type(field, "DuckDB-file"),
        InlinedSqlDialect::Postgres => postgres_inlined_column_type(field),
        InlinedSqlDialect::Quack => duckdb_inlined_column_type(field, "Quack"),
    }
}

fn nested_inlined_column_type(field: &SchemaField, dialect: InlinedSqlDialect) -> Result<String> {
    match dialect {
        InlinedSqlDialect::Sqlite | InlinedSqlDialect::Postgres => Ok("VARCHAR".into()),
        InlinedSqlDialect::DuckDbFile | InlinedSqlDialect::Quack => {
            duckdb_nested_inlined_column_type(field, dialect)
        }
    }
}

fn duckdb_nested_inlined_column_type(
    field: &SchemaField,
    dialect: InlinedSqlDialect,
) -> Result<String> {
    match field.ducklake_type {
        DuckLakeType::Struct => {
            let mut children = Vec::with_capacity(field.children.len());
            for child in &field.children {
                children.push(format!(
                    "{} {}",
                    crate::ducklake_quote_identifier(&child.name),
                    inlined_column_type(child, dialect)?
                ));
            }
            Ok(format!("STRUCT({})", children.join(", ")))
        }
        DuckLakeType::List => {
            let child = field.children.first().ok_or_else(|| {
                StorageError::Backend(format!("DuckLake list field {} has no child", field.name))
            })?;
            Ok(format!("{}[]", inlined_column_type(child, dialect)?))
        }
        DuckLakeType::Map => {
            if field.children.len() != 2 {
                return Err(StorageError::Backend(format!(
                    "DuckLake map field {} must have key and value children",
                    field.name
                )));
            }
            Ok(format!(
                "MAP({}, {})",
                inlined_column_type(&field.children[0], dialect)?,
                inlined_column_type(&field.children[1], dialect)?
            ))
        }
        ref other => Err(StorageError::Unsupported(format!(
            "Inlined append writes do not support nested DuckLake type {other}"
        ))),
    }
}

#[cfg(any(feature = "duckdb", test))]
pub(crate) fn duckdb_inlined_value_needs_cast(field: &SchemaField) -> bool {
    matches!(
        field.ducklake_type,
        DuckLakeType::Struct
            | DuckLakeType::List
            | DuckLakeType::Map
            | DuckLakeType::UInt64
            | DuckLakeType::Int128
            | DuckLakeType::UInt128
            | DuckLakeType::Decimal { .. }
            | DuckLakeType::Date
            | DuckLakeType::Time
            | DuckLakeType::TimeNs
            | DuckLakeType::Timestamp
            | DuckLakeType::TimestampS
            | DuckLakeType::TimestampMs
            | DuckLakeType::TimestampUs
            | DuckLakeType::TimestampNs
            | DuckLakeType::TimestampTz
            | DuckLakeType::TimeTz
            | DuckLakeType::Interval
            | DuckLakeType::Uuid
    )
}

pub(crate) fn postgres_inlined_placeholder_cast(field: &SchemaField) -> Option<String> {
    match &field.ducklake_type {
        DuckLakeType::Decimal { width, scale } => Some(format!("DECIMAL({width},{scale})")),
        DuckLakeType::Time => Some("TIME".into()),
        DuckLakeType::TimeTz => Some("TIMETZ".into()),
        DuckLakeType::Interval => Some("INTERVAL".into()),
        DuckLakeType::Uuid => Some("UUID".into()),
        _ => None,
    }
}

fn sqlite_inlined_column_type(field: &SchemaField) -> Result<String> {
    match &field.ducklake_type {
        DuckLakeType::Boolean => Ok("BOOLEAN".into()),
        DuckLakeType::Int8 => Ok("TINYINT".into()),
        DuckLakeType::Int16 => Ok("SMALLINT".into()),
        DuckLakeType::Int32 => Ok("INTEGER".into()),
        DuckLakeType::Int64 => Ok("BIGINT".into()),
        DuckLakeType::UInt8 => Ok("UTINYINT".into()),
        DuckLakeType::UInt16 => Ok("USMALLINT".into()),
        DuckLakeType::UInt32 => Ok("UINTEGER".into()),
        DuckLakeType::UInt64
        | DuckLakeType::Int128
        | DuckLakeType::UInt128
        | DuckLakeType::Decimal { .. }
        | DuckLakeType::Date
        | DuckLakeType::Time
        | DuckLakeType::TimeNs
        | DuckLakeType::Timestamp
        | DuckLakeType::TimestampS
        | DuckLakeType::TimestampMs
        | DuckLakeType::TimestampUs
        | DuckLakeType::TimestampNs
        | DuckLakeType::TimestampTz
        | DuckLakeType::TimeTz
        | DuckLakeType::Interval
        | DuckLakeType::Float32
        | DuckLakeType::Float64 => Ok("VARCHAR".into()),
        DuckLakeType::Blob => Ok("BLOB".into()),
        DuckLakeType::Varchar | DuckLakeType::Json => Ok("VARCHAR".into()),
        DuckLakeType::Uuid => Ok("VARCHAR".into()),
        other => unsupported_inlined_column_type(other),
    }
}

fn duckdb_inlined_column_type(field: &SchemaField, backend: &str) -> Result<String> {
    match &field.ducklake_type {
        DuckLakeType::Boolean => Ok("BOOLEAN".into()),
        DuckLakeType::Int8 => Ok("TINYINT".into()),
        DuckLakeType::Int16 => Ok("SMALLINT".into()),
        DuckLakeType::Int32 => Ok("INTEGER".into()),
        DuckLakeType::Int64 => Ok("BIGINT".into()),
        DuckLakeType::UInt8 => Ok("UTINYINT".into()),
        DuckLakeType::UInt16 => Ok("USMALLINT".into()),
        DuckLakeType::UInt32 => Ok("UINTEGER".into()),
        DuckLakeType::UInt64 => Ok("UBIGINT".into()),
        DuckLakeType::Int128 => Ok("HUGEINT".into()),
        DuckLakeType::UInt128 => Ok("UHUGEINT".into()),
        DuckLakeType::Float32 => Ok("FLOAT".into()),
        DuckLakeType::Float64 => Ok("DOUBLE".into()),
        DuckLakeType::Decimal { width, scale } if *width <= 38 => {
            Ok(format!("DECIMAL({width},{scale})"))
        }
        DuckLakeType::Decimal { width, .. } => Err(StorageError::Unsupported(format!(
            "{backend} inlined DECIMAL width {width} exceeds DuckDB's limit"
        ))),
        DuckLakeType::Date => Ok("DATE".into()),
        DuckLakeType::Time => Ok("TIME".into()),
        DuckLakeType::TimeNs => Ok("TIME_NS".into()),
        DuckLakeType::Timestamp | DuckLakeType::TimestampUs => Ok("TIMESTAMP".into()),
        DuckLakeType::TimestampS => Ok("TIMESTAMP_S".into()),
        DuckLakeType::TimestampMs => Ok("TIMESTAMP_MS".into()),
        DuckLakeType::TimestampNs => Ok("TIMESTAMP_NS".into()),
        DuckLakeType::TimestampTz if matches!(backend, "Quack") => Ok("TIMESTAMP_TZ".into()),
        DuckLakeType::TimestampTz => Ok("TIMESTAMPTZ".into()),
        DuckLakeType::TimeTz if matches!(backend, "Quack") => Ok("TIME_TZ".into()),
        DuckLakeType::TimeTz => Ok("TIMETZ".into()),
        DuckLakeType::Interval => Ok("INTERVAL".into()),
        DuckLakeType::Blob => Ok("BLOB".into()),
        DuckLakeType::Varchar | DuckLakeType::Json => Ok("VARCHAR".into()),
        DuckLakeType::Uuid => Ok("UUID".into()),
        other => unsupported_inlined_column_type(other),
    }
}

fn postgres_inlined_column_type(field: &SchemaField) -> Result<String> {
    match &field.ducklake_type {
        DuckLakeType::Boolean => Ok("BOOLEAN".into()),
        DuckLakeType::Int8 => Ok("SMALLINT".into()),
        DuckLakeType::Int16 => Ok("SMALLINT".into()),
        DuckLakeType::Int32 => Ok("INTEGER".into()),
        DuckLakeType::Int64 => Ok("BIGINT".into()),
        DuckLakeType::UInt8 => Ok("INTEGER".into()),
        DuckLakeType::UInt16 => Ok("INTEGER".into()),
        DuckLakeType::UInt32 => Ok("BIGINT".into()),
        DuckLakeType::UInt64
        | DuckLakeType::Int128
        | DuckLakeType::UInt128
        | DuckLakeType::Date
        | DuckLakeType::TimeNs
        | DuckLakeType::Timestamp
        | DuckLakeType::TimestampS
        | DuckLakeType::TimestampMs
        | DuckLakeType::TimestampUs
        | DuckLakeType::TimestampNs
        | DuckLakeType::TimestampTz => Ok("VARCHAR".into()),
        DuckLakeType::Float32 => Ok("REAL".into()),
        DuckLakeType::Float64 => Ok("DOUBLE PRECISION".into()),
        DuckLakeType::Decimal { width, scale } => Ok(format!("DECIMAL({width},{scale})")),
        DuckLakeType::Time => Ok("TIME".into()),
        DuckLakeType::TimeTz => Ok("TIMETZ".into()),
        DuckLakeType::Interval => Ok("INTERVAL".into()),
        DuckLakeType::Blob => Ok("BYTEA".into()),
        DuckLakeType::Varchar | DuckLakeType::Json => Ok("BYTEA".into()),
        DuckLakeType::Uuid => Ok("UUID".into()),
        other => unsupported_inlined_column_type(other),
    }
}

fn unsupported_inlined_column_type(ty: &DuckLakeType) -> Result<String> {
    Err(StorageError::Unsupported(format!(
        "Inlined append writes do not support DuckLake type {ty}"
    )))
}

pub(crate) fn validate_inlined_write_schema<T>(
    schema: &TableSchema,
    mut validate_type: T,
) -> Result<()>
where
    T: FnMut(&SchemaField) -> Result<()>,
{
    for field in &schema.columns {
        validate_inlined_write_field(field, &mut validate_type)?;
    }
    Ok(())
}

fn validate_inlined_write_field<T>(field: &SchemaField, validate_type: &mut T) -> Result<()>
where
    T: FnMut(&SchemaField) -> Result<()>,
{
    if is_inlined_system_column_name(&field.name) {
        return Err(StorageError::Unsupported(format!(
            "Inlined append writes cannot use reserved DuckLake inlined-data column name {}",
            field.name
        )));
    }
    match field.ducklake_type {
        DuckLakeType::Geometry => Err(StorageError::Unsupported(
            "Inlined append writes do not support DuckLake GEOMETRY columns".into(),
        )),
        DuckLakeType::Variant => Err(StorageError::Unsupported(
            "Inlined append writes do not support DuckLake VARIANT columns".into(),
        )),
        DuckLakeType::Struct | DuckLakeType::List | DuckLakeType::Map => {
            validate_nested_field_shape(field)?;
            validate_type(field)?;
            for child in &field.children {
                validate_inlined_write_field(child, validate_type)?;
            }
            Ok(())
        }
        _ => validate_type(field),
    }
}

fn validate_nested_field_shape(field: &SchemaField) -> Result<()> {
    match field.ducklake_type {
        DuckLakeType::Struct => Ok(()),
        DuckLakeType::List if field.children.len() == 1 => Ok(()),
        DuckLakeType::Map if field.children.len() == 2 => Ok(()),
        DuckLakeType::List => Err(StorageError::Backend(format!(
            "DuckLake list field {} must have exactly one child",
            field.name
        ))),
        DuckLakeType::Map => Err(StorageError::Backend(format!(
            "DuckLake map field {} must have key and value children",
            field.name
        ))),
        ref other => Err(StorageError::Backend(format!(
            "DuckLake field {} is not nested: {other}",
            field.name
        ))),
    }
}

fn is_inlined_system_column_name(name: &str) -> bool {
    matches!(
        name.to_ascii_lowercase().as_str(),
        "row_id" | "begin_snapshot" | "end_snapshot"
    )
}

pub(crate) fn validate_inlined_append_for_dialect(
    rows_appended: i64,
    schema: &TableSchema,
    batches: &[RecordBatch],
    preserved_row_ids: Option<&[i64]>,
    context: &str,
    dialect: InlinedSqlDialect,
) -> Result<()> {
    crate::validate_inlined_append_rows(rows_appended, batches, preserved_row_ids, context)?;
    validate_inlined_write_schema(schema, |field| {
        inlined_column_type(field, dialect).map(|_| ())
    })
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn ensure_inlined_data_table_sync<Lookup, Register, Create, Quote, TableIdentifier>(
    table_id: TableId,
    schema_version: i64,
    schema: &TableSchema,
    dialect: InlinedSqlDialect,
    mut quote_identifier: Quote,
    mut table_identifier: TableIdentifier,
    mut lookup_existing: Lookup,
    mut register_table: Register,
    mut create_table: Create,
) -> Result<String>
where
    Lookup: FnMut(TableId, i64) -> Result<Option<String>>,
    Register: FnMut(TableId, &str, i64) -> Result<()>,
    Create: FnMut(&str) -> Result<()>,
    Quote: FnMut(&str) -> String,
    TableIdentifier: FnMut(&str) -> String,
{
    if let Some(table_name) = lookup_existing(table_id, schema_version)? {
        return Ok(table_name);
    }

    let table_name = inlined_data_table_name(table_id, schema_version);
    register_table(table_id, &table_name, schema_version)?;

    let column_defs = inlined_column_definitions(schema, &mut quote_identifier, |field| {
        inlined_column_type(field, dialect)
    })?;
    let table = table_identifier(&table_name);
    create_table(&create_inlined_data_table_sql(&table, &column_defs))?;
    Ok(table_name)
}

pub(crate) async fn ensure_inlined_data_table_async<
    Lookup,
    LookupFut,
    Register,
    RegisterFut,
    Create,
    CreateFut,
    Quote,
    TableIdentifier,
>(
    table_id: TableId,
    schema_version: i64,
    schema: &TableSchema,
    dialect: InlinedSqlDialect,
    mut quote_identifier: Quote,
    mut table_identifier: TableIdentifier,
    mut lookup_existing: Lookup,
    mut register_table: Register,
    mut create_table: Create,
) -> Result<String>
where
    Lookup: FnMut(TableId, i64) -> LookupFut,
    LookupFut: Future<Output = Result<Option<String>>>,
    Register: FnMut(TableId, String, i64) -> RegisterFut,
    RegisterFut: Future<Output = Result<()>>,
    Create: FnMut(String) -> CreateFut,
    CreateFut: Future<Output = Result<()>>,
    Quote: FnMut(&str) -> String,
    TableIdentifier: FnMut(&str) -> String,
{
    if let Some(table_name) = lookup_existing(table_id, schema_version).await? {
        return Ok(table_name);
    }

    let table_name = inlined_data_table_name(table_id, schema_version);
    register_table(table_id, table_name.clone(), schema_version).await?;

    let column_defs = inlined_column_definitions(schema, &mut quote_identifier, |field| {
        inlined_column_type(field, dialect)
    })?;
    let table = table_identifier(&table_name);
    create_table(create_inlined_data_table_sql(&table, &column_defs)).await?;
    Ok(table_name)
}

pub(crate) fn inlined_insert_placeholders<Prefix, FieldPlaceholder>(
    schema: &TableSchema,
    prefix_placeholders: Prefix,
    mut field_placeholder: FieldPlaceholder,
) -> Result<String>
where
    Prefix: IntoIterator<Item = String>,
    FieldPlaceholder: FnMut(usize, &SchemaField) -> Result<String>,
{
    let mut placeholders = prefix_placeholders.into_iter().collect::<Vec<_>>();
    placeholders.extend(
        schema
            .columns
            .iter()
            .enumerate()
            .map(|(idx, field)| field_placeholder(idx, field))
            .collect::<Result<Vec<_>>>()?,
    );
    Ok(placeholders.join(", "))
}

pub(crate) fn inlined_insert_sql(table_identifier: &str, placeholders: &str) -> String {
    format!("INSERT INTO {table_identifier} VALUES ({placeholders})")
}

pub(crate) fn inlined_insert_metadata_values<Value, I64Value, NullValue>(
    row_id: i64,
    begin_snapshot: i64,
    mut i64_value: I64Value,
    null_value: NullValue,
) -> Vec<Value>
where
    I64Value: FnMut(i64) -> Value,
    NullValue: FnOnce() -> Value,
{
    vec![i64_value(row_id), i64_value(begin_snapshot), null_value()]
}

pub(crate) fn sql_cast_string_literal(value: &str, target_type: &str) -> String {
    format!("CAST({} AS {target_type})", sql_string_literal(value))
}

pub(crate) fn inlined_read_projection<F>(schema: &TableSchema, projection_expr: F) -> Result<String>
where
    F: FnMut(&SchemaField) -> Result<String>,
{
    Ok(schema
        .columns
        .iter()
        .map(projection_expr)
        .collect::<Result<Vec<_>>>()?
        .join(", "))
}

pub(crate) fn inlined_read_projection_for_dialect<Q>(
    schema: &TableSchema,
    dialect: InlinedSqlDialect,
    mut quote_identifier: Q,
) -> Result<String>
where
    Q: FnMut(&str) -> String,
{
    inlined_read_projection(schema, |field| {
        inlined_read_projection_expr(field, dialect, &mut quote_identifier)
    })
}

pub(crate) fn inlined_read_projection_expr<Q>(
    field: &SchemaField,
    dialect: InlinedSqlDialect,
    mut quote_identifier: Q,
) -> Result<String>
where
    Q: FnMut(&str) -> String,
{
    let quoted = quote_identifier(&field.name);
    match dialect {
        InlinedSqlDialect::Sqlite | InlinedSqlDialect::Quack => Ok(quoted),
        InlinedSqlDialect::DuckDbFile => {
            if duckdb_file_inlined_read_projection_casts_to_varchar(field) {
                Ok(format!("CAST({quoted} AS VARCHAR) AS {quoted}"))
            } else {
                Ok(quoted)
            }
        }
        InlinedSqlDialect::Postgres => {
            let expr = match field.ducklake_type {
                DuckLakeType::Varchar
                | DuckLakeType::Json
                | DuckLakeType::Blob
                | DuckLakeType::Variant
                | DuckLakeType::Geometry => format!("encode({quoted}, 'hex') AS {quoted}"),
                _ => format!("{quoted}::TEXT AS {quoted}"),
            };
            Ok(expr)
        }
    }
}

fn duckdb_file_inlined_read_projection_casts_to_varchar(field: &SchemaField) -> bool {
    matches!(
        field.ducklake_type,
        DuckLakeType::Decimal { .. }
            | DuckLakeType::Int128
            | DuckLakeType::UInt128
            | DuckLakeType::TimeNs
            | DuckLakeType::TimeTz
    )
}

pub(crate) fn read_inlined_data_sql(
    table_identifier: &str,
    projection: &str,
    snapshot_expr: &str,
) -> String {
    format!(
        "SELECT {projection} FROM {table_identifier} \
         WHERE {snapshot_expr} >= begin_snapshot \
           AND ({snapshot_expr} < end_snapshot OR end_snapshot IS NULL) \
         ORDER BY row_id"
    )
}

pub(crate) fn read_inlined_data_with_metadata_sql(
    table_identifier: &str,
    projection: &str,
    snapshot_expr: &str,
) -> String {
    let user_projection = if projection.is_empty() {
        String::new()
    } else {
        format!(", {projection}")
    };
    format!(
        "SELECT row_id, begin_snapshot{user_projection} FROM {table_identifier} \
         WHERE {snapshot_expr} >= begin_snapshot \
           AND ({snapshot_expr} < end_snapshot OR end_snapshot IS NULL) \
         ORDER BY row_id"
    )
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn write_inlined_batch_rows_sync<Value, Prefix, Encode, Insert>(
    schema: &TableSchema,
    batches: &[RecordBatch],
    preserved_row_ids: Option<&[i64]>,
    row_id_start: i64,
    begin_snapshot: i64,
    mut prefix_values: Prefix,
    mut encode_value: Encode,
    mut insert_row: Insert,
) -> Result<()>
where
    Prefix: FnMut(i64, i64) -> Vec<Value>,
    Encode: FnMut(&SchemaField, &dyn Array, usize) -> Result<Value>,
    Insert: FnMut(Vec<Value>) -> Result<()>,
{
    for row in plan_inlined_batch_rows(schema, batches, preserved_row_ids, row_id_start)? {
        let batch = &batches[row.batch_index];
        let mut values = prefix_values(row.row_id, begin_snapshot);
        values.reserve(schema.columns.len());
        for (column_index, field) in schema.columns.iter().enumerate() {
            values.push(encode_value(
                field,
                batch.column(column_index).as_ref(),
                row.row_index,
            )?);
        }
        insert_row(values)?;
    }
    Ok(())
}

pub(crate) async fn write_inlined_batch_rows_async<Value, Prefix, Encode, Insert, InsertFut>(
    schema: &TableSchema,
    batches: &[RecordBatch],
    preserved_row_ids: Option<&[i64]>,
    row_id_start: i64,
    begin_snapshot: i64,
    mut prefix_values: Prefix,
    mut encode_value: Encode,
    mut insert_row: Insert,
) -> Result<()>
where
    Prefix: FnMut(i64, i64) -> Vec<Value>,
    Encode: FnMut(&SchemaField, &dyn Array, usize) -> Result<Value>,
    Insert: FnMut(Vec<Value>) -> InsertFut,
    InsertFut: Future<Output = Result<()>>,
{
    for row in plan_inlined_batch_rows(schema, batches, preserved_row_ids, row_id_start)? {
        let batch = &batches[row.batch_index];
        let mut values = prefix_values(row.row_id, begin_snapshot);
        values.reserve(schema.columns.len());
        for (column_index, field) in schema.columns.iter().enumerate() {
            values.push(encode_value(
                field,
                batch.column(column_index).as_ref(),
                row.row_index,
            )?);
        }
        insert_row(values).await?;
    }
    Ok(())
}

pub(crate) fn plan_inlined_batch_rows(
    schema: &TableSchema,
    batches: &[RecordBatch],
    preserved_row_ids: Option<&[i64]>,
    row_id_start: i64,
) -> Result<Vec<InlinedBatchRow>> {
    let total_rows = batches.iter().try_fold(0_usize, |acc, batch| {
        acc.checked_add(batch.num_rows()).ok_or_else(|| {
            StorageError::Backend("DuckLake inlined append row count overflow".into())
        })
    })?;
    if let Some(row_ids) = preserved_row_ids {
        if row_ids.len() != total_rows {
            return Err(StorageError::Backend(
                "DuckLake inlined append preserved row-id count does not match rows".into(),
            ));
        }
        if row_ids.iter().any(|row_id| *row_id < 0) {
            return Err(StorageError::Backend(
                "DuckLake inlined append preserved row ids contain negative values".into(),
            ));
        }
    }

    let mut rows = Vec::with_capacity(total_rows);
    let mut row_id = row_id_start;
    let mut preserved_index = 0_usize;
    for (batch_index, batch) in batches.iter().enumerate() {
        if batch.num_columns() != schema.columns.len() {
            return Err(StorageError::Backend(format!(
                "DuckLake inlined append batch has {} columns but table has {} columns",
                batch.num_columns(),
                schema.columns.len()
            )));
        }
        for row_index in 0..batch.num_rows() {
            let emit_row_id = match preserved_row_ids {
                Some(row_ids) => {
                    let row_id = row_ids[preserved_index];
                    preserved_index += 1;
                    row_id
                }
                None => {
                    let current = row_id;
                    row_id = row_id.checked_add(1).ok_or_else(|| {
                        StorageError::Backend("DuckLake inlined append row_id overflow".into())
                    })?;
                    current
                }
            };
            rows.push(InlinedBatchRow {
                batch_index,
                row_index,
                row_id: emit_row_id,
            });
        }
    }
    Ok(rows)
}

pub(crate) fn plan_inlined_data_deletions(
    deletions: &[InlinedDataDeletionCommit],
) -> Result<Vec<InlinedDataDeletionPlan>> {
    let mut plans = Vec::new();
    for deletion in deletions {
        let row_ids = deletion.validated_row_ids()?;
        if row_ids.is_empty() {
            continue;
        }
        plans.push(InlinedDataDeletionPlan {
            table_name: deletion.table_name.clone(),
            row_ids,
            rows_deleted: deletion.rows_deleted,
        });
    }
    Ok(plans)
}

pub(crate) fn plan_flushed_inlined_data(
    flushed_tables: &[FlushedInlinedDataCommit],
) -> Result<Vec<FlushedInlinedDataPlan>> {
    let mut plans = Vec::new();
    for flushed in flushed_tables {
        let row_ids = flushed.validated_row_ids()?;
        if row_ids.is_empty() {
            continue;
        }
        let expected_rows = usize::try_from(flushed.rows_flushed).map_err(|_| {
            StorageError::Backend("DuckLake flushed inlined-data row count is out of range".into())
        })?;
        plans.push(FlushedInlinedDataPlan {
            table_name: flushed.table_name.clone(),
            row_ids,
            rows_flushed: flushed.rows_flushed,
            expected_rows,
        });
    }
    Ok(plans)
}

pub(crate) fn plan_inlined_file_deletions(
    table_id: TableId,
    deletions: &[InlinedFileDeletionCommit],
) -> Result<Option<InlinedFileDeletionPlan>> {
    if deletions.is_empty() {
        return Ok(None);
    }

    let mut rows = Vec::new();
    for deletion in deletions {
        for row_id in &deletion.row_ids {
            if *row_id < 0 {
                return Err(StorageError::Backend(format!(
                    "DuckLake inlined file deletion row id cannot be negative: {row_id}"
                )));
            }
            rows.push(InlinedFileDeletionRow {
                data_file_id: deletion.data_file_id.0,
                row_id: *row_id,
            });
        }
    }
    Ok(Some(InlinedFileDeletionPlan {
        table_name: inlined_file_deletion_table_name(table_id),
        rows,
    }))
}

pub(crate) fn plan_flushed_inlined_file_deletions(
    table_id: TableId,
    flushed_deletions: &[FlushedInlinedFileDeletionCommit],
) -> Result<Option<FlushedInlinedFileDeletionPlan>> {
    if flushed_deletions.is_empty() {
        return Ok(None);
    }

    let mut rows = Vec::new();
    for flushed in flushed_deletions {
        rows.extend(flushed.validated_rows()?);
    }
    Ok(Some(FlushedInlinedFileDeletionPlan {
        table_name: inlined_file_deletion_table_name(table_id),
        rows,
    }))
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn apply_inlined_data_deletions_sync<F>(
    deletions: &[InlinedDataDeletionCommit],
    mut close_rows: F,
) -> Result<i64>
where
    F: FnMut(&InlinedDataDeletionPlan) -> Result<Vec<(i64, usize)>>,
{
    let mut total_deleted = 0_i64;
    for deletion in plan_inlined_data_deletions(deletions)? {
        for (row_id, changed) in close_rows(&deletion)? {
            if changed != 1 {
                return Err(StorageError::Backend(format!(
                    "DuckLake inlined-data truncate expected one visible row in {} for row_id {}, updated {}",
                    deletion.table_name, row_id, changed
                )));
            }
        }
        total_deleted = checked_inlined_row_total(
            total_deleted,
            deletion.rows_deleted,
            "DuckLake inlined-data truncate overflow",
        )?;
    }
    Ok(total_deleted)
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn apply_inlined_data_deletions_row_by_row_sync<TableIdentifier, CloseRows>(
    deletions: &[InlinedDataDeletionCommit],
    new_snapshot: i64,
    mut table_identifier: TableIdentifier,
    snapshot_param: &str,
    row_id_param: &str,
    not_equal_op: &str,
    mut close_rows: CloseRows,
) -> Result<i64>
where
    TableIdentifier: FnMut(&str) -> String,
    CloseRows: FnMut(String, &[i64], i64) -> Result<Vec<(i64, usize)>>,
{
    apply_inlined_data_deletions_sync(deletions, |deletion| {
        let table_name = table_identifier(&deletion.table_name);
        let sql =
            close_inlined_data_row_sql(&table_name, snapshot_param, row_id_param, not_equal_op);
        close_rows(sql, &deletion.row_ids, new_snapshot)
    })
}

#[cfg(any(feature = "postgres", feature = "quack"))]
pub(crate) async fn apply_inlined_data_deletions_bulk_sql_async<
    TableIdentifier,
    ExecuteSql,
    ExecuteFut,
    QueryCount,
    QueryCountFut,
>(
    deletions: &[InlinedDataDeletionCommit],
    new_snapshot: i64,
    mut table_identifier: TableIdentifier,
    not_equal_op: &str,
    mut execute_sql: ExecuteSql,
    mut query_count: QueryCount,
) -> Result<i64>
where
    TableIdentifier: FnMut(&str) -> String,
    ExecuteSql: FnMut(String) -> ExecuteFut,
    ExecuteFut: Future<Output = Result<()>>,
    QueryCount: FnMut(String) -> QueryCountFut,
    QueryCountFut: Future<Output = Result<i64>>,
{
    let mut total_deleted = 0_i64;
    for deletion in plan_inlined_data_deletions(deletions)? {
        let table_name = table_identifier(&deletion.table_name);
        let row_ids_sql = crate::sql_utils::join_i64s(&deletion.row_ids);
        execute_sql(close_inlined_data_rows_in_sql(
            &table_name,
            new_snapshot,
            &row_ids_sql,
            not_equal_op,
        ))
        .await?;
        let changed = query_count(count_closed_inlined_data_rows_sql(
            &table_name,
            new_snapshot,
            &row_ids_sql,
        ))
        .await?;
        if changed != deletion.rows_deleted {
            return Err(StorageError::Backend(format!(
                "DuckLake inlined-data truncate expected {} visible rows in {}, closed {}",
                deletion.rows_deleted, deletion.table_name, changed
            )));
        }
        total_deleted = checked_inlined_row_total(
            total_deleted,
            deletion.rows_deleted,
            "DuckLake inlined-data truncate overflow",
        )?;
    }
    Ok(total_deleted)
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn delete_flushed_inlined_data_sync<F>(
    flushed_tables: &[FlushedInlinedDataCommit],
    mut delete_rows: F,
) -> Result<i64>
where
    F: FnMut(&FlushedInlinedDataPlan) -> Result<usize>,
{
    let mut total_flushed = 0_i64;
    for flushed in plan_flushed_inlined_data(flushed_tables)? {
        let deleted = delete_rows(&flushed)?;
        if deleted != flushed.expected_rows {
            return Err(StorageError::Backend(format!(
                "DuckLake flush expected to delete {} rows from {}, deleted {}",
                flushed.expected_rows, flushed.table_name, deleted
            )));
        }
        total_flushed = checked_inlined_row_total(
            total_flushed,
            flushed.rows_flushed,
            "DuckLake flushed inlined-data row count overflow",
        )?;
    }
    Ok(total_flushed)
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn delete_flushed_inlined_data_row_by_row_sync<TableIdentifier, DeleteRows>(
    flushed_tables: &[FlushedInlinedDataCommit],
    flush_snapshot: i64,
    mut table_identifier: TableIdentifier,
    snapshot_param: &str,
    row_id_param: &str,
    mut delete_rows: DeleteRows,
) -> Result<i64>
where
    TableIdentifier: FnMut(&str) -> String,
    DeleteRows: FnMut(String, &[i64], i64) -> Result<usize>,
{
    delete_flushed_inlined_data_sync(flushed_tables, |flushed| {
        let table_name = table_identifier(&flushed.table_name);
        let sql = delete_flushed_inlined_data_row_sql(&table_name, snapshot_param, row_id_param);
        delete_rows(sql, &flushed.row_ids, flush_snapshot)
    })
}

#[cfg(any(feature = "postgres", feature = "quack"))]
pub(crate) async fn delete_flushed_inlined_data_bulk_sql_async<
    TableIdentifier,
    ExecuteSql,
    ExecuteFut,
    QueryCount,
    QueryCountFut,
>(
    flushed_tables: &[FlushedInlinedDataCommit],
    flush_snapshot: i64,
    mut table_identifier: TableIdentifier,
    mut execute_sql: ExecuteSql,
    mut query_count: QueryCount,
) -> Result<i64>
where
    TableIdentifier: FnMut(&str) -> String,
    ExecuteSql: FnMut(String) -> ExecuteFut,
    ExecuteFut: Future<Output = Result<()>>,
    QueryCount: FnMut(String) -> QueryCountFut,
    QueryCountFut: Future<Output = Result<i64>>,
{
    let mut total_flushed = 0_i64;
    for flushed in plan_flushed_inlined_data(flushed_tables)? {
        let table_name = table_identifier(&flushed.table_name);
        let row_ids_sql = crate::sql_utils::join_i64s(&flushed.row_ids);
        let matching = query_count(visible_flushed_inlined_data_rows_sql(
            &table_name,
            flush_snapshot,
            &row_ids_sql,
        ))
        .await?;
        if matching != flushed.rows_flushed {
            return Err(StorageError::Backend(format!(
                "DuckLake flush expected to delete {} rows from {}, matched {}",
                flushed.rows_flushed, flushed.table_name, matching
            )));
        }
        execute_sql(delete_flushed_inlined_data_rows_sql(
            &table_name,
            flush_snapshot,
            &row_ids_sql,
        ))
        .await?;
        let remaining = query_count(visible_flushed_inlined_data_rows_sql(
            &table_name,
            flush_snapshot,
            &row_ids_sql,
        ))
        .await?;
        if remaining != 0 {
            return Err(StorageError::Backend(format!(
                "DuckLake flush expected no remaining rows in {}, found {}",
                flushed.table_name, remaining
            )));
        }
        total_flushed = checked_inlined_row_total(
            total_flushed,
            flushed.rows_flushed,
            "DuckLake flushed inlined-data row count overflow",
        )?;
    }
    Ok(total_flushed)
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn delete_flushed_inlined_file_deletions_sync<TableExists, DeleteRow>(
    table_id: TableId,
    flushed_deletions: &[FlushedInlinedFileDeletionCommit],
    mut table_exists: TableExists,
    mut delete_row: DeleteRow,
) -> Result<()>
where
    TableExists: FnMut(&str) -> Result<bool>,
    DeleteRow: FnMut(&str, &FlushedInlinedFileDeletion) -> Result<usize>,
{
    let Some(plan) = plan_flushed_inlined_file_deletions(table_id, flushed_deletions)? else {
        return Ok(());
    };
    let table_name = plan.table_name;
    if !table_exists(&table_name)? {
        return Err(StorageError::Backend(format!(
            "DuckLake flushed inlined-file deletion table {table_name} does not exist"
        )));
    }
    for row in plan.rows {
        let deleted = delete_row(&table_name, &row)?;
        if deleted != 1 {
            return Err(StorageError::Backend(format!(
                "DuckLake flush expected to delete one inlined-file deletion row from {}, deleted {}",
                table_name, deleted
            )));
        }
    }
    Ok(())
}

#[cfg(any(feature = "postgres", feature = "quack"))]
pub(crate) async fn delete_flushed_inlined_file_deletions_by_predicate_sql_async<
    TableIdentifier,
    TableExists,
    TableExistsFut,
    QueryCount,
    QueryCountFut,
    ExecuteSql,
    ExecuteFut,
>(
    table_id: TableId,
    flushed_deletions: &[FlushedInlinedFileDeletionCommit],
    mut table_identifier: TableIdentifier,
    mut table_exists: TableExists,
    mut query_count: QueryCount,
    mut execute_sql: ExecuteSql,
) -> Result<()>
where
    TableIdentifier: FnMut(&str) -> String,
    TableExists: FnMut(String) -> TableExistsFut,
    TableExistsFut: Future<Output = Result<bool>>,
    QueryCount: FnMut(String) -> QueryCountFut,
    QueryCountFut: Future<Output = Result<i64>>,
    ExecuteSql: FnMut(String) -> ExecuteFut,
    ExecuteFut: Future<Output = Result<()>>,
{
    let Some(plan) = plan_flushed_inlined_file_deletions(table_id, flushed_deletions)? else {
        return Ok(());
    };
    let table_name = plan.table_name;
    if !table_exists(table_name.clone()).await? {
        return Err(StorageError::Backend(format!(
            "DuckLake flushed inlined-file deletion table {table_name} does not exist"
        )));
    }
    let table = table_identifier(&table_name);
    for row in plan.rows {
        let predicate = flushed_inlined_file_deletion_predicate_sql(
            &row.data_file_id.0.to_string(),
            &row.row_id.to_string(),
            &row.begin_snapshot.0.to_string(),
        );
        let matching =
            query_count(format!("SELECT COUNT(*) FROM {table} WHERE {predicate}")).await?;
        if matching != 1 {
            return Err(StorageError::Backend(format!(
                "DuckLake flush expected one inlined-file deletion row in {}, matched {}",
                table_name, matching
            )));
        }
        execute_sql(format!("DELETE FROM {table} WHERE {predicate}")).await?;
    }
    Ok(())
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn write_inlined_file_deletions_sync<TableIdentifier, CreateTable, InsertRows>(
    table_id: TableId,
    deletions: &[InlinedFileDeletionCommit],
    new_snapshot: i64,
    mut table_identifier: TableIdentifier,
    insert_values_sql: &str,
    mut create_table: CreateTable,
    mut insert_rows: InsertRows,
) -> Result<()>
where
    TableIdentifier: FnMut(&str) -> String,
    CreateTable: FnMut(&str) -> Result<()>,
    InsertRows: FnMut(&str, &[InlinedFileDeletionRow], i64) -> Result<()>,
{
    let Some(plan) = plan_inlined_file_deletions(table_id, deletions)? else {
        return Ok(());
    };

    let table = table_identifier(&plan.table_name);
    create_table(&create_inlined_file_deletion_table_sql(&table))?;
    let insert_sql = insert_inlined_file_deletion_sql(&table, insert_values_sql);
    insert_rows(&insert_sql, &plan.rows, new_snapshot)
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn close_inlined_data_rows_for_table_sync<QueryNames, TableExists, CloseRows>(
    table_id: TableId,
    new_snapshot: i64,
    mut query_inlined_table_names: QueryNames,
    mut inlined_table_exists: TableExists,
    mut close_inlined_table_rows: CloseRows,
) -> Result<()>
where
    QueryNames: FnMut(&crate::catalog_queries::CatalogQuery) -> Result<Vec<String>>,
    TableExists: FnMut(&str) -> Result<bool>,
    CloseRows: FnMut(&str, i64) -> Result<()>,
{
    let table_names =
        query_inlined_table_names(&crate::catalog_queries::inlined_data_table_names(table_id))?;
    for table_name in table_names {
        if inlined_table_exists(&table_name)? {
            close_inlined_table_rows(&table_name, new_snapshot)?;
        }
    }
    Ok(())
}

pub(crate) async fn write_inlined_file_deletions_by_row_async<
    TableIdentifier,
    CreateTable,
    CreateFut,
    InsertRow,
    InsertRowFut,
>(
    table_id: TableId,
    deletions: &[InlinedFileDeletionCommit],
    new_snapshot: i64,
    mut table_identifier: TableIdentifier,
    mut create_table: CreateTable,
    mut insert_row: InsertRow,
) -> Result<()>
where
    TableIdentifier: FnMut(&str) -> String,
    CreateTable: FnMut(String) -> CreateFut,
    CreateFut: Future<Output = Result<()>>,
    InsertRow: FnMut(String, InlinedFileDeletionRow, i64) -> InsertRowFut,
    InsertRowFut: Future<Output = Result<()>>,
{
    let Some(plan) = plan_inlined_file_deletions(table_id, deletions)? else {
        return Ok(());
    };

    let table = table_identifier(&plan.table_name);
    create_table(create_inlined_file_deletion_table_sql(&table)).await?;
    for row in plan.rows {
        insert_row(table.clone(), row, new_snapshot).await?;
    }
    Ok(())
}

pub(crate) fn checked_inlined_row_total(
    total: i64,
    delta: i64,
    overflow_message: &'static str,
) -> Result<i64> {
    total
        .checked_add(delta)
        .ok_or_else(|| StorageError::Backend(overflow_message.into()))
}

pub(crate) fn create_inlined_file_deletion_table_sql(table: &str) -> String {
    format!(
        "CREATE TABLE IF NOT EXISTS {table}(file_id BIGINT, row_id BIGINT, begin_snapshot BIGINT)"
    )
}

pub(crate) fn insert_inlined_file_deletion_sql(table: &str, values: &str) -> String {
    format!("INSERT INTO {table} VALUES ({values})")
}

pub(crate) fn delete_inlined_file_deletions_for_files_sql(
    table: &str,
    source_ids: &[i64],
) -> String {
    format!(
        "DELETE FROM {table} WHERE file_id IN ({})",
        crate::sql_utils::join_i64s(source_ids)
    )
}

#[cfg(any(feature = "postgres", feature = "quack"))]
pub(crate) async fn delete_inlined_file_deletions_for_files_async<
    TableIdentifier,
    TableExists,
    TableExistsFut,
    Execute,
    ExecuteFut,
>(
    table_id: TableId,
    source_ids: &[i64],
    mut table_identifier: TableIdentifier,
    mut table_exists: TableExists,
    mut execute: Execute,
) -> Result<()>
where
    TableIdentifier: FnMut(&str) -> String,
    TableExists: FnMut(String) -> TableExistsFut,
    TableExistsFut: Future<Output = Result<bool>>,
    Execute: FnMut(String) -> ExecuteFut,
    ExecuteFut: Future<Output = Result<()>>,
{
    let table_name = inlined_file_deletion_table_name(table_id);
    if !table_exists(table_name.clone()).await? {
        return Ok(());
    }
    execute(delete_inlined_file_deletions_for_files_sql(
        &table_identifier(&table_name),
        source_ids,
    ))
    .await
}

pub(crate) fn close_inlined_data_row_sql(
    table: &str,
    snapshot_param: &str,
    row_id_param: &str,
    not_equal_op: &str,
) -> String {
    format!(
        "UPDATE {table} \
         SET end_snapshot = {snapshot_param} \
         WHERE row_id = {row_id_param} \
           AND end_snapshot IS NULL \
           AND begin_snapshot {not_equal_op} {snapshot_param}"
    )
}

pub(crate) fn close_inlined_data_rows_in_sql(
    table: &str,
    snapshot: i64,
    row_ids_sql: &str,
    not_equal_op: &str,
) -> String {
    format!(
        "UPDATE {table} \
         SET end_snapshot = {snapshot} \
         WHERE row_id IN ({row_ids_sql}) \
           AND end_snapshot IS NULL \
           AND begin_snapshot {not_equal_op} {snapshot}"
    )
}

pub(crate) fn close_all_inlined_data_rows_sql(table: &str, snapshot_expr: &str) -> String {
    format!("UPDATE {table} SET end_snapshot = {snapshot_expr} WHERE end_snapshot IS NULL")
}

pub(crate) fn count_closed_inlined_data_rows_sql(
    table: &str,
    snapshot: i64,
    row_ids_sql: &str,
) -> String {
    format!(
        "SELECT COUNT(*) FROM {table} \
         WHERE row_id IN ({row_ids_sql}) AND end_snapshot = {snapshot}"
    )
}

pub(crate) fn delete_flushed_inlined_data_row_sql(
    table: &str,
    snapshot_param: &str,
    row_id_param: &str,
) -> String {
    format!(
        "DELETE FROM {table} \
         WHERE {snapshot_param} >= begin_snapshot \
           AND ({snapshot_param} < end_snapshot OR end_snapshot IS NULL) \
           AND row_id = {row_id_param}"
    )
}

pub(crate) fn visible_flushed_inlined_data_rows_sql(
    table: &str,
    snapshot: i64,
    row_ids_sql: &str,
) -> String {
    format!(
        "SELECT COUNT(*) FROM {table} \
         WHERE {snapshot} >= begin_snapshot \
           AND ({snapshot} < end_snapshot OR end_snapshot IS NULL) \
           AND row_id IN ({row_ids_sql})"
    )
}

pub(crate) fn delete_flushed_inlined_data_rows_sql(
    table: &str,
    snapshot: i64,
    row_ids_sql: &str,
) -> String {
    format!(
        "DELETE FROM {table} \
         WHERE {snapshot} >= begin_snapshot \
           AND ({snapshot} < end_snapshot OR end_snapshot IS NULL) \
           AND row_id IN ({row_ids_sql})"
    )
}

pub(crate) fn delete_flushed_inlined_file_deletion_sql(
    table: &str,
    file_id_param: &str,
    row_id_param: &str,
    begin_snapshot_param: &str,
) -> String {
    let predicate = flushed_inlined_file_deletion_predicate_sql(
        file_id_param,
        row_id_param,
        begin_snapshot_param,
    );
    format!("DELETE FROM {table} WHERE {predicate}")
}

pub(crate) fn flushed_inlined_file_deletion_predicate_sql(
    file_id_param: &str,
    row_id_param: &str,
    begin_snapshot_param: &str,
) -> String {
    format!(
        "file_id = {file_id_param} \
         AND row_id = {row_id_param} \
         AND begin_snapshot = {begin_snapshot_param}"
    )
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
pub(crate) fn parse_inlined_utf8_text<'a>(raw: &'a [u8], backend: &str) -> Result<&'a str> {
    std::str::from_utf8(raw).map_err(|err| {
        StorageError::Backend(format!(
            "cannot decode {backend} text bytes as UTF-8: {err}"
        ))
    })
}

#[cfg(any(feature = "sqlite", feature = "duckdb", test))]
fn inlined_scalar_type_error(
    value: InlinedSqlScalarRef<'_>,
    backend: &str,
    target: &str,
) -> StorageError {
    let actual = match value {
        InlinedSqlScalarRef::Null => "NULL".into(),
        InlinedSqlScalarRef::Boolean(value) => format!("boolean {value}"),
        InlinedSqlScalarRef::Signed(value) => format!("integer {value}"),
        InlinedSqlScalarRef::Unsigned(value) => format!("unsigned integer {value}"),
        InlinedSqlScalarRef::Float(value) => format!("float {value}"),
        InlinedSqlScalarRef::DecimalText(value) => format!("decimal {value:?}"),
        InlinedSqlScalarRef::Text(value) => format!("text {:?}", String::from_utf8_lossy(value)),
        InlinedSqlScalarRef::Bytes(_) => "bytes".into(),
        InlinedSqlScalarRef::Unsupported(value) => value,
    };
    StorageError::Backend(format!(
        "cannot decode {backend} value {actual} as {target}"
    ))
}

pub(crate) fn parse_inlined_bool_text(raw: &str, backend: &str) -> Result<bool> {
    match raw.trim().to_ascii_lowercase().as_str() {
        "true" | "t" | "1" => Ok(true),
        "false" | "f" | "0" => Ok(false),
        _ => Err(StorageError::Backend(format!(
            "cannot decode {backend} text {raw:?} as boolean"
        ))),
    }
}

pub(crate) fn parse_inlined_i64_text(raw: &str, backend: &str) -> Result<i64> {
    raw.trim().parse::<i64>().map_err(|_| {
        StorageError::Backend(format!("cannot decode {backend} text {raw:?} as integer"))
    })
}

pub(crate) fn parse_inlined_u64_text(raw: &str, backend: &str) -> Result<u64> {
    raw.trim().parse::<u64>().map_err(|_| {
        StorageError::Backend(format!(
            "cannot decode {backend} text {raw:?} as unsigned integer"
        ))
    })
}

pub(crate) fn parse_inlined_f64_text(raw: &str, backend: &str) -> Result<f64> {
    raw.trim().parse::<f64>().map_err(|_| {
        StorageError::Backend(format!("cannot decode {backend} text {raw:?} as float"))
    })
}

pub(crate) fn parse_inlined_utf8_hex_bytes(raw: &str, backend: &str) -> Result<String> {
    let bytes = parse_inlined_hex_bytes(raw, backend)?;
    String::from_utf8(bytes).map_err(|err| {
        StorageError::Backend(format!(
            "cannot decode {backend} byte text {raw:?} as UTF-8: {err}"
        ))
    })
}

pub(crate) fn parse_inlined_decimal128(raw: &str, scale: u8, backend: &str) -> Result<i128> {
    let value = raw.trim();
    let negative = value.starts_with('-');
    let value = value
        .strip_prefix('-')
        .or_else(|| value.strip_prefix('+'))
        .unwrap_or(value);
    let mut parts = value.split('.');
    let whole = parts.next().unwrap_or("");
    let fraction = parts.next().unwrap_or("");
    if parts.next().is_some()
        || !whole.bytes().all(|b| b.is_ascii_digit())
        || !fraction.bytes().all(|b| b.is_ascii_digit())
        || fraction.len() > scale as usize
    {
        return Err(StorageError::Backend(format!(
            "cannot decode {backend} text {raw:?} as decimal"
        )));
    }

    let scale_factor = 10_i128.pow(scale.into());
    let whole = if whole.is_empty() {
        0
    } else {
        whole.parse::<i128>().map_err(|_| {
            StorageError::Backend(format!("cannot decode {backend} text {raw:?} as decimal"))
        })?
    };
    let padded_fraction = format!("{fraction:0<width$}", width = scale as usize);
    let fraction = if padded_fraction.is_empty() {
        0
    } else {
        padded_fraction.parse::<i128>().map_err(|_| {
            StorageError::Backend(format!("cannot decode {backend} text {raw:?} as decimal"))
        })?
    };
    let unscaled = whole
        .checked_mul(scale_factor)
        .and_then(|value| value.checked_add(fraction))
        .ok_or_else(|| {
            StorageError::Backend(format!("cannot decode {backend} text {raw:?} as decimal"))
        })?;
    Ok(if negative { -unscaled } else { unscaled })
}

pub(crate) fn parse_inlined_decimal256(raw: &str, scale: u8, backend: &str) -> Result<i256> {
    parse_inlined_decimal_unscaled_string(raw, scale, backend)?
        .parse::<i256>()
        .map_err(|_| {
            StorageError::Backend(format!("cannot decode {backend} text {raw:?} as decimal"))
        })
}

fn parse_inlined_decimal_unscaled_string(raw: &str, scale: u8, backend: &str) -> Result<String> {
    let value = raw.trim();
    let negative = value.starts_with('-');
    let value = value
        .strip_prefix('-')
        .or_else(|| value.strip_prefix('+'))
        .unwrap_or(value);
    let mut parts = value.split('.');
    let whole = parts.next().unwrap_or("");
    let fraction = parts.next().unwrap_or("");
    if parts.next().is_some()
        || !whole.bytes().all(|b| b.is_ascii_digit())
        || !fraction.bytes().all(|b| b.is_ascii_digit())
        || fraction.len() > scale as usize
    {
        return Err(StorageError::Backend(format!(
            "cannot decode {backend} text {raw:?} as decimal"
        )));
    }

    let padded_fraction = format!("{fraction:0<width$}", width = scale as usize);
    let digits = format!("{whole}{padded_fraction}");
    let digits = digits.trim_start_matches('0');
    if digits.is_empty() {
        return Ok("0".to_string());
    }
    Ok(if negative {
        format!("-{digits}")
    } else {
        digits.to_string()
    })
}

pub(crate) fn parse_inlined_date32(
    raw: &str,
    backend: &str,
    allow_special_values: bool,
) -> Result<i32> {
    if allow_special_values && let Some(days) = parse_special_date32(raw) {
        return Ok(days);
    }
    let date = NaiveDate::parse_from_str(raw.trim(), "%Y-%m-%d").map_err(|err| {
        StorageError::Backend(format!("cannot decode {backend} text as date: {err}"))
    })?;
    let epoch = NaiveDate::from_ymd_opt(1970, 1, 1).expect("valid epoch date");
    let days = date.signed_duration_since(epoch).num_days();
    i32::try_from(days)
        .map_err(|_| StorageError::Backend(format!("{backend} date {raw:?} is out of range")))
}

#[cfg(any(feature = "duckdb", test))]
pub(crate) fn format_inlined_date32(days: i32, backend: &str) -> Result<String> {
    let epoch = NaiveDate::from_ymd_opt(1970, 1, 1).expect("valid epoch date");
    epoch
        .checked_add_signed(chrono::Duration::days(i64::from(days)))
        .map(|date| date.format("%F").to_string())
        .ok_or_else(|| StorageError::Backend(format!("{backend} date {days} is out of range")))
}

pub(crate) fn parse_inlined_time64(raw: &str, unit: TimeUnit, backend: &str) -> Result<i64> {
    let time = NaiveTime::parse_from_str(raw.trim(), "%H:%M:%S%.f").map_err(|err| {
        StorageError::Backend(format!("cannot decode {backend} text as time: {err}"))
    })?;
    let nanos = i64::from(time.num_seconds_from_midnight())
        .checked_mul(1_000_000_000)
        .and_then(|value| value.checked_add(i64::from(time.nanosecond())))
        .ok_or_else(|| StorageError::Backend(format!("{backend} time {raw:?} is out of range")))?;
    match unit {
        TimeUnit::Microsecond => Ok(nanos / 1_000),
        TimeUnit::Nanosecond => Ok(nanos),
        _ => Err(StorageError::Backend(format!(
            "unsupported {backend} time unit {unit:?}"
        ))),
    }
}

pub(crate) fn parse_inlined_timestamp(
    raw: &str,
    unit: TimeUnit,
    backend: &str,
    allow_special_values: bool,
) -> Result<i64> {
    if allow_special_values && let Some(value) = parse_special_timestamp(raw) {
        return Ok(value);
    }
    let (datetime, offset_seconds) = parse_naive_datetime_and_offset(raw, backend)?;
    let utc = datetime.and_utc();
    let seconds = utc.timestamp().checked_sub(offset_seconds).ok_or_else(|| {
        StorageError::Backend(format!("{backend} timestamp {raw:?} is out of range"))
    })?;
    let nanos = i64::from(utc.timestamp_subsec_nanos());

    match unit {
        TimeUnit::Second => Ok(seconds),
        TimeUnit::Millisecond => seconds
            .checked_mul(1_000)
            .and_then(|value| value.checked_add(nanos / 1_000_000))
            .ok_or_else(|| {
                StorageError::Backend(format!("{backend} timestamp {raw:?} is out of range"))
            }),
        TimeUnit::Microsecond => seconds
            .checked_mul(1_000_000)
            .and_then(|value| value.checked_add(nanos / 1_000))
            .ok_or_else(|| {
                StorageError::Backend(format!("{backend} timestamp {raw:?} is out of range"))
            }),
        TimeUnit::Nanosecond => seconds
            .checked_mul(1_000_000_000)
            .and_then(|value| value.checked_add(nanos))
            .ok_or_else(|| {
                StorageError::Backend(format!("{backend} timestamp {raw:?} is out of range"))
            }),
    }
}

fn parse_special_date32(raw: &str) -> Option<i32> {
    match raw.trim().to_ascii_lowercase().as_str() {
        "infinity" | "+infinity" => Some(i32::MAX),
        "-infinity" => Some(-i32::MAX),
        _ => None,
    }
}

fn parse_special_timestamp(raw: &str) -> Option<i64> {
    match raw.trim().to_ascii_lowercase().as_str() {
        "infinity" | "+infinity" => Some(i64::MAX),
        "-infinity" => Some(-i64::MAX),
        _ => None,
    }
}

fn parse_naive_datetime_and_offset(raw: &str, backend: &str) -> Result<(NaiveDateTime, i64)> {
    let mut value = raw.trim().replace('T', " ");
    let offset_seconds = if let Some(without_z) = value.strip_suffix('Z') {
        value = without_z.to_string();
        0
    } else if let Some(idx) = value
        .get(19..)
        .and_then(|rest| rest.rfind(['+', '-']).map(|idx| idx + 19))
    {
        let offset = value[idx..].to_string();
        value.truncate(idx);
        crate::embedded_rows::parse_timezone_offset_seconds(&offset).ok_or_else(|| {
            StorageError::Backend(format!(
                "cannot decode {backend} timestamp offset {offset:?}"
            ))
        })?
    } else {
        0
    };

    let datetime =
        NaiveDateTime::parse_from_str(value.trim(), "%Y-%m-%d %H:%M:%S%.f").map_err(|err| {
            StorageError::Backend(format!("cannot decode {backend} text as timestamp: {err}"))
        })?;
    Ok((datetime, offset_seconds))
}

pub(crate) fn parse_inlined_uuid(raw: &str, backend: &str) -> Result<[u8; 16]> {
    crate::parse_uuid_bytes(raw, backend)
}

pub(crate) fn parse_inlined_hex_bytes(raw: &str, backend: &str) -> Result<Vec<u8>> {
    crate::parse_hex_bytes(raw, backend)
}

pub(crate) fn inlined_scalar_value(
    field: &SchemaField,
    array: &dyn Array,
    row: usize,
    context: &str,
    interval_format: InlinedIntervalFormat,
) -> Result<InlinedScalarValue> {
    if array.is_null(row) {
        return Ok(InlinedScalarValue::Null);
    }

    match &field.ducklake_type {
        DuckLakeType::Boolean => {
            let array = typed_array::<BooleanArray>(field, array)?;
            Ok(InlinedScalarValue::Boolean(array.value(row)))
        }
        DuckLakeType::Int8 => {
            let array = typed_array::<Int8Array>(field, array)?;
            Ok(InlinedScalarValue::Int8(array.value(row)))
        }
        DuckLakeType::Int16 => {
            let array = typed_array::<Int16Array>(field, array)?;
            Ok(InlinedScalarValue::Int16(array.value(row)))
        }
        DuckLakeType::Int32 => {
            let array = typed_array::<Int32Array>(field, array)?;
            Ok(InlinedScalarValue::Int32(array.value(row)))
        }
        DuckLakeType::Int64 => {
            let array = typed_array::<Int64Array>(field, array)?;
            Ok(InlinedScalarValue::Int64(array.value(row)))
        }
        DuckLakeType::UInt8 => {
            let array = typed_array::<UInt8Array>(field, array)?;
            Ok(InlinedScalarValue::UInt8(array.value(row)))
        }
        DuckLakeType::UInt16 => {
            let array = typed_array::<UInt16Array>(field, array)?;
            Ok(InlinedScalarValue::UInt16(array.value(row)))
        }
        DuckLakeType::UInt32 => {
            let array = typed_array::<UInt32Array>(field, array)?;
            Ok(InlinedScalarValue::UInt32(array.value(row)))
        }
        DuckLakeType::UInt64 => {
            let array = typed_array::<UInt64Array>(field, array)?;
            Ok(InlinedScalarValue::UInt64(array.value(row)))
        }
        DuckLakeType::Int128 | DuckLakeType::UInt128 => {
            let array = typed_array::<Decimal256Array>(field, array)?;
            Ok(InlinedScalarValue::Text(crate::format_decimal_i256(
                array.value(row),
                0,
            )))
        }
        DuckLakeType::Float32 => {
            let array = typed_array::<Float32Array>(field, array)?;
            Ok(InlinedScalarValue::Float32(array.value(row)))
        }
        DuckLakeType::Float64 => {
            let array = typed_array::<Float64Array>(field, array)?;
            Ok(InlinedScalarValue::Float64(array.value(row)))
        }
        DuckLakeType::Decimal { scale, .. } => match array.data_type() {
            DataType::Decimal128(_, _) => {
                let array = typed_array::<Decimal128Array>(field, array)?;
                Ok(InlinedScalarValue::Text(crate::format_decimal_i128(
                    array.value(row),
                    *scale,
                )))
            }
            DataType::Decimal256(_, _) => {
                let array = typed_array::<Decimal256Array>(field, array)?;
                Ok(InlinedScalarValue::Text(crate::format_decimal_i256(
                    array.value(row),
                    *scale,
                )))
            }
            actual => Err(inlined_append_type_error(field, actual)),
        },
        DuckLakeType::Date => {
            let array = typed_array::<Date32Array>(field, array)?;
            Ok(InlinedScalarValue::Text(crate::format_date32(
                array.value(row),
                context,
            )?))
        }
        DuckLakeType::Time => {
            let array = typed_array::<Time64MicrosecondArray>(field, array)?;
            Ok(InlinedScalarValue::Text(crate::format_time64(
                array.value(row),
                TimeUnit::Microsecond,
                context,
            )?))
        }
        DuckLakeType::TimeNs => {
            let array = typed_array::<Time64NanosecondArray>(field, array)?;
            Ok(InlinedScalarValue::Text(crate::format_time64(
                array.value(row),
                TimeUnit::Nanosecond,
                context,
            )?))
        }
        DuckLakeType::TimestampS => {
            let array = typed_array::<TimestampSecondArray>(field, array)?;
            Ok(InlinedScalarValue::Text(crate::format_timestamp(
                array.value(row),
                TimeUnit::Second,
                false,
                context,
            )?))
        }
        DuckLakeType::TimestampMs => {
            let array = typed_array::<TimestampMillisecondArray>(field, array)?;
            Ok(InlinedScalarValue::Text(crate::format_timestamp(
                array.value(row),
                TimeUnit::Millisecond,
                false,
                context,
            )?))
        }
        DuckLakeType::Timestamp | DuckLakeType::TimestampUs => {
            let array = typed_array::<TimestampMicrosecondArray>(field, array)?;
            Ok(InlinedScalarValue::Text(crate::format_timestamp(
                array.value(row),
                TimeUnit::Microsecond,
                false,
                context,
            )?))
        }
        DuckLakeType::TimestampTz => {
            let array = typed_array::<TimestampMicrosecondArray>(field, array)?;
            Ok(InlinedScalarValue::Text(crate::format_timestamp(
                array.value(row),
                TimeUnit::Microsecond,
                true,
                context,
            )?))
        }
        DuckLakeType::TimestampNs => {
            let array = typed_array::<TimestampNanosecondArray>(field, array)?;
            Ok(InlinedScalarValue::Text(crate::format_timestamp(
                array.value(row),
                TimeUnit::Nanosecond,
                false,
                context,
            )?))
        }
        DuckLakeType::TimeTz => {
            let array = typed_array::<StringArray>(field, array)?;
            Ok(InlinedScalarValue::Text(array.value(row).to_string()))
        }
        DuckLakeType::Interval => {
            let array = typed_array::<IntervalMonthDayNanoArray>(field, array)?;
            let value = match interval_format {
                InlinedIntervalFormat::PreserveNanoseconds => {
                    crate::format_interval_month_day_nano(array.value(row))
                }
                InlinedIntervalFormat::MicrosecondsOnly(backend) => {
                    crate::format_microsecond_interval_month_day_nano(array.value(row), backend)?
                }
            };
            Ok(InlinedScalarValue::Text(value))
        }
        DuckLakeType::Varchar | DuckLakeType::Json => {
            let array = typed_array::<StringArray>(field, array)?;
            Ok(InlinedScalarValue::Text(array.value(row).to_string()))
        }
        DuckLakeType::Blob => {
            let array = typed_array::<BinaryArray>(field, array)?;
            Ok(InlinedScalarValue::Bytes(array.value(row).to_vec()))
        }
        DuckLakeType::Uuid => {
            let array = typed_array::<FixedSizeBinaryArray>(field, array)?;
            let source = format!("{context} UUID");
            Ok(InlinedScalarValue::Text(crate::format_uuid_bytes(
                array.value(row),
                &source,
            )?))
        }
        other => Err(StorageError::Unsupported(format!(
            "Inlined append writes do not support DuckLake type {other}"
        ))),
    }
}

fn typed_array<'a, T>(field: &SchemaField, array: &'a dyn Array) -> Result<&'a T>
where
    T: Array + 'static,
{
    array
        .as_any()
        .downcast_ref::<T>()
        .ok_or_else(|| inlined_append_type_error(field, array.data_type()))
}

pub(crate) fn inlined_append_type_error(field: &SchemaField, actual: &DataType) -> StorageError {
    StorageError::Backend(format!(
        "DuckLake inlined append column {} expected {}, found {}",
        field.name, field.ducklake_type, actual
    ))
}
