use std::fmt;

use arrow::{
    array::ArrayRef,
    datatypes::{IntervalMonthDayNano, TimeUnit, i256},
    record_batch::RecordBatch,
};
use ducklake_catalog::{DuckLakeType, SchemaField, TableSchema};

use crate::inlined_values::{InlinedColumnBuffer, InlinedSqlDialect};
use crate::quack_protocol::{DuckDbValue, quack_decimal_text};
use crate::{
    Result, StorageError, duckdb_literal_to_json, parse_interval_month_day_nano, parse_uuid_bytes,
};

fn quack_inlined_array<'a>(
    field: &SchemaField,
    values: impl Iterator<Item = &'a DuckDbValue>,
) -> Result<ArrayRef> {
    let mut column = InlinedColumnBuffer::new(field, InlinedSqlDialect::Quack)?;
    let decoder = QuackInlinedValueDecoder {
        field_name: &field.name,
    };
    for value in values {
        crate::inlined_values::push_decoded_inlined_column(&mut column, value, &decoder)?;
    }
    column.into_array()
}

struct QuackInlinedValueDecoder<'a> {
    field_name: &'a str,
}

impl crate::inlined_values::InlinedSqlValueDecoder<&DuckDbValue> for QuackInlinedValueDecoder<'_> {
    fn backend_name(&self) -> &'static str {
        "Quack"
    }

    fn decode_bool(&self, value: &DuckDbValue) -> Result<Option<bool>> {
        quack_value_bool(value, self.field_name)
    }

    fn decode_i64(&self, value: &DuckDbValue) -> Result<Option<i64>> {
        quack_value_i64(value, self.field_name)
    }

    fn decode_u64(&self, value: &DuckDbValue) -> Result<Option<u64>> {
        quack_value_u64(value, self.field_name)
    }

    fn decode_f64(&self, value: &DuckDbValue) -> Result<Option<f64>> {
        quack_value_f64(value, self.field_name)
    }

    fn decode_string(&self, value: &DuckDbValue) -> Result<Option<String>> {
        quack_value_string(value, self.field_name)
            .or_else(|_| quack_value_timetz_string(value, self.field_name))
    }

    fn decode_bytes(&self, value: &DuckDbValue) -> Result<Option<Vec<u8>>> {
        quack_value_bytes(value, self.field_name)
    }

    fn decode_decimal128(&self, value: &DuckDbValue, scale: u8) -> Result<Option<i128>> {
        quack_value_decimal128(value, self.field_name, scale)
    }

    fn decode_decimal256(&self, value: &DuckDbValue, scale: u8) -> Result<Option<i256>> {
        quack_value_decimal256(value, self.field_name, scale)
    }

    fn decode_date32(&self, value: &DuckDbValue) -> Result<Option<i32>> {
        match value {
            DuckDbValue::Null => Ok(None),
            DuckDbValue::DateDays(value) => Ok(Some(*value)),
            other => Err(quack_value_type_error(self.field_name, "DATE", other)),
        }
    }

    fn decode_time64(&self, value: &DuckDbValue, unit: TimeUnit) -> Result<Option<i64>> {
        match (unit, value) {
            (_, DuckDbValue::Null) => Ok(None),
            (TimeUnit::Microsecond, DuckDbValue::TimeMicros(value)) => Ok(Some(*value)),
            (TimeUnit::Nanosecond, DuckDbValue::TimeNanos(value)) => Ok(Some(*value)),
            (TimeUnit::Microsecond, other) => {
                Err(quack_value_type_error(self.field_name, "TIME", other))
            }
            (TimeUnit::Nanosecond, other) => {
                Err(quack_value_type_error(self.field_name, "TIME_NS", other))
            }
            (other_unit, _) => Err(StorageError::Unsupported(format!(
                "Quack inlined TIME unit {other_unit:?} is not supported"
            ))),
        }
    }

    fn decode_timestamp(&self, value: &DuckDbValue, unit: TimeUnit) -> Result<Option<i64>> {
        match (unit, value) {
            (_, DuckDbValue::Null) => Ok(None),
            (TimeUnit::Second, DuckDbValue::TimestampSeconds(value)) => Ok(Some(*value)),
            (TimeUnit::Millisecond, DuckDbValue::TimestampMillis(value)) => Ok(Some(*value)),
            (TimeUnit::Microsecond, DuckDbValue::TimestampMicros(value))
            | (TimeUnit::Microsecond, DuckDbValue::TimestampTzMicros(value)) => Ok(Some(*value)),
            (TimeUnit::Nanosecond, DuckDbValue::TimestampNanos(value)) => Ok(Some(*value)),
            (TimeUnit::Second, other) => Err(quack_value_type_error(
                self.field_name,
                "TIMESTAMP_S",
                other,
            )),
            (TimeUnit::Millisecond, other) => Err(quack_value_type_error(
                self.field_name,
                "TIMESTAMP_MS",
                other,
            )),
            (TimeUnit::Microsecond, other) => {
                Err(quack_value_type_error(self.field_name, "TIMESTAMP", other))
            }
            (TimeUnit::Nanosecond, other) => Err(quack_value_type_error(
                self.field_name,
                "TIMESTAMP_NS",
                other,
            )),
        }
    }

    fn decode_interval(&self, value: &DuckDbValue) -> Result<Option<IntervalMonthDayNano>> {
        quack_value_interval(value, self.field_name)
    }

    fn decode_uuid(&self, value: &DuckDbValue) -> Result<Option<[u8; 16]>> {
        quack_value_uuid_bytes(value, self.field_name)
    }

    fn decode_nested(&self, field: &SchemaField, value: &DuckDbValue) -> Result<Option<String>> {
        if matches!(value, DuckDbValue::Null) {
            return Ok(None);
        }
        let mut json = String::new();
        quack_value_to_json(field, value, &mut json)?;
        Ok(Some(json))
    }
}

