laminar_connectors/schema/json/

decoder.rs

//! JSON format decoder implementing [`FormatDecoder`].
//!
//! Converts raw JSON byte payloads into Arrow `RecordBatch`es.
//! Constructed once at `CREATE SOURCE` time with a frozen Arrow schema;
//! the decoder is stateless after construction so the Ring 1 hot path
//! has zero schema lookups.
#![allow(clippy::disallowed_types)] // cold path: schema management

use std::collections::HashMap;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;

use arrow_array::builder::{
    BooleanBuilder, Float32Builder, Float64Builder, Int16Builder, Int32Builder, Int64Builder,
    Int8Builder, LargeBinaryBuilder, LargeStringBuilder, ListBuilder, StringBuilder,
    TimestampMicrosecondBuilder, TimestampMillisecondBuilder, TimestampNanosecondBuilder,
    TimestampSecondBuilder, UInt16Builder, UInt32Builder, UInt64Builder, UInt8Builder,
};
use arrow_array::{ArrayRef, RecordBatch};
use arrow_schema::{DataType, SchemaRef, TimeUnit};

use crate::schema::error::{SchemaError, SchemaResult};
use crate::schema::json::jsonb::JsonbEncoder;
use crate::schema::traits::FormatDecoder;
use crate::schema::types::RawRecord;

/// Strategy for JSON fields not in the Arrow schema.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum UnknownFieldStrategy {
    /// Silently ignore unknown fields (default).
    Ignore,
    /// Collect unknown fields into an `_extra` `LargeBinary` (JSONB) column.
    CollectExtra,
    /// Return a decode error if any unknown field is encountered.
    Reject,
}

/// Strategy for JSON values that don't match the expected Arrow type.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum TypeMismatchStrategy {
    /// Insert null and increment the mismatch counter (default).
    Null,
    /// Attempt coercion (e.g., `"123"` → `123` for Int64 columns).
    Coerce,
    /// Return a decode error on the first mismatch.
    Reject,
}

impl TypeMismatchStrategy {
    /// Parse from a `WITH` option value (`schema.enforcement`).
    #[must_use]
    pub fn from_enforcement_str(s: &str) -> Option<Self> {
        match s.to_lowercase().as_str() {
            "coerce" => Some(Self::Coerce),
            "strict" => Some(Self::Reject),
            "permissive" => Some(Self::Null),
            _ => None,
        }
    }
}

/// Epoch unit for *numeric* JSON timestamps feeding a `Timestamp` column
/// (strings still go through `timestamp_formats`). Default `Millis`
/// preserves historical behaviour.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum EpochUnit {
    /// Epoch seconds.
    Seconds,
    /// Epoch milliseconds (default).
    #[default]
    Millis,
    /// Epoch microseconds.
    Micros,
    /// Epoch nanoseconds.
    Nanos,
}

impl EpochUnit {
    /// Parse a `json.column.<col>.epoch_unit` value. Accepts the bare
    /// unit and the `epoch_*` spelling the WebSocket connector's
    /// `event.time.format` already uses.
    #[must_use]
    pub fn from_str_opt(s: &str) -> Option<Self> {
        match s.trim().to_lowercase().as_str() {
            "seconds" | "epoch_seconds" => Some(Self::Seconds),
            "millis" | "epoch_millis" => Some(Self::Millis),
            "micros" | "epoch_micros" => Some(Self::Micros),
            "nanos" | "epoch_nanos" => Some(Self::Nanos),
            _ => None,
        }
    }

    /// Nanoseconds per one unit (used for lossless scaling).
    const fn nanos_per(self) -> i64 {
        match self {
            Self::Seconds => 1_000_000_000,
            Self::Millis => 1_000_000,
            Self::Micros => 1_000,
            Self::Nanos => 1,
        }
    }
}

/// Per-column extraction strategy, pre-computed at Ring 2.
#[derive(Debug, Clone)]
enum ColumnExtraction {
    /// Extract field from the default path target (json.path or root).
    DefaultPath,
    /// Extract field from a custom absolute path from root.
    CustomPath { segments: Vec<String> },
}

/// JSON decoder configuration.
#[derive(Debug, Clone)]
pub struct JsonDecoderConfig {
    /// How to handle fields present in JSON but absent from the schema.
    pub unknown_fields: UnknownFieldStrategy,

    /// How to handle type mismatches.
    pub type_mismatch: TypeMismatchStrategy,

    /// Timestamp format patterns to try when parsing string values
    /// into Timestamp columns. Tried in order; first match wins.
    /// Use `"iso8601"` for RFC 3339 / ISO 8601 auto-detection.
    pub timestamp_formats: Vec<String>,

    /// Whether to encode nested objects as JSONB binary format
    /// instead of JSON-serialized Utf8. When true, nested objects
    /// become `LargeBinary` columns with JSONB encoding.
    pub nested_as_jsonb: bool,

    /// Dot-separated path to navigate before field extraction.
    /// e.g. `"data"` → navigate into `{"data": {...}}` before extraction.
    /// e.g. `"data.trade"` → navigate into `{"data":{"trade":{...}}}`.
    pub json_path: Option<Vec<String>>,

    /// Per-column absolute path overrides: `column_name` → path segments.
    /// Parsed from `json.column.<name> = 'path.to.field'` options.
    /// These paths are absolute from the root, ignoring `json_path`.
    pub json_column_paths: HashMap<String, Vec<String>>,

    /// Per-column numeric-timestamp epoch unit, from
    /// `json.column.<name>.epoch_unit`. Absent ⇒ [`EpochUnit::Millis`].
    pub numeric_timestamp_units: HashMap<String, EpochUnit>,

    /// Column names for array-to-rows expansion.
    /// When set, the target (after `json_path`) must be an array.
    /// Each element becomes a row; positional names map array elements to columns.
    pub json_explode: Option<Vec<String>>,
}

impl Default for JsonDecoderConfig {
    fn default() -> Self {
        Self {
            unknown_fields: UnknownFieldStrategy::Ignore,
            type_mismatch: TypeMismatchStrategy::Coerce,
            timestamp_formats: vec![
                "iso8601".into(),
                "%Y-%m-%dT%H:%M:%S%.fZ".into(),
                "%Y-%m-%dT%H:%M:%S%.f%:z".into(),
                "%Y-%m-%d %H:%M:%S%.f".into(),
                "%Y-%m-%d %H:%M:%S".into(),
            ],
            nested_as_jsonb: false,
            json_path: None,
            json_column_paths: HashMap::new(),
            numeric_timestamp_units: HashMap::new(),
            json_explode: None,
        }
    }
}

impl JsonDecoderConfig {
    /// Build config from a [`ConnectorConfig`](crate::config::ConnectorConfig).
    ///
    /// Parses `json.path`, `json.column.*`, `json.explode`,
    /// `schema.enforcement`, and `nested.as.jsonb` properties.
    /// Called once at Ring 2 (`CREATE SOURCE` time).
    #[must_use]
    pub fn from_connector_config(config: &crate::config::ConnectorConfig) -> Self {
        let mut cfg = Self::default();

        if let Some(path) = config.get("json.path") {
            cfg.json_path = Some(path.split('.').map(ToString::to_string).collect());
        }

        let col_props = config.properties_with_prefix("json.column.");
        for (col_name, val) in col_props {
            // `json.column.<col>.epoch_unit = '<unit>'` configures numeric
            // timestamp scaling for <col>; everything else is a path.
            // (Column names are SQL identifiers — no dots — so the suffix
            // split is unambiguous against dotted path *values*.)
            if let Some(col) = col_name.strip_suffix(".epoch_unit") {
                if let Some(unit) = EpochUnit::from_str_opt(&val) {
                    cfg.numeric_timestamp_units.insert(col.to_string(), unit);
                } else {
                    tracing::warn!(
                        column = col,
                        value = %val,
                        "invalid json.column.<col>.epoch_unit (expected \
                         seconds|millis|micros|nanos); using millis"
                    );
                }
                continue;
            }
            cfg.json_column_paths
                .insert(col_name, val.split('.').map(ToString::to_string).collect());
        }

        if let Some(explode) = config.get("json.explode") {
            cfg.json_explode = Some(explode.split(',').map(|s| s.trim().to_string()).collect());
        }

        if let Some(enforcement) = config.get("schema.enforcement") {
            if let Some(strategy) = TypeMismatchStrategy::from_enforcement_str(enforcement) {
                cfg.type_mismatch = strategy;
            }
        }
        if let Some(v) = config.get("nested.as.jsonb") {
            cfg.nested_as_jsonb = v.eq_ignore_ascii_case("true");
        }
        cfg
    }
}

/// Decodes JSON byte payloads into Arrow `RecordBatch`es.
///
/// # Ring Placement
///
/// - **Ring 1**: `decode_batch()` — parse JSON, build columnar Arrow output
/// - **Ring 2**: Construction (`new` / `with_config`) — one-time setup
pub struct JsonDecoder {
    /// Frozen output schema.
    schema: SchemaRef,
    /// Decoder configuration.
    config: JsonDecoderConfig,
    /// Pre-computed field index map: field name → column index.
    field_indices: Vec<(String, usize)>,
    /// Cumulative type mismatch count (diagnostics).
    mismatch_count: AtomicU64,
    /// Ring 2 pre-computed: extraction strategy per schema column.
    column_extractions: Vec<ColumnExtraction>,
    /// Ring 2 pre-computed: numeric epoch unit per schema column
    /// (aligned to `schema.fields()`; default [`EpochUnit::Millis`]).
    column_epoch_units: Vec<EpochUnit>,
    /// Ring 2 pre-computed: explode position → schema column index.
    /// `Some` when `json.explode` is configured; each entry maps an
    /// explode position to the schema column index (or `None` if
    /// the explode name doesn't match any schema column).
    explode_col_indices: Option<Vec<Option<usize>>>,
}

