skardi 0.6.0

//! JSON → Arrow conversion against a table's fixed schema.
//!
//! Conversion never infers from data: every field comes from a source-pack
//! [`FieldMapping`], so upstream additive changes are ignored and upstream
//! *breaking* changes fail conversion with the column, page, row, expected
//! type, and the found JSON *kind* (never the value, which may be
//! sensitive).

// The row-shape DECLARATIONS moved to the pack so a syncer can read a
// provider's rows without linking Arrow. What stays here is the
// conversion, which is the only part that wants a `RecordBatch`.
pub use skardi_source_pack::schema::{ColumnSpec, FieldMapping, FieldType};

use std::sync::Arc;

use arrow::array::{
    ArrayRef, BooleanArray, Float64Array, Int64Array, ListBuilder, RecordBatch, StringArray,
    StringBuilder, TimestampMillisecondArray, UInt64Array,
};
use arrow::datatypes::{DataType, Field, Schema, SchemaRef, TimeUnit};
use chrono::DateTime;
use serde_json::Value;

use super::error::OpenConnectorError;
use super::row_path::RowPath;

/// The Arrow data type a declared [`FieldType`] maps to.
///
/// A free function rather than a method, because `FieldType` now belongs
/// to `skardi-source-pack` and Rust forbids an inherent `impl` on a type
/// from another crate. The restriction states the design correctly: the
/// Arrow mapping is not a property of the declaration, it is something
/// this crate does to it — and the syncer that reads the same declaration
/// never performs it.
///
/// # Examples
///
/// ```
/// use skardi::sources::providers::open_connector::json_to_arrow::arrow_type;
/// use skardi_source_pack::schema::FieldType;
/// use arrow::datatypes::{DataType, TimeUnit};
///
/// assert_eq!(arrow_type(&FieldType::Int64), DataType::Int64);
/// // The declared types name the WIRE encoding; Arrow has one timestamp type
/// // for all of them. That many-to-one is the reason this mapping is the
/// // engine's to make and not a property of the declaration: a pack says how
/// // the provider spells a timestamp, not how it is stored.
/// let millis = DataType::Timestamp(TimeUnit::Millisecond, Some("UTC".into()));
/// assert_eq!(arrow_type(&FieldType::TimestampMillisUtc), millis);
/// assert_eq!(arrow_type(&FieldType::TimestampSecondsStringUtc), millis);
/// ```
///
/// ```
/// # use skardi::sources::providers::open_connector::json_to_arrow::arrow_type;
/// # use skardi_source_pack::schema::FieldType;
/// # use arrow::datatypes::DataType;
/// // `Json` is carried as text: the engine does not parse a provider's nested
/// // object into a struct type it would then have to keep in sync.
/// assert_eq!(arrow_type(&FieldType::Json), DataType::Utf8);
/// ```
pub fn arrow_type(field_type: &FieldType) -> DataType {
    match field_type {
        FieldType::Boolean => DataType::Boolean,
        FieldType::Int64 => DataType::Int64,
        FieldType::UInt64 => DataType::UInt64,
        FieldType::Float64 => DataType::Float64,
        FieldType::Utf8 | FieldType::Json => DataType::Utf8,
        FieldType::TimestampMillisUtc
        | FieldType::TimestampSecondsUtc
        | FieldType::TimestampMillisStringUtc
        | FieldType::TimestampSecondsStringUtc
        | FieldType::TimestampSecondsFractionalStringUtc => {
            DataType::Timestamp(TimeUnit::Millisecond, Some("UTC".into()))
        }
        FieldType::Utf8List | FieldType::Utf8ListFromObjectKey(_) => {
            DataType::List(Arc::new(Field::new("item", DataType::Utf8, true)))
        }
    }
}

/// A column spec with its path pre-parsed.
#[derive(Debug)]
struct CompiledField {
    spec: ColumnSpec,
    path: RowPath,
}

/// Converts rows against one fixed schema. Build once per table; the Arrow
/// schema it produces never changes for the converter's lifetime.
#[derive(Debug)]
pub struct RowConverter {
    fields: Vec<CompiledField>,
    schema: SchemaRef,
}

impl RowConverter {
    /// Compile a converter from static field mappings.
    ///
    /// # Errors
    /// Returns [`OpenConnectorError::InvalidRowPath`] if any mapping path is
    /// not a dotted object-key path.
    pub fn new(mappings: &[FieldMapping]) -> Result<Self, OpenConnectorError> {
        Self::from_columns(mappings.iter().map(ColumnSpec::from).collect())
    }

    /// Compile a converter from owned column specs (the raw-scan path).
    pub fn from_columns(columns: Vec<ColumnSpec>) -> Result<Self, OpenConnectorError> {
        let mut fields = Vec::with_capacity(columns.len());
        let mut arrow_fields = Vec::with_capacity(columns.len());
        for spec in columns {
            // Column paths are row-relative (`user.login`); the `$.` marker
            // belongs to whole-document row paths only. Reject it rather
            // than strip: `$.{path}` would otherwise parse into a literal
            // `"$"` first segment and silently read the wrong key —
            // all-NULL columns instead of a loud error.
            if spec.path.starts_with("$.") {
                return Err(OpenConnectorError::InvalidRowPath {
                    path: spec.path.clone(),
                    reason: format!(
                        "column paths are row-relative object keys \
                         (write '{}', not '{}')",
                        &spec.path[2..],
                        spec.path
                    ),
                });
            }
            let path = RowPath::parse(&format!("$.{}", spec.path))?;
            arrow_fields.push(Field::new(
                &spec.name,
                arrow_type(&spec.field_type),
                spec.nullable,
            ));
            fields.push(CompiledField { spec, path });
        }
        Ok(Self {
            fields,
            schema: Arc::new(Schema::new(arrow_fields)),
        })
    }