fn quack_value_to_json(field: &SchemaField, value: &DuckDbValue, json: &mut String) -> Result<()> {
    if matches!(value, DuckDbValue::Null) {
        json.push_str("null");
        return Ok(());
    }
    if !field.children.is_empty()
        && let DuckDbValue::Utf8(raw) = value
    {
        json.push_str(&duckdb_literal_to_json(field, raw)?);
        return Ok(());
    }

    match &field.ducklake_type {
        DuckLakeType::Struct => {
            let DuckDbValue::Struct(fields) = value else {
                return Err(quack_value_type_error(&field.name, "STRUCT", value));
            };
            json.push('{');
            for (idx, child) in field.children.iter().enumerate() {
                if idx > 0 {
                    json.push(',');
                }
                json.push_str(&crate::json_string(&child.name));
                json.push(':');
                let child_value = fields
                    .iter()
                    .find(|(name, _)| name == &child.name)
                    .map(|(_, value)| value)
                    .ok_or_else(|| {
                        StorageError::Backend(format!(
                            "Quack struct field {} missing child {}",
                            field.name, child.name
                        ))
                    })?;
                quack_value_to_json(child, child_value, json)?;
            }
            json.push('}');
        }
        DuckLakeType::List => {
            let child = field.children.first().ok_or_else(|| {
                StorageError::Backend(format!("DuckLake list field {} has no child", field.name))
            })?;
            let DuckDbValue::List(values) = value else {
                return Err(quack_value_type_error(&field.name, "LIST", value));
            };
            json.push('[');
            for (idx, value) in values.iter().enumerate() {
                if idx > 0 {
                    json.push(',');
                }
                quack_value_to_json(child, value, json)?;
            }
            json.push(']');
        }
        DuckLakeType::Map => {
            if field.children.len() != 2 {
                return Err(StorageError::Backend(format!(
                    "DuckLake map field {} must have key and value children",
                    field.name
                )));
            }
            let DuckDbValue::Map(entries) = value else {
                return Err(quack_value_type_error(&field.name, "MAP", value));
            };
            let key_field = &field.children[0];
            let value_field = &field.children[1];
            json.push('{');
            for (idx, (key, value)) in entries.iter().enumerate() {
                if idx > 0 {
                    json.push(',');
                }
                json.push_str(&crate::json_string(&quack_map_key_to_json_string(
                    key_field, key,
                )?));
                json.push(':');
                quack_value_to_json(value_field, value, json)?;
            }
            json.push('}');
        }
        DuckLakeType::Boolean => match value {
            DuckDbValue::Boolean(value) => json.push_str(if *value { "true" } else { "false" }),
            other => return Err(quack_value_type_error(&field.name, "BOOLEAN", other)),
        },
        DuckLakeType::Int8 | DuckLakeType::Int16 | DuckLakeType::Int32 | DuckLakeType::Int64 => {
            let value = quack_value_i64(value, &field.name)?.ok_or_else(|| {
                StorageError::Backend(format!(
                    "Quack field {} unexpectedly became NULL",
                    field.name
                ))
            })?;
            json.push_str(&value.to_string());
        }
        DuckLakeType::UInt8
        | DuckLakeType::UInt16
        | DuckLakeType::UInt32
        | DuckLakeType::UInt64 => {
            let value = quack_value_u64(value, &field.name)?.ok_or_else(|| {
                StorageError::Backend(format!(
                    "Quack field {} unexpectedly became NULL",
                    field.name
                ))
            })?;
            json.push_str(&value.to_string());
        }
        DuckLakeType::Int128 => {
            let value = quack_value_int128(value, &field.name)?.ok_or_else(|| {
                StorageError::Backend(format!(
                    "Quack field {} unexpectedly became NULL",
                    field.name
                ))
            })?;
            json.push_str(&value.to_string());
        }
        DuckLakeType::UInt128 => {
            let value = quack_value_uint128(value, &field.name)?.ok_or_else(|| {
                StorageError::Backend(format!(
                    "Quack field {} unexpectedly became NULL",
                    field.name
                ))
            })?;
            json.push_str(&value.to_string());
        }
        DuckLakeType::Float32 | DuckLakeType::Float64 => {
            let value = quack_value_f64(value, &field.name)?.ok_or_else(|| {
                StorageError::Backend(format!(
                    "Quack field {} unexpectedly became NULL",
                    field.name
                ))
            })?;
            json.push_str(&value.to_string());
        }
        DuckLakeType::Decimal { scale, .. } => {
            let value = quack_value_decimal256(value, &field.name, *scale)?.ok_or_else(|| {
                StorageError::Backend(format!(
                    "Quack field {} unexpectedly became NULL",
                    field.name
                ))
            })?;
            json.push_str(&quack_decimal_text(&value.to_string(), *scale));
        }
        DuckLakeType::Date => {
            let DuckDbValue::DateDays(value) = value else {
                return Err(quack_value_type_error(&field.name, "DATE", value));
            };
            json.push_str(&value.to_string());
        }
        DuckLakeType::Time => {
            let DuckDbValue::TimeMicros(value) = value else {
                return Err(quack_value_type_error(&field.name, "TIME", value));
            };
            json.push_str(&value.to_string());
        }
        DuckLakeType::TimeNs => {
            let DuckDbValue::TimeNanos(value) = value else {
                return Err(quack_value_type_error(&field.name, "TIME_NS", value));
            };
            json.push_str(&value.to_string());
        }
        DuckLakeType::TimestampS => {
            let DuckDbValue::TimestampSeconds(value) = value else {
                return Err(quack_value_type_error(&field.name, "TIMESTAMP_S", value));
            };
            json.push_str(&value.to_string());
        }
        DuckLakeType::TimestampMs => {
            let DuckDbValue::TimestampMillis(value) = value else {
                return Err(quack_value_type_error(&field.name, "TIMESTAMP_MS", value));
            };
            json.push_str(&value.to_string());
        }
        DuckLakeType::Timestamp | DuckLakeType::TimestampUs => {
            let DuckDbValue::TimestampMicros(value) = value else {
                return Err(quack_value_type_error(&field.name, "TIMESTAMP", value));
            };
            json.push_str(&value.to_string());
        }
        DuckLakeType::TimestampTz => {
            let DuckDbValue::TimestampTzMicros(value) = value else {
                return Err(quack_value_type_error(&field.name, "TIMESTAMPTZ", value));
            };
            json.push_str(&value.to_string());
        }
        DuckLakeType::TimestampNs => {
            let DuckDbValue::TimestampNanos(value) = value else {
                return Err(quack_value_type_error(&field.name, "TIMESTAMP_NS", value));
            };
            json.push_str(&value.to_string());
        }
        DuckLakeType::TimeTz => {
            let value = quack_value_timetz_string(value, &field.name)?.ok_or_else(|| {
                StorageError::Backend(format!(
                    "Quack field {} unexpectedly became NULL",
                    field.name
                ))
            })?;
            json.push_str(&crate::json_string(&value));
        }
        DuckLakeType::Varchar | DuckLakeType::Json => {
            let value = quack_value_string(value, &field.name)?.ok_or_else(|| {
                StorageError::Backend(format!(
                    "Quack field {} unexpectedly became NULL",
                    field.name
                ))
            })?;
            json.push_str(&crate::json_string(&value));
        }
        DuckLakeType::Blob | DuckLakeType::Variant | DuckLakeType::Geometry => {
            let value = quack_value_bytes(value, &field.name)?.ok_or_else(|| {
                StorageError::Backend(format!(
                    "Quack field {} unexpectedly became NULL",
                    field.name
                ))
            })?;
            json.push_str(&crate::json_string(&crate::hex_encode_bytes(&value)));
        }
        DuckLakeType::Uuid => {
            let value = quack_value_uuid_bytes(value, &field.name)?.ok_or_else(|| {
                StorageError::Backend(format!(
                    "Quack field {} unexpectedly became NULL",
                    field.name
                ))
            })?;
            json.push_str(&crate::json_string(&crate::hex_encode_bytes(&value)));
        }
        other => {
            return Err(StorageError::Unsupported(format!(
                "Quack inlined nested DuckLake child type {other} is not supported"
            )));
        }
    }
    Ok(())
}