#[allow(clippy::missing_fields_in_debug)]
impl std::fmt::Debug for JsonDecoder {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("JsonDecoder")
            .field("schema", &self.schema)
            .field("config", &self.config)
            .field(
                "mismatch_count",
                &self.mismatch_count.load(Ordering::Relaxed),
            )
            .finish()
    }
}

impl JsonDecoder {
    /// Creates a new JSON decoder for the given Arrow schema.
    #[must_use]
    pub fn new(schema: SchemaRef) -> Self {
        Self::with_config(schema, JsonDecoderConfig::default())
    }

    /// Creates a new JSON decoder with custom configuration.
    #[must_use]
    pub fn with_config(schema: SchemaRef, config: JsonDecoderConfig) -> Self {
        let field_indices: Vec<(String, usize)> = schema
            .fields()
            .iter()
            .enumerate()
            .map(|(i, f)| (f.name().clone(), i))
            .collect();

        let column_extractions: Vec<ColumnExtraction> = schema
            .fields()
            .iter()
            .map(|f| {
                let col_name = f.name();
                if let Some(path_segments) = config.json_column_paths.get(col_name.as_str()) {
                    ColumnExtraction::CustomPath {
                        segments: path_segments.clone(),
                    }
                } else {
                    ColumnExtraction::DefaultPath
                }
            })
            .collect();

        let column_epoch_units: Vec<EpochUnit> = schema
            .fields()
            .iter()
            .map(|f| {
                config
                    .numeric_timestamp_units
                    .get(f.name().as_str())
                    .copied()
                    .unwrap_or_default()
            })
            .collect();

        let explode_col_indices = config.json_explode.as_ref().map(|names| {
            names
                .iter()
                .map(|name| {
                    field_indices
                        .iter()
                        .find(|(n, _)| n == name)
                        .map(|(_, idx)| *idx)
                })
                .collect()
        });

        Self {
            schema,
            config,
            field_indices,
            mismatch_count: AtomicU64::new(0),
            column_extractions,
            column_epoch_units,
            explode_col_indices,
        }
    }

    /// Returns the cumulative type mismatch count.
    pub fn mismatch_count(&self) -> u64 {
        self.mismatch_count.load(Ordering::Relaxed)
    }
}

impl FormatDecoder for JsonDecoder {
    fn output_schema(&self) -> SchemaRef {
        self.schema.clone()
    }

    #[allow(clippy::too_many_lines)]
    fn decode_batch(&self, records: &[RawRecord]) -> SchemaResult<RecordBatch> {
        if records.is_empty() {
            return Ok(RecordBatch::new_empty(self.schema.clone()));
        }

        let num_fields = self.schema.fields().len();
        let capacity = records.len();

        // Initialise one builder per schema column.
        let mut builders = create_builders(&self.schema, capacity);

        // Optional _extra JSONB column for CollectExtra strategy.
        let collect_extra = matches!(
            self.config.unknown_fields,
            UnknownFieldStrategy::CollectExtra
        );
        let mut extra_builder = if collect_extra {
            Some(LargeBinaryBuilder::with_capacity(capacity, capacity * 64))
        } else {
            None
        };

        let mut jsonb_encoder = if self.config.nested_as_jsonb {
            Some(JsonbEncoder::new())
        } else {
            None
        };

        // Reusable populated-tracking buffer — avoids heap alloc per record.
        let mut populated = vec![false; num_fields];

        for record in records {
            let value: serde_json::Value = serde_json::from_slice(&record.value)
                .map_err(|e| SchemaError::DecodeError(format!("JSON parse error: {e}")))?;

            // Navigate json.path → default target (borrowed, ZERO alloc).
            // Records lacking the configured path (subscribe acks, server
            // status frames) are skipped silently. Erroring the batch would
            // poison the source buffer since failures don't clear consumed
            // messages.
            let default_target: &serde_json::Value =
                match navigate_path_opt(&value, self.config.json_path.as_deref()) {
                    Some(v) => v,
                    None => continue,
                };

            if let Some(ref col_indices) = self.explode_col_indices {
                // === EXPLODE MODE ===
                let arr = default_target.as_array().ok_or_else(|| {
                    SchemaError::DecodeError("json.explode target must be an array".into())
                })?;
                for element in arr {
                    populated.fill(false);
                    match element {
                        serde_json::Value::Array(items) => {
                            for (pos, col_idx_opt) in col_indices.iter().enumerate() {
                                if let Some(col_idx) = col_idx_opt {
                                    let val = items.get(pos).unwrap_or(&serde_json::Value::Null);
                                    let field = &self.schema.fields()[*col_idx];
                                    append_value(
                                        &mut builders[*col_idx],
                                        field.data_type(),
                                        val,
                                        &self.config,
                                        &self.mismatch_count,
                                        jsonb_encoder.as_mut(),
                                        self.column_epoch_units[*col_idx],
                                    )?;
                                    populated[*col_idx] = true;
                                }
                            }
                        }
                        serde_json::Value::Object(obj) => {
                            for (col_idx, (col_name, _)) in self.field_indices.iter().enumerate() {
                                if let Some(val) = obj.get(col_name.as_str()) {
                                    let field = &self.schema.fields()[col_idx];
                                    append_value(
                                        &mut builders[col_idx],
                                        field.data_type(),
                                        val,
                                        &self.config,
                                        &self.mismatch_count,
                                        jsonb_encoder.as_mut(),
                                        self.column_epoch_units[col_idx],
                                    )?;
                                    populated[col_idx] = true;
                                }
                            }
                        }
                        _ => {
                            return Err(SchemaError::DecodeError(
                                "json.explode array elements must be arrays or objects".into(),
                            ));
                        }
                    }
                    // Append nulls for missing fields.
                    for (col_idx, was_populated) in populated.iter().enumerate() {
                        if !was_populated {
                            append_null(&mut builders[col_idx]);
                        }
                    }
                    // _extra column: append null for each exploded row.
                    if let Some(ref mut eb) = extra_builder {
                        eb.append_null();
                    }
                }
            } else {
                // === NORMAL OBJECT MODE (with per-column path support) ===
                let default_obj = default_target.as_object().ok_or_else(|| {
                    SchemaError::DecodeError("JSON value must be an object".into())
                })?;

                populated.fill(false);

                // Collect unknown fields for CollectExtra.
                let mut extra_fields: Option<serde_json::Map<String, serde_json::Value>> =
                    if collect_extra {
                        Some(serde_json::Map::new())
                    } else {
                        None
                    };

                // Per-column extraction using pre-computed ColumnExtraction.
                for (col_idx, (col_name, _)) in self.field_indices.iter().enumerate() {
                    match &self.column_extractions[col_idx] {
                        ColumnExtraction::DefaultPath => {
                            if let Some(val) = default_obj.get(col_name.as_str()) {
                                populated[col_idx] = true;
                                let field = &self.schema.fields()[col_idx];
                                append_value(
                                    &mut builders[col_idx],
                                    field.data_type(),
                                    val,
                                    &self.config,
                                    &self.mismatch_count,
                                    jsonb_encoder.as_mut(),
                                    self.column_epoch_units[col_idx],
                                )?;
                            }
                        }
                        ColumnExtraction::CustomPath { segments } => {
                            // Navigate custom path from root (ZERO alloc).
                            let extracted = navigate_path(&value, segments);
                            if let Some(val) = extracted {
                                populated[col_idx] = true;
                                let field = &self.schema.fields()[col_idx];
                                append_value(
                                    &mut builders[col_idx],
                                    field.data_type(),
                                    val,
                                    &self.config,
                                    &self.mismatch_count,
                                    jsonb_encoder.as_mut(),
                                    self.column_epoch_units[col_idx],
                                )?;
                            }
                        }
                    }
                }

                // Collect unknown fields (fields in default_obj not in schema).
                if collect_extra {
                    for (key, val) in default_obj {
                        if self.field_index(key).is_none() {
                            match self.config.unknown_fields {
                                UnknownFieldStrategy::CollectExtra => {
                                    if let Some(ref mut extra) = extra_fields {
                                        extra.insert(key.clone(), val.clone());
                                    }
                                }
                                UnknownFieldStrategy::Reject => {
                                    return Err(SchemaError::DecodeError(format!(
                                        "unknown field '{key}' not in schema"
                                    )));
                                }
                                UnknownFieldStrategy::Ignore => {}
                            }
                        }
                    }
                } else if matches!(self.config.unknown_fields, UnknownFieldStrategy::Reject) {
                    for key in default_obj.keys() {
                        if self.field_index(key).is_none() {
                            return Err(SchemaError::DecodeError(format!(
                                "unknown field '{key}' not in schema"
                            )));
                        }
                    }
                }

                // Append nulls for missing fields.
                for (col_idx, was_populated) in populated.iter().enumerate() {
                    if !was_populated {
                        append_null(&mut builders[col_idx]);
                    }
                }

                // Append _extra column.
                if let Some(ref mut eb) = extra_builder {
                    if let Some(ref extra) = extra_fields {
                        if extra.is_empty() {
                            eb.append_null();
                        } else {
                            let mut enc = jsonb_encoder
                                .as_mut()
                                .map_or_else(JsonbEncoder::new, |_| JsonbEncoder::new());
                            let bytes = enc.encode(&serde_json::Value::Object(extra.clone()));
                            eb.append_value(&bytes);
                        }
                    } else {
                        eb.append_null();
                    }
                }
            }
        }

        // Finish all builders into arrays.
        let mut columns: Vec<ArrayRef> = builders.into_iter().map(|mut b| b.finish()).collect();

        // Append _extra column if present.
        let final_schema = if let Some(mut eb) = extra_builder {
            columns.push(Arc::new(eb.finish()));
            let mut fields = self.schema.fields().to_vec();
            fields.push(Arc::new(arrow_schema::Field::new(
                "_extra",
                DataType::LargeBinary,
                true,
            )));
            Arc::new(arrow_schema::Schema::new(fields))
        } else {
            self.schema.clone()
        };

        RecordBatch::try_new(final_schema, columns)
            .map_err(|e| SchemaError::DecodeError(format!("RecordBatch construction: {e}")))
    }