    /// The fixed Arrow schema.
    pub fn schema(&self) -> &SchemaRef {
        &self.schema
    }

    /// Convert one page of row objects to a RecordBatch. `page` is 1-based
    /// and only used for error context. An empty page yields an empty batch
    /// with the fixed schema.
    pub fn convert(&self, rows: &[Value], page: usize) -> Result<RecordBatch, OpenConnectorError> {
        let mut columns: Vec<ArrayRef> = Vec::with_capacity(self.fields.len());
        for field in &self.fields {
            columns.push(self.convert_column(field, rows, page)?);
        }
        RecordBatch::try_new(Arc::clone(&self.schema), columns).map_err(|e| {
            OpenConnectorError::ConversionFailed {
                path: "$".to_string(),
                column: "<batch>".to_string(),
                page,
                row: 0,
                expected: "a batch matching the fixed schema".to_string(),
                found: e.to_string(),
            }
        })
    }

    fn convert_column(
        &self,
        field: &CompiledField,
        rows: &[Value],
        page: usize,
    ) -> Result<ArrayRef, OpenConnectorError> {
        let spec = &field.spec;
        let mut cells: Vec<Option<&Value>> = Vec::with_capacity(rows.len());
        for (row_index, row) in rows.iter().enumerate() {
            match field.path.extract(row, page) {
                Ok(value) => cells.push(Some(value)),
                // Absence may become null for a nullable column: either the
                // key is genuinely missing, or a parent on the path is JSON
                // null — providers null out whole objects routinely (GitHub's
                // `commit.author: null`, `issue.user: null`), and a null
                // parent means every leaf under it is absent. A
                // present-but-wrong-shape parent (e.g. `user` changed from
                // object to string) is an upstream *breaking* change and
                // must fail, per this module's contract.
                Err(OpenConnectorError::RowPathNotFound { .. }) if spec.nullable => {
                    cells.push(None)
                }
                Err(OpenConnectorError::RowPathNotFound { .. }) => {
                    return Err(self.failure(field, page, row_index, "missing key"));
                }
                Err(OpenConnectorError::RowPathNotObject { ref found, .. })
                    if found == "null" && spec.nullable =>
                {
                    cells.push(None)
                }
                Err(OpenConnectorError::RowPathNotObject { ref found, .. }) => {
                    return Err(self.failure(field, page, row_index, found));
                }
                Err(e) => return Err(e),
            }
        }

        let fail = |index: usize, found: &str| self.failure(field, page, index, found);
        match spec.field_type {
            FieldType::Boolean => Ok(Arc::new(BooleanArray::from(collect_cells(
                &cells,
                spec,
                |v| v.as_bool(),
                fail,
            )?))),
            FieldType::Int64 => Ok(Arc::new(Int64Array::from(collect_cells(
                &cells,
                spec,
                |v| v.as_i64(),
                fail,
            )?))),
            FieldType::UInt64 => Ok(Arc::new(UInt64Array::from(collect_cells(
                &cells,
                spec,
                |v| v.as_u64(),
                fail,
            )?))),
            FieldType::Float64 => Ok(Arc::new(Float64Array::from(collect_cells(
                &cells,
                spec,
                |v| v.as_f64(),
                fail,
            )?))),
            FieldType::Utf8 => Ok(Arc::new(StringArray::from(collect_cells(
                &cells,
                spec,
                |v| v.as_str().map(str::to_string),
                fail,
            )?))),
            // Opaque values serialize to canonical JSON text, but a present
            // JSON null is SQL NULL like everywhere else — never the 4-char
            // string "null", which would break `IS NULL` and match
            // `= 'null'`. Routing through collect_cells also makes a null in
            // a non-nullable Json column a targeted per-column failure.
            FieldType::Json => Ok(Arc::new(StringArray::from(collect_cells(
                &cells,
                spec,
                |v| Some(v.to_string()),
                fail,
            )?))),
            FieldType::TimestampMillisUtc => Ok(Arc::new(
                TimestampMillisecondArray::from(collect_cells(
                    &cells,
                    spec,
                    parse_timestamp,
                    fail,
                )?)
                .with_timezone("UTC"),
            )),
            FieldType::TimestampSecondsUtc => Ok(Arc::new(
                TimestampMillisecondArray::from(collect_cells_described(
                    &cells,
                    spec,
                    |v| {
                        let seconds = v.as_i64().ok_or_else(|| json_kind(v).to_string())?;
                        seconds.checked_mul(1000).ok_or_else(|| {
                            "an epoch second out of range for millisecond timestamps".to_string()
                        })
                    },
                    fail,
                )?)
                .with_timezone("UTC"),
            )),
            FieldType::TimestampMillisStringUtc => Ok(Arc::new(
                TimestampMillisecondArray::from(collect_cells_described(
                    &cells,
                    spec,
                    |v| parse_epoch_digit_string(v, 1),
                    fail,
                )?)
                .with_timezone("UTC"),
            )),
            FieldType::TimestampSecondsStringUtc => Ok(Arc::new(
                TimestampMillisecondArray::from(collect_cells_described(
                    &cells,
                    spec,
                    |v| parse_epoch_digit_string(v, 1000),
                    fail,
                )?)
                .with_timezone("UTC"),
            )),
            FieldType::TimestampSecondsFractionalStringUtc => Ok(Arc::new(
                TimestampMillisecondArray::from(collect_cells_described(
                    &cells,
                    spec,
                    parse_epoch_fractional_seconds_string,
                    fail,
                )?)
                .with_timezone("UTC"),
            )),
            FieldType::Utf8List => self.convert_string_list(field, &cells, page, None),
            FieldType::Utf8ListFromObjectKey(key) => {
                self.convert_string_list(field, &cells, page, Some(key))
            }
        }
    }