fn quack_map_key_to_json_string(field: &SchemaField, value: &DuckDbValue) -> Result<String> {
    match &field.ducklake_type {
        DuckLakeType::Varchar | DuckLakeType::Json => quack_value_string(value, &field.name)?
            .ok_or_else(|| StorageError::Backend(format!("Quack map key {} is NULL", field.name))),
        DuckLakeType::Boolean => quack_value_bool(value, &field.name)?
            .map(|value| value.to_string())
            .ok_or_else(|| StorageError::Backend(format!("Quack map key {} is NULL", field.name))),
        DuckLakeType::Int8 | DuckLakeType::Int16 | DuckLakeType::Int32 | DuckLakeType::Int64 => {
            quack_value_i64(value, &field.name)?
                .map(|value| value.to_string())
                .ok_or_else(|| {
                    StorageError::Backend(format!("Quack map key {} is NULL", field.name))
                })
        }
        DuckLakeType::UInt8
        | DuckLakeType::UInt16
        | DuckLakeType::UInt32
        | DuckLakeType::UInt64 => quack_value_u64(value, &field.name)?
            .map(|value| value.to_string())
            .ok_or_else(|| StorageError::Backend(format!("Quack map key {} is NULL", field.name))),
        other => Err(StorageError::Unsupported(format!(
            "Quack inlined map key DuckLake type {other} is not supported"
        ))),
    }
}