    #[allow(clippy::unnecessary_literal_bound)]
    fn format_name(&self) -> &str {
        "json"
    }
}

impl JsonDecoder {
    /// O(n) field lookup. For schemas with many fields, consider switching
    /// to a `HashMap`; for typical schemas (<50 fields) linear scan is faster.
    fn field_index(&self, name: &str) -> Option<usize> {
        self.field_indices
            .iter()
            .find(|(n, _)| n == name)
            .map(|(_, idx)| *idx)
    }
}

/// Navigate a dot-path through a JSON value tree. Returns `None` if any
/// segment is missing. Zero allocations — pure pointer-chasing.
fn navigate_path<'a>(
    root: &'a serde_json::Value,
    segments: &[String],
) -> Option<&'a serde_json::Value> {
    let mut current = root;
    for segment in segments {
        current = current.get(segment.as_str())?;
    }
    Some(current)
}

/// Like `navigate_path`, but `None` segments mean "no path configured —
/// return the root unchanged."
fn navigate_path_opt<'a>(
    root: &'a serde_json::Value,
    segments: Option<&[String]>,
) -> Option<&'a serde_json::Value> {
    match segments {
        Some(s) => navigate_path(root, s),
        None => Some(root),
    }
}

// ── Builder helpers ────────────────────────────────────────────────

/// Trait-object wrapper so we can store heterogeneous builders in a `Vec`.
trait ColumnBuilder: Send {
    fn finish(&mut self) -> ArrayRef;
    fn append_null_value(&mut self);
    fn as_any_mut(&mut self) -> &mut dyn std::any::Any;
}

macro_rules! impl_column_builder {
    ($builder:ty, $array:ty) => {
        impl ColumnBuilder for $builder {
            fn finish(&mut self) -> ArrayRef {
                Arc::new(<$builder>::finish(self))
            }
            fn append_null_value(&mut self) {
                self.append_null();
            }
            fn as_any_mut(&mut self) -> &mut dyn std::any::Any {
                self
            }
        }
    };
}

impl_column_builder!(BooleanBuilder, arrow_array::BooleanArray);
impl_column_builder!(Int8Builder, arrow_array::Int8Array);
impl_column_builder!(Int16Builder, arrow_array::Int16Array);
impl_column_builder!(Int32Builder, arrow_array::Int32Array);
impl_column_builder!(Int64Builder, arrow_array::Int64Array);
impl_column_builder!(UInt8Builder, arrow_array::UInt8Array);
impl_column_builder!(UInt16Builder, arrow_array::UInt16Array);
impl_column_builder!(UInt32Builder, arrow_array::UInt32Array);
impl_column_builder!(UInt64Builder, arrow_array::UInt64Array);
impl_column_builder!(Float32Builder, arrow_array::Float32Array);
impl_column_builder!(Float64Builder, arrow_array::Float64Array);
impl_column_builder!(StringBuilder, arrow_array::StringArray);
impl_column_builder!(ListBuilder<StringBuilder>, arrow_array::ListArray);
impl_column_builder!(LargeStringBuilder, arrow_array::LargeStringArray);
impl_column_builder!(LargeBinaryBuilder, arrow_array::LargeBinaryArray);
impl_column_builder!(TimestampSecondBuilder, arrow_array::TimestampSecondArray);
impl_column_builder!(
    TimestampMillisecondBuilder,
    arrow_array::TimestampMillisecondArray
);
impl_column_builder!(
    TimestampMicrosecondBuilder,
    arrow_array::TimestampMicrosecondArray
);
impl_column_builder!(
    TimestampNanosecondBuilder,
    arrow_array::TimestampNanosecondArray
);

fn create_builders(schema: &SchemaRef, capacity: usize) -> Vec<Box<dyn ColumnBuilder>> {
    schema
        .fields()
        .iter()
        .map(|f| create_builder(f.data_type(), capacity))
        .collect()
}

fn create_builder(data_type: &DataType, capacity: usize) -> Box<dyn ColumnBuilder> {
    match data_type {
        DataType::Boolean => Box::new(BooleanBuilder::with_capacity(capacity)),
        DataType::Int8 => Box::new(Int8Builder::with_capacity(capacity)),
        DataType::Int16 => Box::new(Int16Builder::with_capacity(capacity)),
        DataType::Int32 => Box::new(Int32Builder::with_capacity(capacity)),
        DataType::Int64 => Box::new(Int64Builder::with_capacity(capacity)),
        DataType::UInt8 => Box::new(UInt8Builder::with_capacity(capacity)),
        DataType::UInt16 => Box::new(UInt16Builder::with_capacity(capacity)),
        DataType::UInt32 => Box::new(UInt32Builder::with_capacity(capacity)),
        DataType::UInt64 => Box::new(UInt64Builder::with_capacity(capacity)),
        DataType::Float32 => Box::new(Float32Builder::with_capacity(capacity)),
        DataType::Float64 => Box::new(Float64Builder::with_capacity(capacity)),
        DataType::LargeUtf8 => Box::new(LargeStringBuilder::with_capacity(capacity, capacity * 32)),
        DataType::LargeBinary => {
            Box::new(LargeBinaryBuilder::with_capacity(capacity, capacity * 64))
        }
        DataType::Timestamp(TimeUnit::Second, tz) => {
            let builder =
                TimestampSecondBuilder::with_capacity(capacity).with_timezone_opt(tz.clone());
            Box::new(builder)
        }
        DataType::Timestamp(TimeUnit::Millisecond, tz) => {
            let builder =
                TimestampMillisecondBuilder::with_capacity(capacity).with_timezone_opt(tz.clone());
            Box::new(builder)
        }
        DataType::Timestamp(TimeUnit::Microsecond, tz) => {
            let builder =
                TimestampMicrosecondBuilder::with_capacity(capacity).with_timezone_opt(tz.clone());
            Box::new(builder)
        }
        DataType::Timestamp(TimeUnit::Nanosecond, tz) => {
            let builder =
                TimestampNanosecondBuilder::with_capacity(capacity).with_timezone_opt(tz.clone());
            Box::new(builder)
        }
        DataType::List(field) if matches!(field.data_type(), DataType::Utf8) => {
            Box::new(ListBuilder::new(StringBuilder::new()))
        }
        // Fallback: serialize as JSON string.
        _ => Box::new(StringBuilder::with_capacity(capacity, capacity * 32)),
    }
}

fn append_null(builder: &mut Box<dyn ColumnBuilder>) {
    builder.append_null_value();
}

/// Append a JSON value to the appropriate builder column.
#[allow(clippy::too_many_arguments, clippy::too_many_lines)]
fn append_value(
    builder: &mut Box<dyn ColumnBuilder>,
    target_type: &DataType,
    value: &serde_json::Value,
    config: &JsonDecoderConfig,
    mismatch_count: &AtomicU64,
    jsonb_encoder: Option<&mut JsonbEncoder>,
    numeric_ts_unit: EpochUnit,
) -> SchemaResult<()> {
    if value.is_null() {
        builder.append_null_value();
        return Ok(());
    }

    match target_type {
        DataType::Boolean => {
            let b = builder
                .as_any_mut()
                .downcast_mut::<BooleanBuilder>()
                .unwrap();
            match extract_bool(value, config) {
                Ok(v) => b.append_value(v),
                Err(e) => handle_mismatch(builder, config, mismatch_count, &e)?,
            }
        }
        DataType::Int8 => {
            let b = builder.as_any_mut().downcast_mut::<Int8Builder>().unwrap();
            match extract_i8(value, config) {
                Ok(v) => b.append_value(v),
                Err(e) => handle_mismatch(builder, config, mismatch_count, &e)?,
            }
        }
        DataType::Int16 => {
            let b = builder.as_any_mut().downcast_mut::<Int16Builder>().unwrap();
            match extract_i16(value, config) {
                Ok(v) => b.append_value(v),
                Err(e) => handle_mismatch(builder, config, mismatch_count, &e)?,
            }
        }
        DataType::Int32 => {
            let b = builder.as_any_mut().downcast_mut::<Int32Builder>().unwrap();
            match extract_i32(value, config) {
                Ok(v) => b.append_value(v),
                Err(e) => handle_mismatch(builder, config, mismatch_count, &e)?,
            }
        }
        DataType::Int64 => {
            let b = builder.as_any_mut().downcast_mut::<Int64Builder>().unwrap();
            match extract_i64(value, config) {
                Ok(v) => b.append_value(v),
                Err(e) => handle_mismatch(builder, config, mismatch_count, &e)?,
            }
        }
        DataType::UInt8 => {
            let b = builder.as_any_mut().downcast_mut::<UInt8Builder>().unwrap();
            match extract_u8(value, config) {
                Ok(v) => b.append_value(v),
                Err(e) => handle_mismatch(builder, config, mismatch_count, &e)?,
            }
        }
        DataType::UInt16 => {
            let b = builder
                .as_any_mut()
                .downcast_mut::<UInt16Builder>()
                .unwrap();
            match extract_u16(value, config) {
                Ok(v) => b.append_value(v),
                Err(e) => handle_mismatch(builder, config, mismatch_count, &e)?,
            }
        }
        DataType::UInt32 => {
            let b = builder
                .as_any_mut()
                .downcast_mut::<UInt32Builder>()
                .unwrap();
            match extract_u32(value, config) {
                Ok(v) => b.append_value(v),
                Err(e) => handle_mismatch(builder, config, mismatch_count, &e)?,
            }
        }
        DataType::UInt64 => {
            let b = builder
                .as_any_mut()
                .downcast_mut::<UInt64Builder>()
                .unwrap();
            match extract_u64(value, config) {
                Ok(v) => b.append_value(v),
                Err(e) => handle_mismatch(builder, config, mismatch_count, &e)?,
            }
        }
        DataType::Float32 => {
            let b = builder
                .as_any_mut()
                .downcast_mut::<Float32Builder>()
                .unwrap();
            match extract_f32(value, config) {
                Ok(v) => b.append_value(v),
                Err(e) => handle_mismatch(builder, config, mismatch_count, &e)?,
            }
        }
        DataType::Float64 => {
            let b = builder
                .as_any_mut()
                .downcast_mut::<Float64Builder>()
                .unwrap();
            match extract_f64(value, config) {
                Ok(v) => b.append_value(v),
                Err(e) => handle_mismatch(builder, config, mismatch_count, &e)?,
            }
        }
        DataType::LargeUtf8 => {
            let b = builder
                .as_any_mut()
                .downcast_mut::<LargeStringBuilder>()
                .unwrap();
            let s = value_to_string(value);
            b.append_value(&s);
        }
        DataType::LargeBinary => {
            let b = builder
                .as_any_mut()
                .downcast_mut::<LargeBinaryBuilder>()
                .unwrap();
            if let Some(enc) = jsonb_encoder {
                let bytes = enc.encode(value);
                b.append_value(&bytes);
            } else {
                // Fallback: serialize as JSON bytes.
                let bytes = serde_json::to_vec(value).unwrap_or_default();
                b.append_value(&bytes);
            }
        }
        DataType::Timestamp(unit, _) => {
            match extract_timestamp(value, config, *unit, numeric_ts_unit) {
                Ok(ts) => append_timestamp(builder, *unit, ts),
                Err(e) => handle_mismatch(builder, config, mismatch_count, &e)?,
            }
        }
        DataType::List(field) if matches!(field.data_type(), DataType::Utf8) => {
            if let Some(items) = value.as_array() {
                let b = builder
                    .as_any_mut()
                    .downcast_mut::<ListBuilder<StringBuilder>>()
                    .unwrap();
                for item in items {
                    if let Some(s) = item.as_str() {
                        b.values().append_value(s);
                    } else if item.is_null() {
                        b.values().append_null();
                    } else {
                        b.values().append_value(value_to_string(item));
                    }
                }
                b.append(true);
            } else {
                // A non-array value for a list column is a type mismatch; honor
                // the Null/Coerce/Reject policy like the scalar arms above.
                handle_mismatch(
                    builder,
                    config,
                    mismatch_count,
                    &format!("expected array, got {}", json_type_name(value)),
                )?;
            }
        }
        // Unsupported types: serialize as JSON string.
        _ => {
            let b = builder
                .as_any_mut()
                .downcast_mut::<StringBuilder>()
                .unwrap();
            let s = value_to_string(value);
            b.append_value(&s);
        }
    }

    Ok(())
}