    /// Convert a list column. With `pluck: None` elements must be strings;
    /// with `pluck: Some(key)` elements must be objects carrying a string
    /// under `key` (the flattened value).
    fn convert_string_list(
        &self,
        field: &CompiledField,
        cells: &[Option<&Value>],
        page: usize,
        pluck: Option<&'static str>,
    ) -> Result<ArrayRef, OpenConnectorError> {
        let mut builder = ListBuilder::new(StringBuilder::new());
        for (row_index, cell) in cells.iter().enumerate() {
            let Some(value) = cell else {
                builder.append(false);
                continue;
            };
            if value.is_null() {
                if field.spec.nullable {
                    builder.append(false);
                    continue;
                }
                return Err(self.failure(field, page, row_index, "null"));
            }
            let items = value
                .as_array()
                .ok_or_else(|| self.failure(field, page, row_index, json_kind(value)))?;
            for item in items {
                // A JSON null wherever an item's text would come from is an
                // Arrow null item — the list's item field is declared
                // nullable, and a provider null is data, not shape drift. A
                // *missing* pluck key stays a failure: unlike columns, items
                // carry no pack-declared nullability that authorizes reading
                // absence as null, so a dropped upstream key fails loudly
                // instead of silently yielding all-null lists.
                let text = match pluck {
                    None => match item {
                        Value::Null => None,
                        Value::String(text) => Some(text.as_str()),
                        other => {
                            return Err(self.failure(
                                field,
                                page,
                                row_index,
                                &format!("{} array element", json_kind(other)),
                            ));
                        }
                    },
                    Some(key) => match item {
                        Value::Null => None,
                        Value::Object(object) => match object.get(key) {
                            Some(Value::Null) => None,
                            Some(Value::String(text)) => Some(text.as_str()),
                            Some(other) => {
                                return Err(self.failure(
                                    field,
                                    page,
                                    row_index,
                                    &format!("array element whose '{key}' is {}", json_kind(other)),
                                ));
                            }
                            None => {
                                return Err(self.failure(
                                    field,
                                    page,
                                    row_index,
                                    &format!("array element without key '{key}'"),
                                ));
                            }
                        },
                        other => {
                            return Err(self.failure(
                                field,
                                page,
                                row_index,
                                &format!("{} array element", json_kind(other)),
                            ));
                        }
                    },
                };
                match text {
                    Some(text) => builder.values().append_value(text),
                    None => builder.values().append_null(),
                }
            }
            builder.append(true);
        }
        Ok(Arc::new(builder.finish()))
    }

    fn failure(
        &self,
        field: &CompiledField,
        page: usize,
        row: usize,
        found: &str,
    ) -> OpenConnectorError {
        OpenConnectorError::ConversionFailed {
            path: field.path.as_str().to_string(),
            column: field.spec.name.clone(),
            page,
            row,
            expected: field.spec.field_type.label().to_string(),
            found: found.to_string(),
        }
    }
}

/// Project `cells` through `convert`, mapping JSON null to Arrow null for
/// nullable fields and failing otherwise — for the common case where a type
/// mismatch is the only failure mode, described by the value's JSON kind.
fn collect_cells<T, F, E>(
    cells: &[Option<&Value>],
    spec: &ColumnSpec,
    convert: F,
    fail: E,
) -> Result<Vec<Option<T>>, OpenConnectorError>
where
    F: Fn(&Value) -> Option<T>,
    E: Fn(usize, &str) -> OpenConnectorError,
{
    collect_cells_described(
        cells,
        spec,
        |value| convert(value).ok_or_else(|| json_kind(value).to_string()),
        fail,
    )
}

/// [`collect_cells`] with converter-described failures, for conversions
/// with more than one failure mode — an out-of-range epoch second must not
/// masquerade as "found: number" when the value *is* a number.
fn collect_cells_described<T, F, E>(
    cells: &[Option<&Value>],
    spec: &ColumnSpec,
    convert: F,
    fail: E,
) -> Result<Vec<Option<T>>, OpenConnectorError>
where
    F: Fn(&Value) -> Result<T, String>,
    E: Fn(usize, &str) -> OpenConnectorError,
{
    let mut out = Vec::with_capacity(cells.len());
    for (index, cell) in cells.iter().enumerate() {
        match cell {
            None => out.push(None),
            Some(value) if value.is_null() => {
                if spec.nullable {
                    out.push(None);
                } else {
                    return Err(fail(index, "null"));
                }
            }
            Some(value) => match convert(value) {
                Ok(converted) => out.push(Some(converted)),
                Err(found) => return Err(fail(index, &found)),
            },
        }
    }
    Ok(out)
}