fn quack_value_bool(value: &DuckDbValue, field_name: &str) -> Result<Option<bool>> {
    match value {
        DuckDbValue::Null => Ok(None),
        DuckDbValue::Boolean(value) => Ok(Some(*value)),
        other => Err(quack_value_type_error(field_name, "BOOLEAN", other)),
    }
}

fn quack_value_i64(value: &DuckDbValue, field_name: &str) -> Result<Option<i64>> {
    let value = match value {
        DuckDbValue::Null => return Ok(None),
        DuckDbValue::Int8(value) => i64::from(*value),
        DuckDbValue::Int16(value) => i64::from(*value),
        DuckDbValue::Int32(value) => i64::from(*value),
        DuckDbValue::Int64(value) => *value,
        other => return Err(quack_value_type_error(field_name, "signed integer", other)),
    };
    Ok(Some(value))
}

fn quack_value_u64(value: &DuckDbValue, field_name: &str) -> Result<Option<u64>> {
    let value = match value {
        DuckDbValue::Null => return Ok(None),
        DuckDbValue::UInt8(value) => u64::from(*value),
        DuckDbValue::UInt16(value) => u64::from(*value),
        DuckDbValue::UInt32(value) => u64::from(*value),
        DuckDbValue::UInt64(value) => *value,
        other => {
            return Err(quack_value_type_error(
                field_name,
                "unsigned integer",
                other,
            ));
        }
    };
    Ok(Some(value))
}