fn handle_mismatch(
    builder: &mut Box<dyn ColumnBuilder>,
    config: &JsonDecoderConfig,
    mismatch_count: &AtomicU64,
    error_msg: &str,
) -> SchemaResult<()> {
    match config.type_mismatch {
        TypeMismatchStrategy::Null => {
            mismatch_count.fetch_add(1, Ordering::Relaxed);
            builder.append_null_value();
            Ok(())
        }
        TypeMismatchStrategy::Coerce => {
            // Coercion already failed in the extractor — this is a real error.
            Err(SchemaError::DecodeError(format!(
                "type coercion failed: {error_msg}"
            )))
        }
        TypeMismatchStrategy::Reject => Err(SchemaError::DecodeError(format!(
            "type mismatch: {error_msg}"
        ))),
    }
}

// ── Value extractors ───────────────────────────────────────────────

fn extract_bool(value: &serde_json::Value, config: &JsonDecoderConfig) -> Result<bool, String> {
    if let Some(b) = value.as_bool() {
        return Ok(b);
    }
    if matches!(config.type_mismatch, TypeMismatchStrategy::Coerce) {
        if let Some(s) = value.as_str() {
            match s.to_ascii_lowercase().as_str() {
                "true" | "1" | "yes" => return Ok(true),
                "false" | "0" | "no" => return Ok(false),
                _ => {}
            }
        }
        if let Some(n) = value.as_i64() {
            return Ok(n != 0);
        }
    }
    Err(format!("expected boolean, got {}", json_type_name(value)))
}

fn extract_i8(value: &serde_json::Value, config: &JsonDecoderConfig) -> Result<i8, String> {
    if let Some(n) = value.as_i64() {
        if let Ok(v) = i8::try_from(n) {
            return Ok(v);
        }
        return Err(format!("integer {n} out of i8 range"));
    }
    if matches!(config.type_mismatch, TypeMismatchStrategy::Coerce) {
        if let Some(s) = value.as_str() {
            if let Ok(v) = s.parse::<i8>() {
                return Ok(v);
            }
        }
        if let Some(f) = value.as_f64() {
            #[allow(clippy::cast_possible_truncation)]
            let v = f as i8;
            return Ok(v);
        }
    }
    Err(format!("expected i8, got {}", json_type_name(value)))
}

fn extract_i16(value: &serde_json::Value, config: &JsonDecoderConfig) -> Result<i16, String> {
    if let Some(n) = value.as_i64() {
        if let Ok(v) = i16::try_from(n) {
            return Ok(v);
        }
        return Err(format!("integer {n} out of i16 range"));
    }
    if matches!(config.type_mismatch, TypeMismatchStrategy::Coerce) {
        if let Some(s) = value.as_str() {
            if let Ok(v) = s.parse::<i16>() {
                return Ok(v);
            }
        }
        if let Some(f) = value.as_f64() {
            #[allow(clippy::cast_possible_truncation)]
            let v = f as i16;
            return Ok(v);
        }
    }
    Err(format!("expected i16, got {}", json_type_name(value)))
}

fn extract_i32(value: &serde_json::Value, config: &JsonDecoderConfig) -> Result<i32, String> {
    if let Some(n) = value.as_i64() {
        if let Ok(v) = i32::try_from(n) {
            return Ok(v);
        }
        return Err(format!("integer {n} out of i32 range"));
    }
    if matches!(config.type_mismatch, TypeMismatchStrategy::Coerce) {
        if let Some(s) = value.as_str() {
            if let Ok(v) = s.parse::<i32>() {
                return Ok(v);
            }
        }
        if let Some(f) = value.as_f64() {
            #[allow(clippy::cast_possible_truncation)]
            return Ok(f as i32);
        }
    }
    Err(format!("expected i32, got {}", json_type_name(value)))
}

fn extract_i64(value: &serde_json::Value, config: &JsonDecoderConfig) -> Result<i64, String> {
    if let Some(n) = value.as_i64() {
        return Ok(n);
    }
    if let Some(n) = value.as_u64() {
        if let Ok(v) = i64::try_from(n) {
            return Ok(v);
        }
        return Err(format!("u64 {n} out of i64 range"));
    }
    if matches!(config.type_mismatch, TypeMismatchStrategy::Coerce) {
        if let Some(s) = value.as_str() {
            if let Ok(v) = s.parse::<i64>() {
                return Ok(v);
            }
        }
        if let Some(f) = value.as_f64() {
            #[allow(clippy::cast_possible_truncation)]
            return Ok(f as i64);
        }
    }
    Err(format!("expected i64, got {}", json_type_name(value)))
}

fn extract_f32(value: &serde_json::Value, config: &JsonDecoderConfig) -> Result<f32, String> {
    if let Some(f) = value.as_f64() {
        #[allow(clippy::cast_possible_truncation)]
        return Ok(f as f32);
    }
    if let Some(n) = value.as_i64() {
        #[allow(clippy::cast_precision_loss)]
        return Ok(n as f32);
    }
    if matches!(config.type_mismatch, TypeMismatchStrategy::Coerce) {
        if let Some(s) = value.as_str() {
            if let Ok(v) = s.parse::<f32>() {
                return Ok(v);
            }
        }
    }
    Err(format!("expected f32, got {}", json_type_name(value)))
}

fn extract_f64(value: &serde_json::Value, config: &JsonDecoderConfig) -> Result<f64, String> {
    if let Some(f) = value.as_f64() {
        return Ok(f);
    }
    if let Some(n) = value.as_i64() {
        #[allow(clippy::cast_precision_loss)]
        return Ok(n as f64);
    }
    if matches!(config.type_mismatch, TypeMismatchStrategy::Coerce) {
        if let Some(s) = value.as_str() {
            if let Ok(v) = s.parse::<f64>() {
                return Ok(v);
            }
        }
    }
    Err(format!("expected f64, got {}", json_type_name(value)))
}

fn extract_u8(value: &serde_json::Value, config: &JsonDecoderConfig) -> Result<u8, String> {
    if let Some(n) = value.as_u64() {
        if let Ok(v) = u8::try_from(n) {
            return Ok(v);
        }
        return Err(format!("integer {n} out of u8 range"));
    }
    if let Some(n) = value.as_i64() {
        if let Ok(v) = u8::try_from(n) {
            return Ok(v);
        }
        return Err(format!("integer {n} out of u8 range"));
    }
    if matches!(config.type_mismatch, TypeMismatchStrategy::Coerce) {
        if let Some(s) = value.as_str() {
            if let Ok(v) = s.parse::<u8>() {
                return Ok(v);
            }
        }
    }
    Err(format!("expected u8, got {}", json_type_name(value)))
}

fn extract_u16(value: &serde_json::Value, config: &JsonDecoderConfig) -> Result<u16, String> {
    if let Some(n) = value.as_u64() {
        if let Ok(v) = u16::try_from(n) {
            return Ok(v);
        }
        return Err(format!("integer {n} out of u16 range"));
    }
    if let Some(n) = value.as_i64() {
        if let Ok(v) = u16::try_from(n) {
            return Ok(v);
        }
        return Err(format!("integer {n} out of u16 range"));
    }
    if matches!(config.type_mismatch, TypeMismatchStrategy::Coerce) {
        if let Some(s) = value.as_str() {
            if let Ok(v) = s.parse::<u16>() {
                return Ok(v);
            }
        }
    }
    Err(format!("expected u16, got {}", json_type_name(value)))
}