/// Digit-string epoch → epoch millis at the given scale (1 for
/// millis-valued strings, 1000 for seconds-valued strings). Only
/// non-empty ASCII-digit strings convert — an arbitrary string is shape
/// drift and must fail with a description, never parse as zero.
fn parse_epoch_digit_string(value: &Value, scale: i64) -> Result<i64, String> {
    let Some(text) = value.as_str() else {
        return Err(json_kind(value).to_string());
    };
    if text.is_empty() || !text.bytes().all(|b| b.is_ascii_digit()) {
        return Err("a non-epoch string".to_string());
    }
    let epoch: i64 = text
        .parse()
        .map_err(|_| "an epoch out of range for millisecond timestamps".to_string())?;
    epoch
        .checked_mul(scale)
        .ok_or_else(|| "an epoch out of range for millisecond timestamps".to_string())
}

/// Fractional epoch-seconds string → epoch millis, flooring anything below
/// a millisecond (Slack `ts`: `"1700000000.123456"`).
///
/// Accepts `<digits>` and `<digits>.<digits>`; a bare `.`, a sign, an
/// exponent, whitespace or any other shape is drift and fails with a
/// description rather than parsing as zero — the same rule
/// [`parse_epoch_digit_string`] applies, widened by exactly one dot.
///
/// Flooring is done on the digit string, not via floating point: `f64`
/// cannot represent `1700000000.123456` exactly, and the nearest double
/// rounds *below* it, which would put a value one millisecond early for
/// some inputs and misorder rows against the `ts` string beside them.
fn parse_epoch_fractional_seconds_string(value: &Value) -> Result<i64, String> {
    const OUT_OF_RANGE: &str = "an epoch out of range for millisecond timestamps";

    let Some(text) = value.as_str() else {
        return Err(json_kind(value).to_string());
    };
    let (whole, fraction) = text.split_once('.').unwrap_or((text, "0"));
    let digits = |s: &str| !s.is_empty() && s.bytes().all(|b| b.is_ascii_digit());
    if !digits(whole) || !digits(fraction) {
        return Err("a non-epoch string".to_string());
    }
    let seconds: i64 = whole.parse().map_err(|_| OUT_OF_RANGE.to_string())?;
    // First three fractional digits, right-padded — `.5` is 500ms, and
    // `.123456` truncates to 123ms.
    let millis = fraction
        .bytes()
        .chain(std::iter::repeat(b'0'))
        .take(3)
        .fold(0i64, |acc, b| acc * 10 + i64::from(b - b'0'));
    seconds
        .checked_mul(1000)
        .and_then(|base| base.checked_add(millis))
        .ok_or_else(|| OUT_OF_RANGE.to_string())
}

/// RFC 3339 string or epoch-millis number → epoch millis.
fn parse_timestamp(value: &Value) -> Option<i64> {
    if let Some(text) = value.as_str() {
        let parsed = DateTime::parse_from_rfc3339(text).ok()?;
        return Some(parsed.timestamp_millis());
    }
    value.as_i64()
}
/// Short human-readable kind of a JSON value, for error messages.
// The shared repo-wide kind vocabulary (article-prefixed: "a boolean").
use crate::util::json::json_kind;

#[cfg(test)]
mod tests {
    use super::*;
    use arrow::array::{Array, StringArray};
    use serde_json::json;

    const FIELDS: &[FieldMapping] = &[
        FieldMapping {
            name: "id",
            path: "id",
            field_type: FieldType::UInt64,
            nullable: false,
        },
        FieldMapping {
            name: "title",
            path: "title",
            field_type: FieldType::Utf8,
            nullable: false,
        },
        FieldMapping {
            name: "author_login",
            path: "user.login",
            field_type: FieldType::Utf8,
            nullable: true,
        },
        FieldMapping {
            name: "score",
            path: "score",
            field_type: FieldType::Float64,
            nullable: true,
        },
        FieldMapping {
            name: "labels",
            path: "labels",
            field_type: FieldType::Utf8List,
            nullable: true,
        },
        FieldMapping {
            name: "created_at",
            path: "created_at",
            field_type: FieldType::TimestampMillisUtc,
            nullable: true,
        },
        FieldMapping {
            name: "raw",
            path: "raw",
            field_type: FieldType::Json,
            nullable: true,
        },
    ];

    fn converter() -> RowConverter {
        RowConverter::new(FIELDS).unwrap()
    }

    fn row(id: u64, title: &str) -> Value {
        json!({
            "id": id,
            "title": title,
            "user": {"login": "octocat"},
            "score": 9.5,
            "labels": ["bug", "p1"],
            "created_at": "2026-01-01T00:00:00Z",
            "raw": {"nested": [1, 2]},
            "extra_upstream_field": "ignored"
        })
    }

    #[test]
    fn converts_full_rows_and_ignores_extra_fields() {
        let batch = converter().convert(&[row(1, "a"), row(2, "b")], 1).unwrap();
        assert_eq!(batch.num_rows(), 2);
        assert_eq!(batch.num_columns(), 7);
        let ids = batch
            .column(0)
            .as_any()
            .downcast_ref::<UInt64Array>()
            .unwrap();
        assert_eq!(ids.value(1), 2);
        let authors = batch
            .column(2)
            .as_any()
            .downcast_ref::<StringArray>()
            .unwrap();
        assert_eq!(authors.value(0), "octocat");
    }