fn quack_value_f64(value: &DuckDbValue, field_name: &str) -> Result<Option<f64>> {
    let value = match value {
        DuckDbValue::Null => return Ok(None),
        DuckDbValue::Float32(value) => f64::from(*value),
        DuckDbValue::Float64(value) => *value,
        other => return Err(quack_value_type_error(field_name, "floating point", other)),
    };
    Ok(Some(value))
}

fn quack_value_string(value: &DuckDbValue, field_name: &str) -> Result<Option<String>> {
    match value {
        DuckDbValue::Null => Ok(None),
        DuckDbValue::Utf8(value) => Ok(Some(value.clone())),
        other => Err(quack_value_type_error(field_name, "VARCHAR", other)),
    }
}

fn quack_value_bytes(value: &DuckDbValue, field_name: &str) -> Result<Option<Vec<u8>>> {
    match value {
        DuckDbValue::Null => Ok(None),
        DuckDbValue::Blob(value) => Ok(Some(value.clone())),
        other => Err(quack_value_type_error(field_name, "BLOB", other)),
    }
}

fn quack_value_int128(value: &DuckDbValue, field_name: &str) -> Result<Option<i128>> {
    let value = match value {
        DuckDbValue::Null => return Ok(None),
        DuckDbValue::Int8(value) => i128::from(*value),
        DuckDbValue::Int16(value) => i128::from(*value),
        DuckDbValue::Int32(value) => i128::from(*value),
        DuckDbValue::Int64(value) => i128::from(*value),
        DuckDbValue::Int128(value) => *value,
        DuckDbValue::Decimal {
            value, scale: 0, ..
        } => *value,
        other => return Err(quack_value_type_error(field_name, "HUGEINT", other)),
    };
    Ok(Some(value))
}

fn quack_value_uint128(value: &DuckDbValue, field_name: &str) -> Result<Option<i256>> {
    let value = match value {
        DuckDbValue::Null => return Ok(None),
        DuckDbValue::UInt8(value) => i256::from_parts(u128::from(*value), 0),
        DuckDbValue::UInt16(value) => i256::from_parts(u128::from(*value), 0),
        DuckDbValue::UInt32(value) => i256::from_parts(u128::from(*value), 0),
        DuckDbValue::UInt64(value) => i256::from_parts(u128::from(*value), 0),
        DuckDbValue::UInt128(value) => i256::from_parts(*value, 0),
        DuckDbValue::Decimal {
            value, scale: 0, ..
        } if *value >= 0 => i256::from_i128(*value),
        DuckDbValue::Decimal {
            value, scale: 0, ..
        } => {
            return Err(quack_range_error(field_name, value, "UHUGEINT"));
        }
        other => return Err(quack_value_type_error(field_name, "UHUGEINT", other)),
    };
    Ok(Some(value))
}