fn extract_u32(value: &serde_json::Value, config: &JsonDecoderConfig) -> Result<u32, String> {
    if let Some(n) = value.as_u64() {
        if let Ok(v) = u32::try_from(n) {
            return Ok(v);
        }
        return Err(format!("integer {n} out of u32 range"));
    }
    if let Some(n) = value.as_i64() {
        if let Ok(v) = u32::try_from(n) {
            return Ok(v);
        }
        return Err(format!("integer {n} out of u32 range"));
    }
    if matches!(config.type_mismatch, TypeMismatchStrategy::Coerce) {
        if let Some(s) = value.as_str() {
            if let Ok(v) = s.parse::<u32>() {
                return Ok(v);
            }
        }
    }
    Err(format!("expected u32, got {}", json_type_name(value)))
}

fn extract_u64(value: &serde_json::Value, config: &JsonDecoderConfig) -> Result<u64, String> {
    if let Some(n) = value.as_u64() {
        return Ok(n);
    }
    if let Some(n) = value.as_i64() {
        if let Ok(v) = u64::try_from(n) {
            return Ok(v);
        }
        return Err(format!("integer {n} out of u64 range"));
    }
    if matches!(config.type_mismatch, TypeMismatchStrategy::Coerce) {
        if let Some(s) = value.as_str() {
            if let Ok(v) = s.parse::<u64>() {
                return Ok(v);
            }
        }
    }
    Err(format!("expected u64, got {}", json_type_name(value)))
}

/// Extracts a timestamp value as an i64 in the specified [`TimeUnit`].
///
/// For numeric JSON values, scales from `from` (the column's configured
/// [`EpochUnit`], default millis) to the target Arrow `TimeUnit`. For
/// string values, tries the configured timestamp format patterns (the
/// `from` unit does not apply — strings carry their own resolution).
fn extract_timestamp(
    value: &serde_json::Value,
    config: &JsonDecoderConfig,
    unit: TimeUnit,
    from: EpochUnit,
) -> Result<i64, String> {
    // Numeric values: scale from the configured epoch unit.
    if let Some(n) = value.as_i64() {
        return checked_epoch_to_unit(n, from, unit);
    }
    if let Some(f) = value.as_f64() {
        // 2^63 == i64::MAX + 1; exclusive upper bound for a lossless f64->i64.
        const POW2_63: f64 = 9_223_372_036_854_775_808.0;
        // `f as i64` saturates (NaN -> 0, +-huge -> i64::MIN/MAX), silently
        // turning a garbage timestamp into a valid-looking one that poisons
        // event-time/watermarks. Round to the nearest integer in `from`
        // units and reject out-of-range/non-finite so the configured
        // type-mismatch strategy applies, like any other bad value.
        let rounded = f.round();
        if !(-POW2_63..POW2_63).contains(&rounded) {
            return Err(format!("timestamp {f} out of i64 range"));
        }
        #[allow(clippy::cast_possible_truncation)] // range-checked; already integral
        let v = rounded as i64;
        return checked_epoch_to_unit(v, from, unit);
    }

    // String values: try configured timestamp formats.
    if let Some(s) = value.as_str() {
        for fmt in &config.timestamp_formats {
            if fmt == "iso8601" {
                if let Ok(nanos) = arrow_cast::parse::string_to_timestamp_nanos(s) {
                    return Ok(nanos_to_unit(nanos, unit));
                }
                continue;
            }
            if let Ok(ndt) = chrono::NaiveDateTime::parse_from_str(s, fmt) {
                let nanos = ndt.and_utc().timestamp_nanos_opt().unwrap_or(0);
                return Ok(nanos_to_unit(nanos, unit));
            }
        }
        return Err(format!("cannot parse timestamp from string: {s}"));
    }

    Err(format!("expected timestamp, got {}", json_type_name(value)))
}

/// Scales an integer epoch `value` from `from` units to the target Arrow
/// `TimeUnit`. Up-scaling (e.g. seconds→nanos) is checked and errors
/// rather than wrapping on i64 overflow; down-scaling (e.g. nanos→millis)
/// truncates toward zero, the conventional and expected precision loss.
///
/// `from == EpochUnit::Millis` reproduces the historical
/// `checked_millis_to_unit` behaviour bit-for-bit.
fn checked_epoch_to_unit(value: i64, from: EpochUnit, to: TimeUnit) -> Result<i64, String> {
    let from_ns = from.nanos_per();
    let to_ns = match to {
        TimeUnit::Second => 1_000_000_000,
        TimeUnit::Millisecond => 1_000_000,
        TimeUnit::Microsecond => 1_000,
        TimeUnit::Nanosecond => 1,
    };
    if from_ns >= to_ns {
        // Exact integer factor (both are powers-of-ten multiples of 1ns).
        let factor = from_ns / to_ns;
        value
            .checked_mul(factor)
            .ok_or_else(|| format!("timestamp {value} ({from:?}) out of i64 {to:?} range"))
    } else {
        let factor = to_ns / from_ns;
        Ok(value / factor)
    }
}

/// Converts nanoseconds to the target time unit.
fn nanos_to_unit(nanos: i64, unit: TimeUnit) -> i64 {
    match unit {
        TimeUnit::Second => nanos / 1_000_000_000,
        TimeUnit::Millisecond => nanos / 1_000_000,
        TimeUnit::Microsecond => nanos / 1_000,
        TimeUnit::Nanosecond => nanos,
    }
}

/// Appends a timestamp value to the appropriate builder based on [`TimeUnit`].
fn append_timestamp(builder: &mut Box<dyn ColumnBuilder>, unit: TimeUnit, value: i64) {
    match unit {
        TimeUnit::Second => {
            builder
                .as_any_mut()
                .downcast_mut::<TimestampSecondBuilder>()
                .unwrap()
                .append_value(value);
        }
        TimeUnit::Millisecond => {
            builder
                .as_any_mut()
                .downcast_mut::<TimestampMillisecondBuilder>()
                .unwrap()
                .append_value(value);
        }
        TimeUnit::Microsecond => {
            builder
                .as_any_mut()
                .downcast_mut::<TimestampMicrosecondBuilder>()
                .unwrap()
                .append_value(value);
        }
        TimeUnit::Nanosecond => {
            builder
                .as_any_mut()
                .downcast_mut::<TimestampNanosecondBuilder>()
                .unwrap()
                .append_value(value);
        }
    }
}

fn value_to_string(value: &serde_json::Value) -> String {
    match value {
        serde_json::Value::String(s) => s.clone(),
        other => other.to_string(),
    }
}