    #[test]
    fn missing_required_key_fails_with_context() {
        let mut value = row(1, "a");
        value.as_object_mut().unwrap().remove("title");
        let err = converter().convert(&[value], 2).unwrap_err();
        assert!(matches!(
            err,
            OpenConnectorError::ConversionFailed { ref column, page: 2, row: 0, ref found, .. }
                if column == "title" && found == "missing key"
        ));
    }

    #[test]
    fn missing_nullable_key_becomes_null() {
        let mut value = row(1, "a");
        value.as_object_mut().unwrap().remove("score");
        let batch = converter().convert(&[value], 1).unwrap();
        let scores = batch
            .column(3)
            .as_any()
            .downcast_ref::<Float64Array>()
            .unwrap();
        assert!(scores.is_null(0));
    }

    #[test]
    fn nullable_column_fails_on_shape_mismatch() {
        // `user` changed from object to string upstream: a nullable
        // user.login column must NOT quietly become all-null — this is a
        // breaking change and has to fail conversion.
        let mut value = row(1, "a");
        value["user"] = json!("octocat");
        let err = converter().convert(&[value], 1).unwrap_err();
        match err {
            OpenConnectorError::ConversionFailed { column, found, .. } => {
                assert_eq!(column, "author_login");
                assert_eq!(found, "a string");
            }
            other => panic!("expected ConversionFailed, got {other}"),
        }
    }

    #[test]
    fn null_parent_object_nulls_for_nullable_column() {
        // Providers null out whole objects (GitHub `commit.author: null`,
        // `issue.user: null`): a JSON-null parent means every leaf under it
        // is absent — NULL for a nullable column, not a structural failure.
        let mut value = row(1, "a");
        value["user"] = Value::Null;
        let batch = converter().convert(&[value], 1).unwrap();
        let authors = batch
            .column(2)
            .as_any()
            .downcast_ref::<StringArray>()
            .unwrap();
        assert!(authors.is_null(0));
    }

    #[test]
    fn null_parent_object_fails_for_required_column() {
        let converter = RowConverter::new(&[FieldMapping {
            name: "author_login",
            path: "user.login",
            field_type: FieldType::Utf8,
            nullable: false,
        }])
        .unwrap();
        let err = converter
            .convert(&[serde_json::json!({"user": null})], 1)
            .unwrap_err();
        assert!(matches!(
            err,
            OpenConnectorError::ConversionFailed { ref column, ref found, .. }
                if column == "author_login" && found == "null"
        ));
    }

    #[test]
    fn object_list_plucks_the_declared_string_key() {
        // The design's `$.labels[*].name` flattening: an array of objects
        // becomes List<Utf8> of one declared key.
        let converter = RowConverter::new(&[FieldMapping {
            name: "labels",
            path: "labels",
            field_type: FieldType::Utf8ListFromObjectKey("name"),
            nullable: true,
        }])
        .unwrap();
        let batch = converter
            .convert(
                &[
                    serde_json::json!({"labels": [{"name": "bug", "color": "red"}, {"name": "p1"}]}),
                    serde_json::json!({"labels": []}),
                    serde_json::json!({"labels": null}),
                    serde_json::json!({}),
                ],
                1,
            )
            .unwrap();
        let lists = batch
            .column(0)
            .as_any()
            .downcast_ref::<arrow::array::ListArray>()
            .unwrap();
        let first = lists.value(0);
        let first = first.as_any().downcast_ref::<StringArray>().unwrap();
        assert_eq!(first.value(0), "bug");
        assert_eq!(first.value(1), "p1");
        assert_eq!(lists.value(1).len(), 0, "empty array stays an empty list");
        assert!(lists.is_null(2), "JSON null becomes NULL");
        assert!(lists.is_null(3), "absent key becomes NULL");
    }

    #[test]
    fn object_list_shape_mismatches_fail_with_the_specific_kind() {
        let converter = RowConverter::new(&[FieldMapping {
            name: "labels",
            path: "labels",
            field_type: FieldType::Utf8ListFromObjectKey("name"),
            nullable: true,
        }])
        .unwrap();
        for (bad, found) in [
            // A dropped upstream key is shape drift and must fail loudly —
            // it must never silently become an all-null list.
            (
                serde_json::json!({"labels": [{"color": "red"}]}),
                "array element without key 'name'",
            ),
            (
                serde_json::json!({"labels": [{"name": 42}]}),
                "array element whose 'name' is a number",
            ),
            (
                serde_json::json!({"labels": ["bug"]}),
                "a string array element",
            ),
        ] {
            let err = converter.convert(&[bad], 1).unwrap_err();
            assert!(
                matches!(
                    err,
                    OpenConnectorError::ConversionFailed { found: ref f, .. } if f == found
                ),
                "expected found={found}, got {err}"
            );
        }
    }