fn quack_value_decimal128(
    value: &DuckDbValue,
    field_name: &str,
    expected_scale: u8,
) -> Result<Option<i128>> {
    let value = match value {
        DuckDbValue::Null => return Ok(None),
        DuckDbValue::Decimal { value, scale, .. } if *scale == expected_scale => *value,
        DuckDbValue::Decimal { scale, .. } => {
            return Err(StorageError::Backend(format!(
                "Quack inlined field {field_name} DECIMAL scale {scale} does not match DuckLake scale {expected_scale}"
            )));
        }
        DuckDbValue::Int8(value) if expected_scale == 0 => i128::from(*value),
        DuckDbValue::Int16(value) if expected_scale == 0 => i128::from(*value),
        DuckDbValue::Int32(value) if expected_scale == 0 => i128::from(*value),
        DuckDbValue::Int64(value) if expected_scale == 0 => i128::from(*value),
        DuckDbValue::Int128(value) if expected_scale == 0 => *value,
        other => return Err(quack_value_type_error(field_name, "DECIMAL", other)),
    };
    Ok(Some(value))
}

fn quack_value_decimal256(
    value: &DuckDbValue,
    field_name: &str,
    expected_scale: u8,
) -> Result<Option<i256>> {
    let value = match value {
        DuckDbValue::Null => return Ok(None),
        DuckDbValue::Decimal { value, scale, .. } if *scale == expected_scale => {
            i256::from_i128(*value)
        }
        DuckDbValue::Decimal { scale, .. } => {
            return Err(StorageError::Backend(format!(
                "Quack inlined field {field_name} DECIMAL scale {scale} does not match DuckLake scale {expected_scale}"
            )));
        }
        DuckDbValue::Int128(value) if expected_scale == 0 => i256::from_i128(*value),
        DuckDbValue::UInt128(value) if expected_scale == 0 => i256::from_parts(*value, 0),
        DuckDbValue::UInt8(value) if expected_scale == 0 => i256::from_parts(u128::from(*value), 0),
        DuckDbValue::UInt16(value) if expected_scale == 0 => {
            i256::from_parts(u128::from(*value), 0)
        }
        DuckDbValue::UInt32(value) if expected_scale == 0 => {
            i256::from_parts(u128::from(*value), 0)
        }
        DuckDbValue::UInt64(value) if expected_scale == 0 => {
            i256::from_parts(u128::from(*value), 0)
        }
        other => return Err(quack_value_type_error(field_name, "DECIMAL256", other)),
    };
    Ok(Some(value))
}

fn quack_value_timetz_string(value: &DuckDbValue, field_name: &str) -> Result<Option<String>> {
    match value {
        DuckDbValue::Null => Ok(None),
        DuckDbValue::Utf8(value) => Ok(Some(value.clone())),
        DuckDbValue::TimeTzMicros(value) => Ok(Some(quack_format_timetz(*value))),
        other => Err(quack_value_type_error(field_name, "TIMETZ", other)),
    }
}

fn quack_value_interval(
    value: &DuckDbValue,
    field_name: &str,
) -> Result<Option<IntervalMonthDayNano>> {
    match value {
        DuckDbValue::Null => Ok(None),
        DuckDbValue::Interval {
            months,
            days,
            micros,
        } => {
            let nanos = micros.checked_mul(1_000).ok_or_else(|| {
                quack_range_error(field_name, micros, "INTERVAL MonthDayNano nanoseconds")
            })?;
            Ok(Some(IntervalMonthDayNano::new(*months, *days, nanos)))
        }
        DuckDbValue::Utf8(value) => parse_interval_month_day_nano(value, "Quack").map(Some),
        other => Err(quack_value_type_error(field_name, "INTERVAL", other)),
    }
}