fn json_type_name(value: &serde_json::Value) -> &'static str {
    match value {
        serde_json::Value::Null => "null",
        serde_json::Value::Bool(_) => "boolean",
        serde_json::Value::Number(_) => "number",
        serde_json::Value::String(_) => "string",
        serde_json::Value::Array(_) => "array",
        serde_json::Value::Object(_) => "object",
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use arrow_array::cast::AsArray;
    use arrow_schema::{Field, Schema};

    fn make_schema(fields: Vec<(&str, DataType, bool)>) -> SchemaRef {
        Arc::new(Schema::new(
            fields
                .into_iter()
                .map(|(name, dt, nullable)| Field::new(name, dt, nullable))
                .collect::<Vec<_>>(),
        ))
    }

    fn json_record(json: &str) -> RawRecord {
        RawRecord::new(json.as_bytes().to_vec())
    }

    // ── Basic decode tests ────────────────────────────────────

    #[test]
    fn test_decode_empty_batch() {
        let schema = make_schema(vec![("id", DataType::Int64, false)]);
        let decoder = JsonDecoder::new(schema.clone());
        let batch = decoder.decode_batch(&[]).unwrap();
        assert_eq!(batch.num_rows(), 0);
        assert_eq!(batch.schema(), schema);
    }

    #[test]
    fn test_decode_single_record() {
        let schema = make_schema(vec![
            ("id", DataType::Int64, false),
            ("name", DataType::Utf8, true),
        ]);
        let decoder = JsonDecoder::new(schema);
        let records = vec![json_record(r#"{"id": 42, "name": "Alice"}"#)];
        let batch = decoder.decode_batch(&records).unwrap();

        assert_eq!(batch.num_rows(), 1);
        assert_eq!(
            batch
                .column(0)
                .as_primitive::<arrow_array::types::Int64Type>()
                .value(0),
            42
        );
        assert_eq!(batch.column(1).as_string::<i32>().value(0), "Alice");
    }

    #[test]
    fn test_decode_multiple_records() {
        let schema = make_schema(vec![
            ("x", DataType::Int64, false),
            ("y", DataType::Float64, false),
        ]);
        let decoder = JsonDecoder::new(schema);
        let records = vec![
            json_record(r#"{"x": 1, "y": 1.5}"#),
            json_record(r#"{"x": 2, "y": 2.5}"#),
            json_record(r#"{"x": 3, "y": 3.5}"#),
        ];
        let batch = decoder.decode_batch(&records).unwrap();

        assert_eq!(batch.num_rows(), 3);
        let x_col = batch
            .column(0)
            .as_primitive::<arrow_array::types::Int64Type>();
        assert_eq!(x_col.value(0), 1);
        assert_eq!(x_col.value(1), 2);
        assert_eq!(x_col.value(2), 3);
    }

    #[test]
    fn test_decode_all_types() {
        let schema = make_schema(vec![
            ("bool_col", DataType::Boolean, false),
            ("int_col", DataType::Int64, false),
            ("float_col", DataType::Float64, false),
            ("str_col", DataType::Utf8, false),
        ]);
        let decoder = JsonDecoder::new(schema);
        let records = vec![json_record(
            r#"{"bool_col": true, "int_col": 42, "float_col": 3.14, "str_col": "hello"}"#,
        )];
        let batch = decoder.decode_batch(&records).unwrap();

        assert_eq!(batch.num_rows(), 1);
        assert!(batch.column(0).as_boolean().value(0));
        assert_eq!(
            batch
                .column(1)
                .as_primitive::<arrow_array::types::Int64Type>()
                .value(0),
            42
        );
        let f = batch
            .column(2)
            .as_primitive::<arrow_array::types::Float64Type>()
            .value(0);
        assert!((f - 3.14).abs() < f64::EPSILON);
        assert_eq!(batch.column(3).as_string::<i32>().value(0), "hello");
    }

    // ── Null handling ─────────────────────────────────────────

    #[test]
    fn test_decode_null_values() {
        let schema = make_schema(vec![
            ("a", DataType::Int64, true),
            ("b", DataType::Utf8, true),
        ]);
        let decoder = JsonDecoder::new(schema);
        let records = vec![json_record(r#"{"a": null, "b": null}"#)];
        let batch = decoder.decode_batch(&records).unwrap();

        assert!(batch.column(0).is_null(0));
        assert!(batch.column(1).is_null(0));
    }

    #[test]
    fn test_decode_missing_field_becomes_null() {
        let schema = make_schema(vec![
            ("a", DataType::Int64, true),
            ("b", DataType::Utf8, true),
        ]);
        let decoder = JsonDecoder::new(schema);
        let records = vec![json_record(r#"{"a": 1}"#)]; // "b" missing
        let batch = decoder.decode_batch(&records).unwrap();

        assert_eq!(
            batch
                .column(0)
                .as_primitive::<arrow_array::types::Int64Type>()
                .value(0),
            1
        );
        assert!(batch.column(1).is_null(0));
    }

    // ── Type mismatch strategies ──────────────────────────────

    #[test]
    fn test_mismatch_null_strategy() {
        let schema = make_schema(vec![("x", DataType::Int64, true)]);
        let config = JsonDecoderConfig {
            type_mismatch: TypeMismatchStrategy::Null,
            ..Default::default()
        };
        let decoder = JsonDecoder::with_config(schema, config);
        let records = vec![json_record(r#"{"x": "not_a_number"}"#)];
        let batch = decoder.decode_batch(&records).unwrap();

        assert!(batch.column(0).is_null(0));
        assert_eq!(decoder.mismatch_count(), 1);
    }

    #[test]
    fn test_mismatch_coerce_strategy() {
        let schema = make_schema(vec![("x", DataType::Int64, true)]);
        let config = JsonDecoderConfig {
            type_mismatch: TypeMismatchStrategy::Coerce,
            ..Default::default()
        };
        let decoder = JsonDecoder::with_config(schema, config);
        let records = vec![json_record(r#"{"x": "123"}"#)];
        let batch = decoder.decode_batch(&records).unwrap();

        assert_eq!(
            batch
                .column(0)
                .as_primitive::<arrow_array::types::Int64Type>()
                .value(0),
            123
        );
    }

    #[test]
    fn test_mismatch_reject_strategy() {
        let schema = make_schema(vec![("x", DataType::Int64, false)]);
        let config = JsonDecoderConfig {
            type_mismatch: TypeMismatchStrategy::Reject,
            ..Default::default()
        };
        let decoder = JsonDecoder::with_config(schema, config);
        let records = vec![json_record(r#"{"x": "not_a_number"}"#)];
        let result = decoder.decode_batch(&records);

        assert!(result.is_err());
        assert!(result.unwrap_err().to_string().contains("type mismatch"));
    }

    // ── Unknown field strategies ──────────────────────────────

    #[test]
    fn test_unknown_fields_ignore() {
        let schema = make_schema(vec![("a", DataType::Int64, false)]);
        let decoder = JsonDecoder::new(schema);
        let records = vec![json_record(r#"{"a": 1, "unknown": "value"}"#)];
        let batch = decoder.decode_batch(&records).unwrap();

        assert_eq!(batch.num_columns(), 1);
        assert_eq!(batch.num_rows(), 1);
    }

    #[test]
    fn test_unknown_fields_reject() {
        let schema = make_schema(vec![("a", DataType::Int64, false)]);
        let config = JsonDecoderConfig {
            unknown_fields: UnknownFieldStrategy::Reject,
            ..Default::default()
        };
        let decoder = JsonDecoder::with_config(schema, config);
        let records = vec![json_record(r#"{"a": 1, "unknown": "value"}"#)];
        let result = decoder.decode_batch(&records);

        assert!(result.is_err());
        assert!(result.unwrap_err().to_string().contains("unknown field"));
    }

    #[test]
    fn test_unknown_fields_collect_extra() {
        let schema = make_schema(vec![("a", DataType::Int64, false)]);
        let config = JsonDecoderConfig {
            unknown_fields: UnknownFieldStrategy::CollectExtra,
            ..Default::default()
        };
        let decoder = JsonDecoder::with_config(schema, config);
        let records = vec![json_record(r#"{"a": 1, "extra1": "v1", "extra2": 42}"#)];
        let batch = decoder.decode_batch(&records).unwrap();

        // Schema should have an extra `_extra` column.
        assert_eq!(batch.num_columns(), 2);
        assert_eq!(batch.schema().field(1).name(), "_extra");
        assert!(!batch.column(1).is_null(0));
    }

    // ── Timestamp parsing ─────────────────────────────────────

    #[test]
    fn test_decode_timestamp_iso8601() {
        let schema = make_schema(vec![(
            "ts",
            DataType::Timestamp(TimeUnit::Nanosecond, Some("UTC".into())),
            false,
        )]);
        let decoder = JsonDecoder::new(schema);
        let records = vec![json_record(r#"{"ts": "2025-01-15T10:30:00Z"}"#)];
        let batch = decoder.decode_batch(&records).unwrap();

        assert!(!batch.column(0).is_null(0));
    }

    #[test]
    fn test_decode_timestamp_epoch_millis() {
        let schema = make_schema(vec![(
            "ts",
            DataType::Timestamp(TimeUnit::Nanosecond, None),
            false,
        )]);
        let decoder = JsonDecoder::new(schema);
        let records = vec![json_record(r#"{"ts": 1705312200000}"#)];
        let batch = decoder.decode_batch(&records).unwrap();

        let ts_col = batch
            .column(0)
            .as_primitive::<arrow_array::types::TimestampNanosecondType>();
        // 1705312200000 ms * 1_000_000 = nanos
        assert_eq!(ts_col.value(0), 1_705_312_200_000_000_000);
    }

    #[test]
    fn test_decode_out_of_range_float_timestamp_is_rejected() {
        // A garbage float epoch-ms must not silently saturate to i64::MAX
        // and poison event-time/watermarks — it routes through the
        // configured type-mismatch path instead.
        let schema = make_schema(vec![(
            "ts",
            DataType::Timestamp(TimeUnit::Millisecond, None),
            false,
        )]);
        let decoder = JsonDecoder::new(schema);
        let records = vec![json_record(r#"{"ts": 1e30}"#)];
        let result = decoder.decode_batch(&records);

        assert!(result.is_err());
        assert!(result.unwrap_err().to_string().contains("out of i64 range"));
    }

    #[test]
    fn test_decode_timestamp_overflow_on_nanosecond_scaling_is_rejected() {
        // A ms value that fits i64 but overflows when scaled to nanoseconds
        // must error, not wrap into a bogus (watermark-poisoning) timestamp.
        let schema = make_schema(vec![(
            "ts",
            DataType::Timestamp(TimeUnit::Nanosecond, None),
            false,
        )]);
        let decoder = JsonDecoder::new(schema);
        let records = vec![json_record(r#"{"ts": 9999999999999999}"#)];
        let result = decoder.decode_batch(&records);

        assert!(result.is_err());
        assert!(result
            .unwrap_err()
            .to_string()
            .contains("out of i64 Nanosecond range"));
    }

    // ── Nested objects as LargeBinary ─────────────────────────

    #[test]
    fn test_decode_nested_object_as_json_string() {
        let schema = make_schema(vec![("data", DataType::LargeBinary, true)]);
        let decoder = JsonDecoder::new(schema);
        let records = vec![json_record(r#"{"data": {"nested": true}}"#)];
        let batch = decoder.decode_batch(&records).unwrap();

        assert!(!batch.column(0).is_null(0));
    }

    // ── Error cases ───────────────────────────────────────────

    #[test]
    fn test_decode_invalid_json() {
        let schema = make_schema(vec![("a", DataType::Int64, false)]);
        let decoder = JsonDecoder::new(schema);
        let records = vec![RawRecord::new(b"not json".to_vec())];
        let result = decoder.decode_batch(&records);

        assert!(result.is_err());
        assert!(result.unwrap_err().to_string().contains("JSON parse error"));
    }

    #[test]
    fn test_decode_non_object_json() {
        let schema = make_schema(vec![("a", DataType::Int64, false)]);
        let decoder = JsonDecoder::new(schema);
        let records = vec![json_record("[1, 2, 3]")];
        let result = decoder.decode_batch(&records);

        assert!(result.is_err());
        assert!(result
            .unwrap_err()
            .to_string()
            .contains("must be an object"));
    }

    // ── FormatDecoder trait ───────────────────────────────────

    #[test]
    fn test_format_name() {
        let schema = make_schema(vec![("a", DataType::Int64, false)]);
        let decoder = JsonDecoder::new(schema);
        assert_eq!(decoder.format_name(), "json");
    }

    #[test]
    fn test_output_schema() {
        let schema = make_schema(vec![
            ("a", DataType::Int64, false),
            ("b", DataType::Utf8, true),
        ]);
        let decoder = JsonDecoder::new(schema.clone());
        assert_eq!(decoder.output_schema(), schema);
    }

    #[test]
    fn test_decode_one() {
        let schema = make_schema(vec![("x", DataType::Int64, false)]);
        let decoder = JsonDecoder::new(schema);
        let record = json_record(r#"{"x": 99}"#);
        let batch = decoder.decode_one(&record).unwrap();
        assert_eq!(batch.num_rows(), 1);
        assert_eq!(
            batch
                .column(0)
                .as_primitive::<arrow_array::types::Int64Type>()
                .value(0),
            99
        );
    }

    // ── Int/Float numeric coercion ────────────────────────────

    #[test]
    fn test_decode_int_from_float_json() {
        // JSON number 42.0 is parsed as f64 by serde_json. With the default
        // Coerce strategy, it is coerced to Int64 = 42.
        let schema = make_schema(vec![("x", DataType::Int64, true)]);
        let decoder = JsonDecoder::new(schema);
        let records = vec![json_record(r#"{"x": 42.0}"#)];
        let batch = decoder.decode_batch(&records).unwrap();
        assert_eq!(
            batch
                .column(0)
                .as_primitive::<arrow_array::types::Int64Type>()
                .value(0),
            42
        );
    }

    #[test]
    fn test_decode_float_from_int_json() {
        // JSON integer 42 should decode as Float64 = 42.0.
        let schema = make_schema(vec![("x", DataType::Float64, false)]);
        let decoder = JsonDecoder::new(schema);
        let records = vec![json_record(r#"{"x": 42}"#)];
        let batch = decoder.decode_batch(&records).unwrap();
        let val = batch
            .column(0)
            .as_primitive::<arrow_array::types::Float64Type>()
            .value(0);
        assert!((val - 42.0).abs() < f64::EPSILON);
    }

    // ── Coercion tests (string→numeric, int→float, etc.) ────

    #[test]
    fn test_decode_string_number_to_float64() {
        let schema = make_schema(vec![("price", DataType::Float64, false)]);
        let decoder = JsonDecoder::new(schema);
        let records = vec![json_record(r#"{"price": "187.52"}"#)];
        let batch = decoder.decode_batch(&records).unwrap();
        let val = batch
            .column(0)
            .as_primitive::<arrow_array::types::Float64Type>()
            .value(0);
        assert!((val - 187.52).abs() < f64::EPSILON);
    }

    #[test]
    fn test_decode_string_to_int() {
        let schema = make_schema(vec![("qty", DataType::Int32, false)]);
        let decoder = JsonDecoder::new(schema);
        let records = vec![json_record(r#"{"qty": "100"}"#)];
        let batch = decoder.decode_batch(&records).unwrap();
        assert_eq!(
            batch
                .column(0)
                .as_primitive::<arrow_array::types::Int32Type>()
                .value(0),
            100
        );
    }

    #[test]
    fn test_decode_epoch_millis_to_timestamp_millis() {
        let schema = make_schema(vec![(
            "ts",
            DataType::Timestamp(TimeUnit::Millisecond, None),
            false,
        )]);
        let decoder = JsonDecoder::new(schema);
        let records = vec![json_record(r#"{"ts": 1705312200000}"#)];
        let batch = decoder.decode_batch(&records).unwrap();
        let ts_col = batch
            .column(0)
            .as_primitive::<arrow_array::types::TimestampMillisecondType>();
        assert_eq!(ts_col.value(0), 1_705_312_200_000);
    }

    #[test]
    fn test_decode_int_to_float_promotion() {
        let schema = make_schema(vec![("val", DataType::Float64, false)]);
        let decoder = JsonDecoder::new(schema);
        let records = vec![json_record(r#"{"val": 100}"#)];
        let batch = decoder.decode_batch(&records).unwrap();
        let val = batch
            .column(0)
            .as_primitive::<arrow_array::types::Float64Type>()
            .value(0);
        assert!((val - 100.0).abs() < f64::EPSILON);
    }

    #[test]
    fn test_decode_string_boolean() {
        let schema = make_schema(vec![("active", DataType::Boolean, false)]);
        let decoder = JsonDecoder::new(schema);
        let records = vec![json_record(r#"{"active": "true"}"#)];
        let batch = decoder.decode_batch(&records).unwrap();
        assert!(batch.column(0).as_boolean().value(0));
    }

    #[test]
    fn test_coerce_fails_on_unconvertible() {
        // With default Coerce, a string that can't be parsed as Int64 should error.
        let schema = make_schema(vec![("x", DataType::Int64, true)]);
        let decoder = JsonDecoder::new(schema);
        let records = vec![json_record(r#"{"x": "not_a_number"}"#)];
        let result = decoder.decode_batch(&records);
        assert!(result.is_err());
        assert!(result
            .unwrap_err()
            .to_string()
            .contains("type coercion failed"));
    }

    #[test]
    fn test_enforcement_str_parsing() {
        assert_eq!(
            TypeMismatchStrategy::from_enforcement_str("coerce"),
            Some(TypeMismatchStrategy::Coerce)
        );
        assert_eq!(
            TypeMismatchStrategy::from_enforcement_str("STRICT"),
            Some(TypeMismatchStrategy::Reject)
        );
        assert_eq!(
            TypeMismatchStrategy::from_enforcement_str("Permissive"),
            Some(TypeMismatchStrategy::Null)
        );
        assert_eq!(TypeMismatchStrategy::from_enforcement_str("unknown"), None);
    }

    // ── Small integer types ──────────────────────────────────

    #[test]
    fn test_decode_i8_and_u8() {
        let schema = make_schema(vec![
            ("signed", DataType::Int8, false),
            ("unsigned", DataType::UInt8, false),
        ]);
        let decoder = JsonDecoder::new(schema);
        let records = vec![json_record(r#"{"signed": -5, "unsigned": 200}"#)];
        let batch = decoder.decode_batch(&records).unwrap();
        assert_eq!(
            batch
                .column(0)
                .as_primitive::<arrow_array::types::Int8Type>()
                .value(0),
            -5
        );
        assert_eq!(
            batch
                .column(1)
                .as_primitive::<arrow_array::types::UInt8Type>()
                .value(0),
            200
        );
    }

    // ── json.path tests ─────────────────────────────────────────

    #[test]
    fn test_json_path_single() {
        let schema = make_schema(vec![("id", DataType::Int64, false)]);
        let config = JsonDecoderConfig {
            json_path: Some(vec!["data".into()]),
            ..Default::default()
        };
        let decoder = JsonDecoder::with_config(schema, config);
        let records = vec![json_record(r#"{"data":{"id":1}}"#)];
        let batch = decoder.decode_batch(&records).unwrap();
        assert_eq!(batch.num_rows(), 1);
        assert_eq!(
            batch
                .column(0)
                .as_primitive::<arrow_array::types::Int64Type>()
                .value(0),
            1
        );
    }

    #[test]
    fn test_json_path_multi() {
        let schema = make_schema(vec![("id", DataType::Int64, false)]);
        let config = JsonDecoderConfig {
            json_path: Some(vec!["a".into(), "b".into()]),
            ..Default::default()
        };
        let decoder = JsonDecoder::with_config(schema, config);
        let records = vec![json_record(r#"{"a":{"b":{"id":1}}}"#)];
        let batch = decoder.decode_batch(&records).unwrap();
        assert_eq!(
            batch
                .column(0)
                .as_primitive::<arrow_array::types::Int64Type>()
                .value(0),
            1
        );
    }

    /// Records that don't carry the configured json.path are skipped, not
    /// erroring the batch. Real-world: WebSocket sources receive subscribe
    /// acks and control frames mixed with data frames; failing the batch
    /// would poison the source's input buffer.
    #[test]
    fn test_json_path_missing_skips_record() {
        let schema = make_schema(vec![("id", DataType::Int64, false)]);
        let config = JsonDecoderConfig {
            json_path: Some(vec!["data".into()]),
            ..Default::default()
        };
        let decoder = JsonDecoder::with_config(schema, config);
        let records = vec![
            json_record(r#"{"success":true,"op":"subscribe"}"#), // ack — no `data`
            json_record(r#"{"data":{"id":42}}"#),
        ];
        let batch = decoder.decode_batch(&records).unwrap();
        assert_eq!(batch.num_rows(), 1);
        assert_eq!(
            batch
                .column(0)
                .as_primitive::<arrow_array::types::Int64Type>()
                .value(0),
            42
        );
    }

    // ── json.column.* tests ─────────────────────────────────────

    #[test]
    fn test_json_column_custom_path() {
        let schema = make_schema(vec![
            ("p", DataType::Float64, false),
            ("stream_name", DataType::Utf8, true),
        ]);
        let mut col_paths = HashMap::new();
        col_paths.insert("stream_name".into(), vec!["stream".into()]);
        let config = JsonDecoderConfig {
            json_path: Some(vec!["data".into()]),
            json_column_paths: col_paths,
            ..Default::default()
        };
        let decoder = JsonDecoder::with_config(schema, config);
        let records = vec![json_record(
            r#"{"stream":"btcusdt@trade","data":{"p":67523.0}}"#,
        )];
        let batch = decoder.decode_batch(&records).unwrap();
        assert_eq!(batch.num_rows(), 1);
        let price = batch
            .column(0)
            .as_primitive::<arrow_array::types::Float64Type>()
            .value(0);
        assert!((price - 67523.0).abs() < f64::EPSILON);
        assert_eq!(batch.column(1).as_string::<i32>().value(0), "btcusdt@trade");
    }

    #[test]
    fn test_json_column_deep_path() {
        let schema = make_schema(vec![
            ("p", DataType::Float64, false),
            ("ts", DataType::Int64, true),
        ]);
        let mut col_paths = HashMap::new();
        col_paths.insert("ts".into(), vec!["meta".into(), "timestamp".into()]);
        let config = JsonDecoderConfig {
            json_path: Some(vec!["data".into()]),
            json_column_paths: col_paths,
            ..Default::default()
        };
        let decoder = JsonDecoder::with_config(schema, config);
        let records = vec![json_record(
            r#"{"meta":{"timestamp":123456},"data":{"p":99.5}}"#,
        )];
        let batch = decoder.decode_batch(&records).unwrap();
        assert_eq!(batch.num_rows(), 1);
        assert_eq!(
            batch
                .column(1)
                .as_primitive::<arrow_array::types::Int64Type>()
                .value(0),
            123456
        );
    }

    #[test]
    fn test_json_column_missing_returns_null() {
        let schema = make_schema(vec![
            ("id", DataType::Int64, false),
            ("missing_col", DataType::Utf8, true),
        ]);
        let mut col_paths = HashMap::new();
        col_paths.insert("missing_col".into(), vec!["nowhere".into(), "gone".into()]);
        let config = JsonDecoderConfig {
            json_column_paths: col_paths,
            ..Default::default()
        };
        let decoder = JsonDecoder::with_config(schema, config);
        let records = vec![json_record(r#"{"id":42}"#)];
        let batch = decoder.decode_batch(&records).unwrap();
        assert_eq!(
            batch
                .column(0)
                .as_primitive::<arrow_array::types::Int64Type>()
                .value(0),
            42
        );
        assert!(batch.column(1).is_null(0));
    }

    #[test]
    fn test_json_column_path_to_nested_string_array() {
        use arrow_array::Array;
        use arrow_schema::Field;
        let schema = make_schema(vec![(
            "tags",
            DataType::List(Arc::new(Field::new("item", DataType::Utf8, true))),
            true,
        )]);
        let mut col_paths = HashMap::new();
        col_paths.insert("tags".into(), vec!["record".into(), "tags".into()]);
        let config = JsonDecoderConfig {
            json_column_paths: col_paths,
            ..Default::default()
        };
        let decoder = JsonDecoder::with_config(schema, config);
        let records = vec![json_record(r#"{"record":{"tags":["en","es"]}}"#)];
        let batch = decoder.decode_batch(&records).unwrap();

        let list = batch
            .column(0)
            .as_any()
            .downcast_ref::<arrow_array::ListArray>()
            .expect("List column");
        let vals = list.value(0);
        let strs = vals.as_string::<i32>();
        assert_eq!(strs.len(), 2);
        assert_eq!(strs.value(0), "en");
        assert_eq!(strs.value(1), "es");
    }

    #[test]
    fn test_list_column_non_array_honors_reject_strategy() {
        use arrow_schema::Field;
        let schema = make_schema(vec![(
            "tags",
            DataType::List(Arc::new(Field::new("item", DataType::Utf8, true))),
            true,
        )]);
        let config = JsonDecoderConfig {
            type_mismatch: TypeMismatchStrategy::Reject,
            ..Default::default()
        };
        let decoder = JsonDecoder::with_config(schema, config);
        // A scalar where the list column expects an array must be rejected, not
        // silently coerced to NULL.
        let records = vec![json_record(r#"{"tags": "en"}"#)];
        let result = decoder.decode_batch(&records);

        assert!(result.is_err());
        assert!(result.unwrap_err().to_string().contains("type mismatch"));
    }

    // ── json.explode tests ──────────────────────────────────────

    #[test]
    fn test_json_explode_arrays() {
        let schema = make_schema(vec![
            ("price", DataType::Utf8, true),
            ("qty", DataType::Utf8, true),
        ]);
        let config = JsonDecoderConfig {
            json_explode: Some(vec!["price".into(), "qty".into()]),
            ..Default::default()
        };
        let decoder = JsonDecoder::with_config(schema, config);
        let records = vec![json_record(r#"[["67523","1.5"],["67522","0.8"]]"#)];
        let batch = decoder.decode_batch(&records).unwrap();
        assert_eq!(batch.num_rows(), 2);
        assert_eq!(batch.column(0).as_string::<i32>().value(0), "67523");
        assert_eq!(batch.column(0).as_string::<i32>().value(1), "67522");
        assert_eq!(batch.column(1).as_string::<i32>().value(0), "1.5");
        assert_eq!(batch.column(1).as_string::<i32>().value(1), "0.8");
    }

    #[test]
    fn test_json_explode_objects() {
        let schema = make_schema(vec![
            ("id", DataType::Int64, true),
            ("name", DataType::Utf8, true),
        ]);
        let config = JsonDecoderConfig {
            json_explode: Some(vec!["id".into(), "name".into()]),
            ..Default::default()
        };
        let decoder = JsonDecoder::with_config(schema, config);
        let records = vec![json_record(
            r#"[{"id":1,"name":"Alice"},{"id":2,"name":"Bob"}]"#,
        )];
        let batch = decoder.decode_batch(&records).unwrap();
        assert_eq!(batch.num_rows(), 2);
        assert_eq!(
            batch
                .column(0)
                .as_primitive::<arrow_array::types::Int64Type>()
                .value(0),
            1
        );
        assert_eq!(batch.column(1).as_string::<i32>().value(1), "Bob");
    }

    // ── Combined json.path + json.explode ───────────────────────

    #[test]
    fn test_json_path_plus_explode() {
        let schema = make_schema(vec![
            ("price", DataType::Utf8, true),
            ("qty", DataType::Utf8, true),
        ]);
        let config = JsonDecoderConfig {
            json_path: Some(vec!["bids".into()]),
            json_explode: Some(vec!["price".into(), "qty".into()]),
            ..Default::default()
        };
        let decoder = JsonDecoder::with_config(schema, config);
        let records = vec![json_record(r#"{"bids":[["67523","1.5"],["67522","0.8"]]}"#)];
        let batch = decoder.decode_batch(&records).unwrap();
        assert_eq!(batch.num_rows(), 2);
        assert_eq!(batch.column(0).as_string::<i32>().value(0), "67523");
        assert_eq!(batch.column(1).as_string::<i32>().value(1), "0.8");
    }

    // ── from_connector_config tests ─────────────────────────────

    #[test]
    fn test_from_connector_config() {
        let mut config = crate::config::ConnectorConfig::new("websocket");
        config.set("json.path", "data.trade");
        config.set("json.column.stream_name", "stream");
        config.set("json.column.ts", "meta.timestamp");
        config.set("json.explode", "price, qty");
        config.set("schema.enforcement", "strict");
        config.set("nested.as.jsonb", "true");

        let cfg = JsonDecoderConfig::from_connector_config(&config);
        assert_eq!(cfg.json_path, Some(vec!["data".into(), "trade".into()]));
        assert_eq!(
            cfg.json_column_paths.get("stream_name"),
            Some(&vec!["stream".into()])
        );
        assert_eq!(
            cfg.json_column_paths.get("ts"),
            Some(&vec!["meta".into(), "timestamp".into()])
        );
        assert_eq!(cfg.json_explode, Some(vec!["price".into(), "qty".into()]));
        assert_eq!(cfg.type_mismatch, TypeMismatchStrategy::Reject);
        assert!(cfg.nested_as_jsonb);
    }

    #[test]
    fn test_default_config_unchanged() {
        let schema = make_schema(vec![
            ("a", DataType::Int64, false),
            ("b", DataType::Utf8, true),
        ]);
        let decoder = JsonDecoder::new(schema);
        let records = vec![
            json_record(r#"{"a": 1, "b": "hello"}"#),
            json_record(r#"{"a": 2, "b": "world"}"#),
        ];
        let batch = decoder.decode_batch(&records).unwrap();
        assert_eq!(batch.num_rows(), 2);
        assert_eq!(
            batch
                .column(0)
                .as_primitive::<arrow_array::types::Int64Type>()
                .value(0),
            1
        );
        assert_eq!(batch.column(1).as_string::<i32>().value(1), "world");
    }

    #[test]
    fn test_unknown_fields_with_path() {
        let schema = make_schema(vec![("a", DataType::Int64, false)]);
        let config = JsonDecoderConfig {
            json_path: Some(vec!["data".into()]),
            unknown_fields: UnknownFieldStrategy::CollectExtra,
            ..Default::default()
        };
        let decoder = JsonDecoder::with_config(schema, config);
        let records = vec![json_record(r#"{"data":{"a":1,"extra":"value"}}"#)];
        let batch = decoder.decode_batch(&records).unwrap();
        assert_eq!(batch.num_columns(), 2); // a + _extra
        assert_eq!(batch.schema().field(1).name(), "_extra");
        assert!(!batch.column(1).is_null(0));
    }

    // ── Per-column numeric epoch unit ─────────────────────────

    #[test]
    fn from_connector_config_parses_epoch_unit_without_polluting_paths() {
        use crate::config::ConnectorConfig;
        let mut props = std::collections::HashMap::new();
        props.insert("json.column.evt".to_string(), "time_us".to_string());
        props.insert(
            "json.column.evt.epoch_unit".to_string(),
            "micros".to_string(),
        );
        let cc = ConnectorConfig::with_properties("websocket", props);
        let cfg = JsonDecoderConfig::from_connector_config(&cc);
        assert_eq!(
            cfg.numeric_timestamp_units.get("evt"),
            Some(&EpochUnit::Micros)
        );
        // The `.epoch_unit` key must NOT become a phantom path column.
        assert!(cfg.json_column_paths.contains_key("evt"));
        assert!(!cfg.json_column_paths.contains_key("evt.epoch_unit"));
    }

    #[test]
    fn numeric_micros_decodes_into_millis_timestamp_column() {
        // End-to-end: a numeric microsecond field (Jetstream time_us shape)
        // mapped to a Timestamp(ms) column lands at a wall-clock-plausible
        // millisecond value instead of being misread as epoch-millis.
        let schema = make_schema(vec![(
            "ts",
            DataType::Timestamp(TimeUnit::Millisecond, None),
            false,
        )]);
        let mut units = HashMap::new();
        units.insert("ts".to_string(), EpochUnit::Micros);
        let config = JsonDecoderConfig {
            numeric_timestamp_units: units,
            ..Default::default()
        };
        let decoder = JsonDecoder::with_config(schema, config);
        // 1_779_019_200_123_456 µs == 1_779_019_200_123 ms (~2026-05-17).
        let records = vec![json_record(r#"{"ts": 1779019200123456}"#)];
        let batch = decoder.decode_batch(&records).unwrap();
        let col = batch
            .column(0)
            .as_primitive::<arrow_array::types::TimestampMillisecondType>();
        assert_eq!(col.value(0), 1_779_019_200_123);

        // Without the override the same number is (legacy) read as millis.
        let schema2 = make_schema(vec![(
            "ts",
            DataType::Timestamp(TimeUnit::Millisecond, None),
            false,
        )]);
        let d2 = JsonDecoder::new(schema2);
        let b2 = d2.decode_batch(&records).unwrap();
        let c2 = b2
            .column(0)
            .as_primitive::<arrow_array::types::TimestampMillisecondType>();
        assert_eq!(c2.value(0), 1_779_019_200_123_456);
    }
}