    #[test]
    fn json_null_list_elements_become_null_items() {
        // The item field is declared nullable (List<Utf8> with nullable
        // items), so an explicit provider null — a null element, or a pluck
        // key holding null — must convert to an Arrow null item, not fail.
        let converter = RowConverter::new(&[
            FieldMapping {
                name: "tags",
                path: "tags",
                field_type: FieldType::Utf8List,
                nullable: true,
            },
            FieldMapping {
                name: "labels",
                path: "labels",
                field_type: FieldType::Utf8ListFromObjectKey("name"),
                nullable: true,
            },
        ])
        .unwrap();
        let batch = converter
            .convert(
                &[serde_json::json!({
                    "tags": ["a", null, "b"],
                    "labels": [{"name": null}, {"name": "bug"}, null]
                })],
                1,
            )
            .unwrap();

        let tags = batch
            .column(0)
            .as_any()
            .downcast_ref::<arrow::array::ListArray>()
            .unwrap()
            .value(0);
        let tags = tags.as_any().downcast_ref::<StringArray>().unwrap();
        assert_eq!(tags.value(0), "a");
        assert!(tags.is_null(1), "null element is a null item");
        assert_eq!(tags.value(2), "b");

        let labels = batch
            .column(1)
            .as_any()
            .downcast_ref::<arrow::array::ListArray>()
            .unwrap()
            .value(0);
        let labels = labels.as_any().downcast_ref::<StringArray>().unwrap();
        assert!(labels.is_null(0), "pluck key holding null is a null item");
        assert_eq!(labels.value(1), "bug");
        assert!(labels.is_null(2), "null element is a null item");
    }

    #[test]
    fn missing_parent_object_still_nulls_for_nullable_column() {
        // A genuinely-absent parent (no `user` key at all) is absence, not a
        // shape change — nullable user.login becomes null.
        let mut value = row(1, "a");
        value.as_object_mut().unwrap().remove("user");
        let batch = converter().convert(&[value], 1).unwrap();
        let authors = batch
            .column(2)
            .as_any()
            .downcast_ref::<StringArray>()
            .unwrap();
        assert!(authors.is_null(0));
    }

    #[test]
    fn null_in_required_field_fails() {
        let mut value = row(1, "a");
        value["title"] = Value::Null;
        let err = converter().convert(&[value], 1).unwrap_err();
        assert!(matches!(
            err,
            OpenConnectorError::ConversionFailed { ref column, ref found, .. }
                if column == "title" && found == "null"
        ));
    }

    #[test]
    fn wrong_type_fails_with_kind_not_value() {
        let mut value = row(1, "a");
        value["id"] = json!("not-a-number");
        let err = converter().convert(&[value], 1).unwrap_err();
        match err {
            OpenConnectorError::ConversionFailed { found, column, .. } => {
                assert_eq!(column, "id");
                assert_eq!(found, "a string");
                assert!(!found.contains("not-a-number"), "error must not echo data");
            }
            other => panic!("expected ConversionFailed, got {other}"),
        }
    }

    #[test]
    fn float_for_int_fails() {
        let mut value = row(1, "a");
        value["id"] = json!(1.5);
        let err = converter().convert(&[value], 1).unwrap_err();
        assert!(matches!(err, OpenConnectorError::ConversionFailed { .. }));
    }

    #[test]
    fn timestamps_parse_rfc3339_and_epoch() {
        let mut value = row(1, "a");
        let batch = converter().convert(&[value.clone()], 1).unwrap();
        let ts = batch
            .column(5)
            .as_any()
            .downcast_ref::<TimestampMillisecondArray>()
            .unwrap();
        assert_eq!(ts.value(0), 1767225600000);

        value["created_at"] = json!(1767225600000i64);
        let batch = converter().convert(&[value], 1).unwrap();
        let ts = batch
            .column(5)
            .as_any()
            .downcast_ref::<TimestampMillisecondArray>()
            .unwrap();
        assert_eq!(ts.value(0), 1767225600000);
    }

    #[test]
    fn epoch_seconds_convert_to_millis_and_reject_other_shapes() {
        // Slack-style `created`/`updated` are whole epoch seconds; the
        // seconds variant scales to millis, and anything that is not an
        // integer fails with its kind — a fractional or string timestamp
        // must never be misread.
        let converter = RowConverter::new(&[FieldMapping {
            name: "created",
            path: "created",
            field_type: FieldType::TimestampSecondsUtc,
            nullable: true,
        }])
        .unwrap();

        let batch = converter
            .convert(
                &[
                    serde_json::json!({"created": 1735689600}), // 2025-01-01T00:00:00Z
                    serde_json::json!({"created": null}),
                    serde_json::json!({}),
                ],
                1,
            )
            .unwrap();
        let created = batch
            .column(0)
            .as_any()
            .downcast_ref::<TimestampMillisecondArray>()
            .unwrap();
        assert_eq!(created.value(0), 1_735_689_600_000);
        assert!(created.is_null(1), "JSON null stays NULL");
        assert!(created.is_null(2), "absent key stays NULL");

        for (bad, found) in [
            (serde_json::json!({"created": 1735689600.5}), "a number"),
            (
                serde_json::json!({"created": "2025-01-01T00:00:00Z"}),
                "a string",
            ),
            // Overflow is its own diagnosis — "found: number" would
            // contradict the expected type, which is also a number.
            (
                serde_json::json!({"created": i64::MAX}),
                "an epoch second out of range for millisecond timestamps",
            ),
        ] {
            let err = converter.convert(&[bad], 1).unwrap_err();
            assert!(
                matches!(
                    err,
                    OpenConnectorError::ConversionFailed { found: ref f, ref expected, .. }
                        if f == found && expected == "epoch-seconds timestamp"
                ),
                "got {err}"
            );
        }
    }