fn quack_value_uuid_bytes(value: &DuckDbValue, field_name: &str) -> Result<Option<[u8; 16]>> {
    match value {
        DuckDbValue::Null => Ok(None),
        DuckDbValue::Uuid(value) => Ok(Some(*value)),
        DuckDbValue::Blob(value) => value
            .as_slice()
            .try_into()
            .map(Some)
            .map_err(|_| quack_range_error(field_name, value.len(), "UUID BLOB length 16")),
        DuckDbValue::Utf8(value) => parse_uuid_bytes(value, "Quack").map(Some),
        other => Err(quack_value_type_error(field_name, "UUID", other)),
    }
}
fn quack_format_timetz(raw: i64) -> String {
    const OFFSET_BITS: u32 = 24;
    const OFFSET_MASK: u64 = (1_u64 << OFFSET_BITS) - 1;
    const MAX_OFFSET: i32 = 16 * 60 * 60 - 1;

    let bits = raw as u64;
    let micros = (bits >> OFFSET_BITS) as i64;
    let offset = MAX_OFFSET - (bits & OFFSET_MASK) as i32;
    let sign = if offset < 0 { '-' } else { '+' };
    let mut offset_seconds = offset.abs();
    let offset_hours = offset_seconds / 3_600;
    offset_seconds %= 3_600;
    let offset_minutes = offset_seconds / 60;
    let offset_seconds = offset_seconds % 60;

    let mut out = quack_format_time_micros(micros);
    out.push(sign);
    out.push_str(&format!("{offset_hours:02}"));
    if offset_minutes != 0 || offset_seconds != 0 {
        out.push_str(&format!(":{offset_minutes:02}"));
    }
    if offset_seconds != 0 {
        out.push_str(&format!(":{offset_seconds:02}"));
    }
    out
}

fn quack_format_time_micros(micros: i64) -> String {
    const MICROS_PER_SEC: i64 = 1_000_000;
    const MICROS_PER_MINUTE: i64 = 60 * MICROS_PER_SEC;
    const MICROS_PER_HOUR: i64 = 60 * MICROS_PER_MINUTE;

    let mut remaining = micros;
    let hour = remaining / MICROS_PER_HOUR;
    remaining -= hour * MICROS_PER_HOUR;
    let minute = remaining / MICROS_PER_MINUTE;
    remaining -= minute * MICROS_PER_MINUTE;
    let second = remaining / MICROS_PER_SEC;
    remaining -= second * MICROS_PER_SEC;

    if remaining == 0 {
        format!("{hour:02}:{minute:02}:{second:02}")
    } else {
        let fraction = format!("{remaining:06}");
        format!(
            "{hour:02}:{minute:02}:{second:02}.{}",
            fraction.trim_end_matches('0')
        )
    }
}

fn quack_range_error(
    field_name: &str,
    value: impl fmt::Display,
    target_type: &str,
) -> StorageError {
    StorageError::Backend(format!(
        "Quack inlined field {field_name} value {value} is out of range for {target_type}"
    ))
}

fn quack_value_type_error(field_name: &str, expected: &str, actual: &DuckDbValue) -> StorageError {
    StorageError::Backend(format!(
        "Quack inlined field {field_name} expected {expected}, got {actual:?}"
    ))
}

pub(super) fn quack_inlined_record_batch(
    schema: &TableSchema,
    rows: &[Vec<DuckDbValue>],
) -> Result<Option<RecordBatch>> {
    if schema.columns.is_empty() || rows.is_empty() {
        return Ok(None);
    }
    for row in rows {
        if row.len() != schema.columns.len() {
            return Err(StorageError::Backend(format!(
                "Quack inlined row column mismatch: expected {}, got {}",
                schema.columns.len(),
                row.len()
            )));
        }
    }

    let arrays = schema
        .columns
        .iter()
        .enumerate()
        .map(|(idx, field)| quack_inlined_array(field, rows.iter().map(move |row| &row[idx])))
        .collect::<Result<Vec<_>>>()?;
    crate::inlined_values::inlined_record_batch_from_arrays(schema, arrays)
}