    #[test]
    fn fractional_epoch_strings_floor_to_millis_and_reject_other_shapes() {
        // Slack's message `ts` is "<epoch seconds>.<microseconds>", which
        // every other reader refuses: the digit-string readers stop at the
        // dot, and the numeric ones want a JSON number. Sub-millisecond
        // precision floors, so the column stays ordered the same way as the
        // `ts` string it is derived from.
        let converter = RowConverter::new(&[FieldMapping {
            name: "sent_at",
            path: "ts",
            field_type: FieldType::TimestampSecondsFractionalStringUtc,
            nullable: true,
        }])
        .unwrap();

        let batch = converter
            .convert(
                &[
                    serde_json::json!({"ts": "1735689600.123456"}),
                    // A bare seconds string is legal and means .000.
                    serde_json::json!({"ts": "1735689600"}),
                    // Fewer than three fractional digits right-pad: .5 is
                    // half a second, not half a millisecond.
                    serde_json::json!({"ts": "1735689600.5"}),
                    serde_json::json!({"ts": null}),
                ],
                1,
            )
            .unwrap();
        let ts = batch
            .column(0)
            .as_any()
            .downcast_ref::<TimestampMillisecondArray>()
            .unwrap();
        assert_eq!(ts.value(0), 1_735_689_600_123, "microseconds floor away");
        assert_eq!(ts.value(1), 1_735_689_600_000);
        assert_eq!(ts.value(2), 1_735_689_600_500);
        assert!(ts.is_null(3));

        // Two `ts` one microsecond apart land on the same millisecond —
        // the reason the raw string stays its own column rather than this
        // one doubling as the row key.
        let collided = converter
            .convert(
                &[
                    serde_json::json!({"ts": "1735689600.123456"}),
                    serde_json::json!({"ts": "1735689600.123457"}),
                ],
                1,
            )
            .unwrap();
        let collided = collided
            .column(0)
            .as_any()
            .downcast_ref::<TimestampMillisecondArray>()
            .unwrap();
        assert_eq!(collided.value(0), collided.value(1));

        for (bad, found) in [
            // The number spelling belongs to the numeric readers; accepting
            // it here would blur this variant's contract with theirs.
            (
                serde_json::json!({"ts": 1_735_689_600.123_456_f64}),
                "a number",
            ),
            (
                serde_json::json!({"ts": "2025-01-01T00:00:00Z"}),
                "a non-epoch string",
            ),
            (serde_json::json!({"ts": ""}), "a non-epoch string"),
            // A dot with nothing on one side of it is drift, not zero.
            (
                serde_json::json!({"ts": "1735689600."}),
                "a non-epoch string",
            ),
            (serde_json::json!({"ts": ".123"}), "a non-epoch string"),
            (
                serde_json::json!({"ts": "-1735689600.1"}),
                "a non-epoch string",
            ),
            (
                serde_json::json!({"ts": "1735689600.1e3"}),
                "a non-epoch string",
            ),
            // Two dots would otherwise leave "1.2" in the fraction.
            (
                serde_json::json!({"ts": "1735689600.1.2"}),
                "a non-epoch string",
            ),
            (
                serde_json::json!({"ts": "9223372036854775807.9"}),
                "an epoch out of range for millisecond timestamps",
            ),
        ] {
            let err = converter.convert(&[bad], 1).unwrap_err();
            assert!(
                matches!(
                    err,
                    OpenConnectorError::ConversionFailed { found: ref f, ref expected, .. }
                        if f == found && expected == "fractional epoch-seconds string"
                ),
                "got {err}"
            );
        }
    }

    #[test]
    fn epoch_digit_strings_convert_at_both_scales_and_reject_other_shapes() {
        // Feishu-style timestamps are DIGIT STRINGS — millis for im
        // (`create_time: "1609296809000"`), seconds for wiki
        // (`obj_create_time: "1642402790"`). Each variant converts only
        // non-empty ASCII-digit strings; numbers, RFC 3339 strings, and
        // arbitrary text fail with their kind or shape — never parse as
        // zero.
        let converter = RowConverter::new(&[
            FieldMapping {
                name: "create_time",
                path: "create_time",
                field_type: FieldType::TimestampMillisStringUtc,
                nullable: true,
            },
            FieldMapping {
                name: "obj_create_time",
                path: "obj_create_time",
                field_type: FieldType::TimestampSecondsStringUtc,
                nullable: true,
            },
        ])
        .unwrap();

        let batch = converter
            .convert(
                &[
                    serde_json::json!({
                        "create_time": "1735689600000", // 2025-01-01T00:00:00Z
                        "obj_create_time": "1735689600",
                    }),
                    serde_json::json!({"create_time": null}),
                ],
                1,
            )
            .unwrap();
        for column in [0, 1] {
            let ts = batch
                .column(column)
                .as_any()
                .downcast_ref::<TimestampMillisecondArray>()
                .unwrap();
            assert_eq!(ts.value(0), 1_735_689_600_000);
            assert!(ts.is_null(1), "null and absent keys stay NULL");
        }

        for (bad, found, expected) in [
            // A JSON number is the OTHER spelling — accepting it here
            // would blur the two variants' contracts.
            (
                serde_json::json!({"create_time": 1735689600000_i64}),
                "a number",
                "epoch-millis digit string",
            ),
            (
                serde_json::json!({"create_time": "2025-01-01T00:00:00Z"}),
                "a non-epoch string",
                "epoch-millis digit string",
            ),
            (
                serde_json::json!({"create_time": ""}),
                "a non-epoch string",
                "epoch-millis digit string",
            ),
            (
                serde_json::json!({"obj_create_time": "-5"}),
                "a non-epoch string",
                "epoch-seconds digit string",
            ),
            // Overflow is its own diagnosis, at both the parse and the
            // seconds→millis scale step.
            (
                serde_json::json!({"obj_create_time": "99999999999999999999"}),
                "an epoch out of range for millisecond timestamps",
                "epoch-seconds digit string",
            ),
            (
                serde_json::json!({"obj_create_time": i64::MAX.to_string()}),
                "an epoch out of range for millisecond timestamps",
                "epoch-seconds digit string",
            ),
        ] {
            let err = converter.convert(&[bad], 1).unwrap_err();
            assert!(
                matches!(
                    err,
                    OpenConnectorError::ConversionFailed { found: ref f, expected: ref e, .. }
                        if f == found && e == expected
                ),
                "got {err}"
            );
        }
    }

    #[test]
    fn non_string_list_element_fails() {
        let mut value = row(1, "a");
        value["labels"] = json!(["bug", 42]);
        let err = converter().convert(&[value], 1).unwrap_err();
        assert!(matches!(err, OpenConnectorError::ConversionFailed { .. }));
    }

    #[test]
    fn opaque_field_serializes_json() {
        let batch = converter().convert(&[row(1, "a")], 1).unwrap();
        let raw = batch
            .column(6)
            .as_any()
            .downcast_ref::<StringArray>()
            .unwrap();
        assert!(raw.value(0).contains("nested"));
    }

    #[test]
    fn json_null_becomes_arrow_null_not_the_string_null() {
        // A present JSON null in a nullable Json column is SQL NULL — never
        // the 4-char string "null", which would make `IS NULL` miss and
        // `= 'null'` match. Raw scans type every object/array field as Json,
        // so nullable nested SaaS fields (assignee: null) hit this arm.
        let mut with_null = row(1, "a");
        with_null["raw"] = Value::Null;
        let mut absent = row(2, "b");
        absent.as_object_mut().unwrap().remove("raw");

        let batch = converter().convert(&[with_null, absent], 1).unwrap();
        let raw = batch
            .column(6)
            .as_any()
            .downcast_ref::<StringArray>()
            .unwrap();
        assert!(raw.is_null(0), "present JSON null must be Arrow null");
        assert!(raw.is_null(1), "absent key stays Arrow null");
    }

    #[test]
    fn json_null_fails_for_non_nullable_json_column() {
        // Same null discipline as every other type: a required opaque field
        // fails with a targeted per-column error, not a batch-level one.
        let converter = RowConverter::new(&[FieldMapping {
            name: "raw",
            path: "raw",
            field_type: FieldType::Json,
            nullable: false,
        }])
        .unwrap();
        let err = converter
            .convert(&[serde_json::json!({"raw": null})], 3)
            .unwrap_err();
        assert!(matches!(
            err,
            OpenConnectorError::ConversionFailed { ref column, page: 3, row: 0, ref found, .. }
                if column == "raw" && found == "null"
        ));
    }

    #[test]
    fn empty_page_yields_empty_batch_with_schema() {
        let converter = converter();
        let batch = converter.convert(&[], 1).unwrap();
        assert_eq!(batch.num_rows(), 0);
        assert_eq!(batch.schema().fields().len(), 7);
    }

    #[test]
    fn owned_columns_build_the_same_converter_as_static_mappings() {
        // The raw-scan path derives ColumnSpecs at planning time; they must
        // produce exactly the schema and conversion the static path does.
        let owned = RowConverter::from_columns(FIELDS.iter().map(ColumnSpec::from).collect())
            .expect("owned converter");
        assert_eq!(owned.schema(), converter().schema());

        let batch = owned.convert(&[row(7, "seven")], 1).unwrap();
        let ids = batch
            .column(0)
            .as_any()
            .downcast_ref::<UInt64Array>()
            .unwrap();
        assert_eq!(ids.value(0), 7);
    }

    #[test]
    fn dollar_prefixed_column_paths_are_rejected_not_silently_misread() {
        // `$.user.login` would survive the `$.{path}` concatenation as a
        // literal `"$"` first segment and read the wrong key — nullable
        // columns would silently go all-NULL. Reject it with the corrected
        // spelling instead.
        let err = RowConverter::new(&[FieldMapping {
            name: "author",
            path: "$.user.login",
            field_type: FieldType::Utf8,
            nullable: true,
        }])
        .unwrap_err();
        assert!(matches!(
            err,
            OpenConnectorError::InvalidRowPath { ref path, ref reason }
                if path == "$.user.login"
                    && reason.contains("write 'user.login', not '$.user.login'")
        ));
    }

    #[test]
    fn invalid_mapping_path_is_rejected() {
        let err = RowConverter::new(&[FieldMapping {
            name: "x",
            path: "a..b",
            field_type: FieldType::Utf8,
            nullable: false,
        }])
        .unwrap_err();
        assert!(matches!(err, OpenConnectorError::InvalidRowPath { .. }));
    }
}