arrow_json/reader/

mod.rs

// SPDX-License-Identifier: Apache-2.0
// SPDX-FileCopyrightText: See the NOTICE file distributed with this work for additional information regarding copyright ownership
// SPDX-FileContributor: Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements

//! JSON reader
//!
//! This JSON reader allows JSON records to be read into the Arrow memory
//! model. Records are loaded in batches and are then converted from the record-oriented
//! representation to the columnar arrow data model.
//!
//! The reader ignores whitespace between JSON values, including `\n` and `\r`, allowing
//! parsing of sequences of one or more arbitrarily formatted JSON values, including
//! but not limited to newline-delimited JSON.
//!
//! # Timestamps
//!
//! Timestamps are parsed with [`arrow_cast::parse::string_to_datetime`]. A
//! timestamp without an explicit offset that falls in a daylight savings gap or
//! overlap of the target timezone is resolved as described there, rather than
//! returning an error.
//!
//! # Basic Usage
//!
//! [`Reader`] can be used directly with synchronous data sources, such as [`std::fs::File`]
//!
//! ```
//! # use arrow_schema::*;
//! # use std::fs::File;
//! # use std::io::BufReader;
//! # use std::sync::Arc;
//!
//! let schema = Arc::new(Schema::new(vec![
//!     Field::new("a", DataType::Float64, false),
//!     Field::new("b", DataType::Float64, false),
//!     Field::new("c", DataType::Boolean, true),
//! ]));
//!
//! let file = File::open("test/data/basic.json").unwrap();
//!
//! let mut json = arrow_json::ReaderBuilder::new(schema).build(BufReader::new(file)).unwrap();
//! let batch = json.next().unwrap().unwrap();
//! ```
//!
//! # Async Usage
//!
//! The lower-level [`Decoder`] can be integrated with various forms of async
//! data streams, and can be used with any of the async IO primitives found
//! within the Rust ecosystem.
//!
//! For example, it can be used with an arbitrary `Stream` of `Bytes` as shown below
//!
//! ```
//! # use std::task::{Poll, ready};
//! # use bytes::{Buf, Bytes};
//! # use arrow_schema::ArrowError;
//! # use futures::stream::{Stream, StreamExt};
//! # use arrow_array::RecordBatch;
//! # use arrow_json::reader::Decoder;
//! #
//! fn decode_stream<S: Stream<Item = Bytes> + Unpin>(
//!     mut decoder: Decoder,
//!     mut input: S,
//! ) -> impl Stream<Item = Result<RecordBatch, ArrowError>> {
//!     let mut buffered = Bytes::new();
//!     futures::stream::poll_fn(move |cx| {
//!         loop {
//!             if buffered.is_empty() {
//!                 buffered = match ready!(input.poll_next_unpin(cx)) {
//!                     Some(b) => b,
//!                     None => break,
//!                 };
//!             }
//!             let decoded = match decoder.decode(buffered.as_ref()) {
//!                 Ok(decoded) => decoded,
//!                 Err(e) => return Poll::Ready(Some(Err(e))),
//!             };
//!             let read = buffered.len();
//!             buffered.advance(decoded);
//!             if decoded != read {
//!                 break
//!             }
//!         }
//!
//!         Poll::Ready(decoder.flush().transpose())
//!     })
//! }
//!
//! ```
//!
//! In a similar vein, it can also be used with tokio-based IO primitives
//!
//! ```
//! # use std::sync::Arc;
//! # use arrow_schema::{DataType, Field, Schema};
//! # use std::pin::Pin;
//! # use std::task::{Poll, ready};
//! # use futures::{Stream, TryStreamExt};
//! # use tokio::io::AsyncBufRead;
//! # use arrow_array::RecordBatch;
//! # use arrow_json::reader::Decoder;
//! # use arrow_schema::ArrowError;
//! fn decode_stream<R: AsyncBufRead + Unpin>(
//!     mut decoder: Decoder,
//!     mut reader: R,
//! ) -> impl Stream<Item = Result<RecordBatch, ArrowError>> {
//!     futures::stream::poll_fn(move |cx| {
//!         loop {
//!             let b = match ready!(Pin::new(&mut reader).poll_fill_buf(cx)) {
//!                 Ok(b) if b.is_empty() => break,
//!                 Ok(b) => b,
//!                 Err(e) => return Poll::Ready(Some(Err(e.into()))),
//!             };
//!             let read = b.len();
//!             let decoded = match decoder.decode(b) {
//!                 Ok(decoded) => decoded,
//!                 Err(e) => return Poll::Ready(Some(Err(e))),
//!             };
//!             Pin::new(&mut reader).consume(decoded);
//!             if decoded != read {
//!                 break;
//!             }
//!         }
//!
//!         Poll::Ready(decoder.flush().transpose())
//!     })
//! }
//! ```
//!
//! # Customizing the decoder
//!
//! The decoding of each data type can be customized using
//! [`ReaderBuilder::with_decoder_factory`]. For example, you can override the
//! default decoding of binary data from hex strings to decode from an array of
//! integers instead, or provide decoders for types with no built-in decoding.
//! See the example on [`DecoderFactory`].
//!

use std::io::BufRead;
use std::sync::Arc;

use arrow_array::cast::AsArray;
use arrow_array::timezone::Tz;
use arrow_array::types::*;
use arrow_array::{ArrayRef, RecordBatch, RecordBatchReader, downcast_integer};
use arrow_schema::{ArrowError, DataType, Field, FieldRef, Schema, SchemaRef, TimeUnit};
use chrono::Utc;
use serde_core::Serialize;

use crate::StructMode;
use crate::reader::binary_array::{
    BinaryArrayDecoder, BinaryViewDecoder, FixedSizeBinaryArrayDecoder,
};
use crate::reader::boolean_array::BooleanArrayDecoder;
use crate::reader::decimal_array::DecimalArrayDecoder;
use crate::reader::list_array::{
    FixedSizeListArrayDecoder, ListArrayDecoder, ListViewArrayDecoder,
};
use crate::reader::map_array::MapArrayDecoder;
use crate::reader::null_array::NullArrayDecoder;
use crate::reader::primitive_array::PrimitiveArrayDecoder;
use crate::reader::run_end_array::RunEndEncodedArrayDecoder;
use crate::reader::string_array::StringArrayDecoder;
use crate::reader::string_view_array::StringViewArrayDecoder;
use crate::reader::struct_array::StructArrayDecoder;
use crate::reader::tape::{TapeDecoder, TapeDecoderOptions};
use crate::reader::timestamp_array::TimestampArrayDecoder;

pub use schema::*;
// `mod tape` is private by design: exposing only these two types means downstream can read a
// `Tape` but not construct one, keeping `TapeDecoder` an implementation detail.
pub use tape::{Tape, TapeElement};
pub use value_iter::ValueIter;

mod binary_array;
mod boolean_array;
mod decimal_array;
mod list_array;
mod map_array;
mod null_array;
mod primitive_array;
mod run_end_array;
mod schema;
mod serializer;
mod string_array;
mod string_view_array;
mod struct_array;
mod tape;
mod timestamp_array;
mod value_iter;

/// A builder for [`Reader`] and [`Decoder`]
#[derive(Clone)]
pub struct ReaderBuilder {
    batch_size: usize,
    coerce_primitive: bool,
    strict_mode: bool,
    ignore_type_conflicts: bool,
    allow_non_finite_floats: bool,
    is_field: bool,
    struct_mode: StructMode,
    flatten_top_level_arrays: bool,
    decoder_factory: Option<Arc<dyn DecoderFactory>>,
    schema: SchemaRef,
}

impl ReaderBuilder {
    /// Create a new [`ReaderBuilder`] with the provided [`SchemaRef`]
    ///
    /// This could be obtained using [`infer_json_schema`] if not known
    ///
    /// Any columns not present in `schema` will be ignored, unless `strict_mode` is set to true.
    /// In this case, an error is returned when a column is missing from `schema`.
    ///
    /// [`infer_json_schema`]: crate::reader::infer_json_schema
    pub fn new(schema: SchemaRef) -> Self {
        Self {
            batch_size: 1024,
            coerce_primitive: false,
            strict_mode: false,
            ignore_type_conflicts: false,
            allow_non_finite_floats: false,
            is_field: false,
            struct_mode: Default::default(),
            flatten_top_level_arrays: false,
            decoder_factory: None,
            schema,
        }
    }

    /// Create a new [`ReaderBuilder`] that will parse JSON values of `field.data_type()`
    ///
    /// Unlike [`ReaderBuilder::new`] this does not require the root of the JSON data
    /// to be an object, i.e. `{..}`, allowing for parsing of any valid JSON value(s)
    ///
    /// ```
    /// # use std::sync::Arc;
    /// # use arrow_array::cast::AsArray;
    /// # use arrow_array::types::Int32Type;
    /// # use arrow_json::ReaderBuilder;
    /// # use arrow_schema::{DataType, Field};
    /// // Root of JSON schema is a numeric type
    /// let data = "1\n2\n3\n";
    /// let field = Arc::new(Field::new("int", DataType::Int32, true));
    /// let mut reader = ReaderBuilder::new_with_field(field.clone()).build(data.as_bytes()).unwrap();
    /// let b = reader.next().unwrap().unwrap();
    /// let values = b.column(0).as_primitive::<Int32Type>().values();
    /// assert_eq!(values, &[1, 2, 3]);
    ///
    /// // Root of JSON schema is a list type
    /// let data = "[1, 2, 3, 4, 5, 6, 7]\n[1, 2, 3]";
    /// let field = Field::new_list("int", field.clone(), true);
    /// let mut reader = ReaderBuilder::new_with_field(field).build(data.as_bytes()).unwrap();
    /// let b = reader.next().unwrap().unwrap();
    /// let list = b.column(0).as_list::<i32>();
    ///
    /// assert_eq!(list.offsets().as_ref(), &[0, 7, 10]);
    /// let list_values = list.values().as_primitive::<Int32Type>();
    /// assert_eq!(list_values.values(), &[1, 2, 3, 4, 5, 6, 7, 1, 2, 3]);
    /// ```
    pub fn new_with_field(field: impl Into<FieldRef>) -> Self {
        Self {
            batch_size: 1024,
            coerce_primitive: false,
            strict_mode: false,
            ignore_type_conflicts: false,
            allow_non_finite_floats: false,
            is_field: true,
            struct_mode: Default::default(),
            flatten_top_level_arrays: false,
            decoder_factory: None,
            schema: Arc::new(Schema::new([field.into()])),
        }
    }

    /// Sets the batch size in rows to read
    pub fn with_batch_size(self, batch_size: usize) -> Self {
        Self { batch_size, ..self }
    }

    /// Sets if the decoder should coerce primitive values (bool and number) into string
    /// when the Schema's column is Utf8 or LargeUtf8.
    pub fn with_coerce_primitive(self, coerce_primitive: bool) -> Self {
        Self {
            coerce_primitive,
            ..self
        }
    }

    /// Accepts the nonstandard number literals `NaN`, `Infinity`, and `-Infinity`.
    ///
    /// Disabled by default. Floating-point columns preserve their IEEE values;
    /// other types follow the existing coercion and type-conflict options.
    /// Applies to known-schema readers only, not schema inference.
    pub fn with_allow_non_finite_floats(self, allow_non_finite_floats: bool) -> Self {
        Self {
            allow_non_finite_floats,
            ..self
        }
    }

    /// Sets if the decoder should return an error if it encounters a column not
    /// present in `schema`. If `struct_mode` is `ListOnly` the value of
    /// `strict_mode` is effectively `true`. It is required for all fields of
    /// the struct to be in the list: without field names, there is no way to
    /// determine which field is missing.
    pub fn with_strict_mode(self, strict_mode: bool) -> Self {
        Self {
            strict_mode,
            ..self
        }
    }

    /// Set the [`StructMode`] for the reader, which determines whether structs
    /// can be decoded from JSON as objects or lists. For more details refer to
    /// the enum documentation. Default is to use `ObjectOnly`.
    pub fn with_struct_mode(self, struct_mode: StructMode) -> Self {
        Self {
            struct_mode,
            ..self
        }
    }

    /// Sets whether the decoder should produce NULL instead of returning an error if it encounters
    /// value that can not be parsed into the specified column type.
    ///
    /// For example, if the type is declared to be a nullable array of `DataType::Int32` but the
    /// reader encounters a string value `"foo"` and the value `ignore_type_conflicts` is:
    ///
    /// * `false` (the default): The reader will return an error.
    ///
    /// * `true`: The reader will fill in NULL value for that array element.
    ///
    /// NOTE: An inferred NULL due to a type conflict will still produce parsing errors for
    /// non-nullable fields, the same as any other NULL or missing value.
    pub fn with_ignore_type_conflicts(self, ignore_type_conflicts: bool) -> Self {
        Self {
            ignore_type_conflicts,
            ..self
        }
    }
    /// Sets whether to flatten top-level arrays.
    ///
    /// * When `true`, each element of a top-level array will be treated as its own row.
    /// * When `false` (the default), the entire top-level array will be treated as one row.
    ///
    /// For example, consider this input file:
    /// ```text
    /// [{ "a": 1 }, { "a": 2 }, { "b": 3 }]
    /// [{ "a": 4 }, { "a": 5 }, { "b": 6 }]
    /// ```
    ///
    /// By default, this would be parsed as two rows, each an array containing three elements.
    /// With this option set to `true`, however, this would be parsed as six rows.
    ///
    /// Note that even with this option set to `true`, top-level objects are still permitted
    /// and will be parsed as individual rows in the exact same manner as when this option is
    /// set to `false`. It is even possible to mix top-level arrays with top-level objects.
    pub fn with_flatten(self, flatten_top_level_arrays: bool) -> Self {
        Self {
            flatten_top_level_arrays,
            ..self
        }
    }

    /// Set a hook for customizing decoding behavior. See [`DecoderFactory`].
    pub fn with_decoder_factory(self, decoder_factory: Arc<dyn DecoderFactory>) -> Self {
        Self {
            decoder_factory: Some(decoder_factory),
            ..self
        }
    }

    /// Create a [`Reader`] with the provided [`BufRead`]
    pub fn build<R: BufRead>(self, reader: R) -> Result<Reader<R>, ArrowError> {
        Ok(Reader {
            reader,
            decoder: self.build_decoder()?,
        })
    }

    /// Create a [`Decoder`]
    pub fn build_decoder(self) -> Result<Decoder, ArrowError> {
        let (field, nullable) = if self.is_field {
            let field = self.schema.fields[0].clone();
            let nullable = field.is_nullable();
            (field, nullable)
        } else {
            // The root has no field of its own; synthesize a nameless one
            let data_type = DataType::Struct(self.schema.fields.clone());
            (Arc::new(Field::new("", data_type, false)), false)
        };

        let ctx = DecoderContext {
            coerce_primitive: self.coerce_primitive,
            strict_mode: self.strict_mode,
            struct_mode: self.struct_mode,
            ignore_type_conflicts: self.ignore_type_conflicts,
            decoder_factory: self.decoder_factory,
        };
        let decoder = ctx.make_decoder(&field, nullable)?;

        let num_fields = self.schema.flattened_fields().len();

        Ok(Decoder {
            decoder,
            is_field: self.is_field,
            tape_decoder: TapeDecoder::new(TapeDecoderOptions {
                batch_size: self.batch_size,
                num_fields,
                flatten_top_level_arrays: self.flatten_top_level_arrays,
                allow_non_finite_floats: self.allow_non_finite_floats,
            }),
            batch_size: self.batch_size,
            schema: self.schema,
        })
    }
}

/// Reads JSON data with a known schema directly into arrow [`RecordBatch`]
///
/// Lines consisting solely of ASCII whitespace are ignored
pub struct Reader<R> {
    reader: R,
    decoder: Decoder,
}

impl<R> std::fmt::Debug for Reader<R> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Reader")
            .field("decoder", &self.decoder)
            .finish()
    }
}

impl<R: BufRead> Reader<R> {
    /// Reads the next [`RecordBatch`] returning `Ok(None)` if EOF
    fn read(&mut self) -> Result<Option<RecordBatch>, ArrowError> {
        loop {
            let buf = self.reader.fill_buf()?;
            if buf.is_empty() {
                break;
            }
            let read = buf.len();

            let decoded = self.decoder.decode(buf)?;
            self.reader.consume(decoded);
            if decoded != read {
                break;
            }
        }
        self.decoder.flush()
    }
}

impl<R: BufRead> Iterator for Reader<R> {
    type Item = Result<RecordBatch, ArrowError>;

    fn next(&mut self) -> Option<Self::Item> {
        self.read().transpose()
    }
}

impl<R: BufRead> RecordBatchReader for Reader<R> {
    fn schema(&self) -> SchemaRef {
        self.decoder.schema.clone()
    }
}

/// A low-level interface for reading JSON data from a byte stream
///
/// See [`Reader`] for a higher-level interface for interface with [`BufRead`]
///
/// The push-based interface facilitates integration with sources that yield arbitrarily
/// delimited bytes ranges, such as [`BufRead`], or a chunked byte stream received from
/// object storage
///
/// ```
/// # use std::io::BufRead;
/// # use arrow_array::RecordBatch;
/// # use arrow_json::reader::{Decoder, ReaderBuilder};
/// # use arrow_schema::{ArrowError, SchemaRef};
/// #
/// fn read_from_json<R: BufRead>(
///     mut reader: R,
///     schema: SchemaRef,
/// ) -> Result<impl Iterator<Item = Result<RecordBatch, ArrowError>>, ArrowError> {
///     let mut decoder = ReaderBuilder::new(schema).build_decoder()?;
///     let mut next = move || {
///         loop {
///             // Decoder is agnostic that buf doesn't contain whole records
///             let buf = reader.fill_buf()?;
///             if buf.is_empty() {
///                 break; // Input exhausted
///             }
///             let read = buf.len();
///             let decoded = decoder.decode(buf)?;
///
///             // Consume the number of bytes read
///             reader.consume(decoded);
///             if decoded != read {
///                 break; // Read batch size
///             }
///         }
///         decoder.flush()
///     };
///     Ok(std::iter::from_fn(move || next().transpose()))
/// }
/// ```
pub struct Decoder {
    tape_decoder: TapeDecoder,
    decoder: Box<dyn ArrayDecoder>,
    batch_size: usize,
    is_field: bool,
    schema: SchemaRef,
}

impl std::fmt::Debug for Decoder {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Decoder")
            .field("schema", &self.schema)
            .field("batch_size", &self.batch_size)
            .finish()
    }
}

impl Decoder {
    /// Read JSON objects from `buf`, returning the number of bytes read
    ///
    /// This method returns once `batch_size` objects have been parsed since the
    /// last call to [`Self::flush`], or `buf` is exhausted. Any remaining bytes
    /// should be included in the next call to [`Self::decode`]
    ///
    /// There is no requirement that `buf` contains a whole number of records, facilitating
    /// integration with arbitrary byte streams, such as those yielded by [`BufRead`]
    pub fn decode(&mut self, buf: &[u8]) -> Result<usize, ArrowError> {
        self.tape_decoder.decode(buf)
    }

    /// Serialize `rows` to this [`Decoder`]
    ///
    /// This provides a simple way to convert [serde]-compatible datastructures into arrow
    /// [`RecordBatch`].
    ///
    /// Custom conversion logic as described in [arrow_array::builder] will likely outperform this,
    /// especially where the schema is known at compile-time, however, this provides a mechanism
    /// to get something up and running quickly
    ///
    /// It can be used with [`serde_json::Value`]
    ///
    /// ```
    /// # use std::sync::Arc;
    /// # use serde_json::{Value, json};
    /// # use arrow_array::cast::AsArray;
    /// # use arrow_array::types::Float32Type;
    /// # use arrow_json::ReaderBuilder;
    /// # use arrow_schema::{DataType, Field, Schema};
    /// let json = vec![json!({"float": 2.3}), json!({"float": 5.7})];
    ///
    /// let schema = Schema::new(vec![Field::new("float", DataType::Float32, true)]);
    /// let mut decoder = ReaderBuilder::new(Arc::new(schema)).build_decoder().unwrap();
    ///
    /// decoder.serialize(&json).unwrap();
    /// let batch = decoder.flush().unwrap().unwrap();
    /// assert_eq!(batch.num_rows(), 2);
    /// assert_eq!(batch.num_columns(), 1);
    /// let values = batch.column(0).as_primitive::<Float32Type>().values();
    /// assert_eq!(values, &[2.3, 5.7])
    /// ```
    ///
    /// Or with arbitrary [`Serialize`] types
    ///
    /// ```
    /// # use std::sync::Arc;
    /// # use arrow_json::ReaderBuilder;
    /// # use arrow_schema::{DataType, Field, Schema};
    /// # use serde::Serialize;
    /// # use arrow_array::cast::AsArray;
    /// # use arrow_array::types::{Float32Type, Int32Type};
    /// #
    /// #[derive(Serialize)]
    /// struct MyStruct {
    ///     int32: i32,
    ///     float: f32,
    /// }
    ///
    /// let schema = Schema::new(vec![
    ///     Field::new("int32", DataType::Int32, false),
    ///     Field::new("float", DataType::Float32, false),
    /// ]);
    ///
    /// let rows = vec![
    ///     MyStruct{ int32: 0, float: 3. },
    ///     MyStruct{ int32: 4, float: 67.53 },
    /// ];
    ///
    /// let mut decoder = ReaderBuilder::new(Arc::new(schema)).build_decoder().unwrap();
    /// decoder.serialize(&rows).unwrap();
    ///
    /// let batch = decoder.flush().unwrap().unwrap();
    ///
    /// // Expect batch containing two columns
    /// let int32 = batch.column(0).as_primitive::<Int32Type>();
    /// assert_eq!(int32.values(), &[0, 4]);
    ///
    /// let float = batch.column(1).as_primitive::<Float32Type>();
    /// assert_eq!(float.values(), &[3., 67.53]);
    /// ```
    ///
    /// Or even complex nested types
    ///
    /// ```
    /// # use std::collections::BTreeMap;
    /// # use std::sync::Arc;
    /// # use arrow_array::StructArray;
    /// # use arrow_cast::display::{ArrayFormatter, FormatOptions};
    /// # use arrow_json::ReaderBuilder;
    /// # use arrow_schema::{DataType, Field, Fields, Schema};
    /// # use serde::Serialize;
    /// #
    /// #[derive(Serialize)]
    /// struct MyStruct {
    ///     int32: i32,
    ///     list: Vec<f64>,
    ///     nested: Vec<Option<Nested>>,
    /// }
    ///
    /// impl MyStruct {
    ///     /// Returns the [`Fields`] for [`MyStruct`]
    ///     fn fields() -> Fields {
    ///         let nested = DataType::Struct(Nested::fields());
    ///         Fields::from([
    ///             Arc::new(Field::new("int32", DataType::Int32, false)),
    ///             Arc::new(Field::new_list(
    ///                 "list",
    ///                 Field::new("element", DataType::Float64, false),
    ///                 false,
    ///             )),
    ///             Arc::new(Field::new_list(
    ///                 "nested",
    ///                 Field::new("element", nested, true),
    ///                 true,
    ///             )),
    ///         ])
    ///     }
    /// }
    ///
    /// #[derive(Serialize)]
    /// struct Nested {
    ///     map: BTreeMap<String, Vec<String>>
    /// }
    ///
    /// impl Nested {
    ///     /// Returns the [`Fields`] for [`Nested`]
    ///     fn fields() -> Fields {
    ///         let element = Field::new("element", DataType::Utf8, false);
    ///         Fields::from([
    ///             Arc::new(Field::new_map(
    ///                 "map",
    ///                 "entries",
    ///                 Field::new("key", DataType::Utf8, false),
    ///                 Field::new_list("value", element, false),
    ///                 false, // sorted
    ///                 false, // nullable
    ///             ))
    ///         ])
    ///     }
    /// }
    ///
    /// let data = vec![
    ///     MyStruct {
    ///         int32: 34,
    ///         list: vec![1., 2., 34.],
    ///         nested: vec![
    ///             None,
    ///             Some(Nested {
    ///                 map: vec![
    ///                     ("key1".to_string(), vec!["foo".to_string(), "bar".to_string()]),
    ///                     ("key2".to_string(), vec!["baz".to_string()])
    ///                 ].into_iter().collect()
    ///             })
    ///         ]
    ///     },
    ///     MyStruct {
    ///         int32: 56,
    ///         list: vec![],
    ///         nested: vec![]
    ///     },
    ///     MyStruct {
    ///         int32: 24,
    ///         list: vec![-1., 245.],
    ///         nested: vec![None]
    ///     }
    /// ];
    ///
    /// let schema = Schema::new(MyStruct::fields());
    /// let mut decoder = ReaderBuilder::new(Arc::new(schema)).build_decoder().unwrap();
    /// decoder.serialize(&data).unwrap();
    /// let batch = decoder.flush().unwrap().unwrap();
    /// assert_eq!(batch.num_rows(), 3);
    /// assert_eq!(batch.num_columns(), 3);
    ///
    /// // Convert to StructArray to format
    /// let s = StructArray::from(batch);
    /// let options = FormatOptions::default().with_null("null");
    /// let formatter = ArrayFormatter::try_new(&s, &options).unwrap();
    ///
    /// assert_eq!(&formatter.value(0).to_string(), "{int32: 34, list: [1.0, 2.0, 34.0], nested: [null, {map: {key1: [foo, bar], key2: [baz]}}]}");
    /// assert_eq!(&formatter.value(1).to_string(), "{int32: 56, list: [], nested: []}");
    /// assert_eq!(&formatter.value(2).to_string(), "{int32: 24, list: [-1.0, 245.0], nested: [null]}");
    /// ```
    ///
    /// Note: this ignores any batch size setting, and always decodes all rows
    ///
    /// [serde]: https://docs.rs/serde/latest/serde/
    pub fn serialize<S: Serialize>(&mut self, rows: &[S]) -> Result<(), ArrowError> {
        self.tape_decoder.serialize(rows)
    }

    /// True if the decoder is currently part way through decoding a record.
    pub fn has_partial_record(&self) -> bool {
        self.tape_decoder.has_partial_row()
    }

    /// The number of unflushed records, including the partially decoded record (if any).
    pub fn len(&self) -> usize {
        self.tape_decoder.num_buffered_rows()
    }

    /// True if there are no records to flush, i.e. [`Self::len`] is zero.
    pub fn is_empty(&self) -> bool {
        self.len() == 0
    }

    /// Flushes the currently buffered data to a [`RecordBatch`]
    ///
    /// Returns `Ok(None)` if no buffered data, i.e. [`Self::is_empty`] is true.
    ///
    /// Note: This will return an error if called part way through decoding a record,
    /// i.e. [`Self::has_partial_record`] is true.
    pub fn flush(&mut self) -> Result<Option<RecordBatch>, ArrowError> {
        let tape = self.tape_decoder.finish()?;

        if tape.num_rows() == 0 {
            return Ok(None);
        }

        // First offset is null sentinel
        let mut next_object = 1;
        let pos = (0..tape.num_rows())
            .map(|_| {
                let next = tape.next(next_object, "row")?;
                Ok(std::mem::replace(&mut next_object, next))
            })
            .collect::<Result<Vec<_>, ArrowError>>()?;

        let decoded = self.decoder.decode(&tape, &pos)?;
        self.tape_decoder.clear();

        let batch = match self.is_field {
            true => RecordBatch::try_new(self.schema.clone(), vec![decoded])?,
            false => {
                RecordBatch::from(decoded.as_struct().clone()).with_schema(self.schema.clone())?
            }
        };

        Ok(Some(batch))
    }
}

/// Decodes a column of JSON values from a [`Tape`] into an [`ArrayRef`]
///
/// Implement alongside [`DecoderFactory`] to override or add support for a type.
pub trait ArrayDecoder: Send {
    /// Decode elements from `tape` starting at the indexes contained in `pos`
    ///
    /// `pos` contains one tape index per output row, so the returned array must have
    /// exactly `pos.len()` elements and the field's data type. A row's value may be
    /// [`TapeElement::Null`].
    fn decode(&mut self, tape: &Tape<'_>, pos: &[u32]) -> Result<ArrayRef, ArrowError>;
}

/// A trait to create custom decoders for specific data types.
///
/// Overrides the reader's decoder for a data type, or adds support for one it does not
/// handle. The reader-side counterpart of [`EncoderFactory`]; register an
/// implementation with [`ReaderBuilder::with_decoder_factory`].
///
/// # Example: Decode `Binary` from a JSON array of integers
///
/// This example shows how to decode a `Binary` column from a JSON array of
/// integers, e.g. `[104, 105]` for the bytes `b"hi"`, as created by the
/// example in [`EncoderFactory`].
///
/// ```
/// # use std::sync::Arc;
/// # use arrow_array::{Array, ArrayRef, BinaryArray};
/// # use arrow_array::types::Float64Type;
/// # use arrow_array::cast::AsArray;
/// # use arrow_json::reader::{ArrayDecoder, DecoderContext, DecoderFactory, Tape, TapeElement};
/// # use arrow_json::ReaderBuilder;
/// # use arrow_schema::{ArrowError, DataType, Field, FieldRef, Fields, Schema};
///
/// /// Decodes `[104, 105]` into the bytes `b"hi"`
/// struct IntArrayBinaryDecoder;
///
/// impl ArrayDecoder for IntArrayBinaryDecoder {
///     fn decode(&mut self, tape: &Tape<'_>, pos: &[u32]) -> Result<ArrayRef, ArrowError> {
///         let mut values: Vec<Option<Vec<u8>>> = Vec::with_capacity(pos.len());
///         for p in pos {
///             match tape.get(*p) {
///                 TapeElement::Null => values.push(None),
///                 TapeElement::StartList(end) => {
///                     let mut bytes = Vec::new();
///                     let mut cur = p + 1;
///                     while cur < end {
///                         match tape.get(cur) {
///                             // JSON text yields `Number`; serde yields `I32`
///                             TapeElement::Number(idx) => {
///                                 let s = tape.get_string(idx);
///                                 bytes.push(s.parse::<u8>().map_err(|e| {
///                                     ArrowError::JsonError(format!("invalid byte {s}: {e}"))
///                                 })?);
///                             }
///                             TapeElement::I32(v) => bytes.push(v as u8),
///                             _ => return Err(tape.error(cur, "byte")),
///                         }
///                         cur = tape.next(cur, "byte")?;
///                     }
///                     values.push(Some(bytes));
///                 }
///                 _ => return Err(tape.error(*p, "list of bytes")),
///             }
///         }
///         Ok(Arc::new(BinaryArray::from_iter(values.iter().map(|v| v.as_deref()))))
///     }
/// }
///
/// /// A factory that overrides the reader's default decoder for `Binary` to decode from an array of integers
/// #[derive(Debug)]
/// struct IntArrayBinaryDecoderFactory;
///
/// impl DecoderFactory for IntArrayBinaryDecoderFactory {
///     fn make_default_decoder(
///         &self,
///         ctx: &DecoderContext,
///         field: &FieldRef,
///         is_nullable: bool,
///     ) -> Result<Option<Box<dyn ArrayDecoder>>, ArrowError> {
///         match field.data_type() {
///             DataType::Binary => Ok(Some(Box::new(IntArrayBinaryDecoder))),
///             // Returning `None` uses the reader's default decoder
///             _ => Ok(None),
///         }
///     }
/// }
///
/// let nested = Fields::from(vec![Field::new("inner", DataType::Binary, true)]);
/// let schema = Arc::new(Schema::new(vec![
///     Field::new("bytes", DataType::Binary, true),
///     Field::new("float", DataType::Float64, true),
///     Field::new("nested", DataType::Struct(nested), true),
/// ]));
///
/// let json = r#"{"bytes": [104, 105], "float": 1.0, "nested": {"inner": [104, 105]}}
/// {"float": 2.3}
/// {"bytes": [98], "nested": {"inner": [98]}}
/// "#;
///
/// let batch = ReaderBuilder::new(schema)
///     .with_decoder_factory(Arc::new(IntArrayBinaryDecoderFactory))
///     .build(json.as_bytes())
///     .unwrap()
///     .next()
///     .unwrap()
///     .unwrap();
///
/// let bytes = batch.column(0).as_binary::<i32>();
/// assert_eq!(bytes.value(0), b"hi");
/// assert!(bytes.is_null(1));
/// assert_eq!(bytes.value(2), b"b");
///
/// // The override applies wherever `Binary` appears, including nested, while types
/// // the factory declines are decoded as usual
/// let inner = batch.column(2).as_struct().column(0).as_binary::<i32>();
/// assert_eq!(inner.value(0), b"hi");
/// assert_eq!(batch.column(1).as_primitive::<Float64Type>().value(0), 1.0);
/// ```
///
/// # Example: Decode an [extension type] from a JSON string
///
/// The choice of decoder can also depend on field metadata rather than just
/// the data type, for example to recognize an [extension type]. The reader's
/// own decoder for the field can be invoked with
/// [`DecoderContext::make_builtin_decoder`].
///
/// This example upper-cases the strings produced by the builtin `Utf8` decoder
/// for a fictional extension type `apache.shout`.
///
/// ```
/// # use std::sync::Arc;
/// # use arrow_array::{ArrayRef, StringArray};
/// # use arrow_array::cast::AsArray;
/// # use arrow_json::reader::{ArrayDecoder, DecoderContext, DecoderFactory, Tape};
/// # use arrow_json::ReaderBuilder;
/// # use arrow_schema::{ArrowError, DataType, Field, FieldRef, Schema};
/// # use arrow_schema::extension::EXTENSION_TYPE_NAME_KEY;
///
/// /// Upper-cases whatever the reader's own decoder produced
/// struct ShoutDecoder(Box<dyn ArrayDecoder>);
///
/// impl ArrayDecoder for ShoutDecoder {
///     fn decode(&mut self, tape: &Tape<'_>, pos: &[u32]) -> Result<ArrayRef, ArrowError> {
///         let inner = self.0.decode(tape, pos)?;
///         let values = inner.as_string::<i32>();
///         Ok(Arc::new(StringArray::from_iter(
///             values.iter().map(|v| v.map(str::to_uppercase)),
///         )))
///     }
/// }
///
/// /// A factory that recognizes the `apache.shout` extension type and upper-cases its strings
/// #[derive(Debug)]
/// struct ShoutDecoderFactory;
///
/// impl DecoderFactory for ShoutDecoderFactory {
///     fn make_default_decoder(
///         &self,
///         ctx: &DecoderContext,
///         field: &FieldRef,
///         is_nullable: bool,
///     ) -> Result<Option<Box<dyn ArrayDecoder>>, ArrowError> {
///         if field.metadata().get(EXTENSION_TYPE_NAME_KEY).map(String::as_str)
///             == Some("apache.shout")
///         {
///             let inner = ctx.make_builtin_decoder(field, is_nullable)?;
///             return Ok(Some(Box::new(ShoutDecoder(inner))));
///         }
///         Ok(None)
///     }
/// }
///
/// let schema = Arc::new(Schema::new(vec![
///     Field::new("shout", DataType::Utf8, true)
///         .with_metadata([(EXTENSION_TYPE_NAME_KEY, "apache.shout")]),
///     Field::new("plain", DataType::Utf8, true),
/// ]));
///
/// let json = r#"{"shout": "hi", "plain": "hi"}"#;
///
/// let batch = ReaderBuilder::new(schema)
///     .with_decoder_factory(Arc::new(ShoutDecoderFactory))
///     .build(json.as_bytes())
///     .unwrap()
///     .next()
///     .unwrap()
///     .unwrap();
///
/// // Dispatched on metadata, and decoded by the reader's own `Utf8` decoder
/// assert_eq!(batch.column(0).as_string::<i32>().value(0), "HI");
/// // The factory declines fields without the extension type, so they are decoded as usual
/// assert_eq!(batch.column(1).as_string::<i32>().value(0), "hi");
/// ```
///
/// [`EncoderFactory`]: crate::EncoderFactory
/// [extension type]: https://arrow.apache.org/docs/format/Columnar.html#extension-types
pub trait DecoderFactory: std::fmt::Debug + Send + Sync {
    /// Make a decoder for `field`, or `Ok(None)` to use the reader's default.
    ///
    /// Receives the [`FieldRef`] rather than just its [`DataType`] so decoder
    /// selection can consider the field's metadata, e.g. to identify [extension
    /// types]. The root of a [`ReaderBuilder::new`] schema is presented as a
    /// synthesized nameless `Struct` field.
    ///
    /// Use [`DecoderContext::make_decoder`] on `ctx` to build child decoders, and
    /// [`DecoderContext::make_builtin_decoder`] to build on the reader's own decoder
    /// for this field. Calling `make_decoder` with the field this was invoked with
    /// recurses back here and loops.
    ///
    /// `is_nullable` folds in ancestor nullability, so it may differ from
    /// `field.is_nullable()` in either direction: a nullable struct widens its
    /// children, while a run-end encoded array narrows its values.
    ///
    /// [extension types]: https://arrow.apache.org/docs/format/Columnar.html#extension-types
    fn make_default_decoder(
        &self,
        _ctx: &DecoderContext,
        _field: &FieldRef,
        _is_nullable: bool,
    ) -> Result<Option<Box<dyn ArrayDecoder>>, ArrowError> {
        Ok(None)
    }
}

/// Validates the output of a [`DecoderFactory`] decoder before it reaches the arrays
/// built from it, some of which are constructed without further checks.
struct CheckedDecoder {
    inner: Box<dyn ArrayDecoder>,
    data_type: DataType,
}

impl ArrayDecoder for CheckedDecoder {
    fn decode(&mut self, tape: &Tape<'_>, pos: &[u32]) -> Result<ArrayRef, ArrowError> {
        let array = self.inner.decode(tape, pos)?;
        if array.data_type() != &self.data_type {
            return Err(ArrowError::JsonError(format!(
                "custom decoder returned {} for a field of type {}",
                array.data_type(),
                self.data_type
            )));
        }
        if array.len() != pos.len() {
            return Err(ArrowError::JsonError(format!(
                "custom decoder returned {} values for {} rows",
                array.len(),
                pos.len()
            )));
        }
        Ok(array)
    }
}

/// Context for decoder creation, containing configuration.
///
/// This context is passed through the decoder creation process and contains
/// all the configuration needed to create decoders recursively.
pub struct DecoderContext {
    /// Whether to coerce primitives to strings
    coerce_primitive: bool,
    /// Whether to validate struct fields strictly
    strict_mode: bool,
    /// How to decode struct fields
    struct_mode: StructMode,
    /// Whether to treat columns with incompatible types as missing (i.e. NULL)
    ignore_type_conflicts: bool,
    /// An optional hook for customizing decoding behavior
    decoder_factory: Option<Arc<dyn DecoderFactory>>,
}

impl DecoderContext {
    /// Returns whether to coerce primitive types (e.g., number to string)
    pub fn coerce_primitive(&self) -> bool {
        self.coerce_primitive
    }

    /// Returns whether to validate struct fields strictly
    pub fn strict_mode(&self) -> bool {
        self.strict_mode
    }

    /// Returns how to decode struct fields
    pub fn struct_mode(&self) -> StructMode {
        self.struct_mode
    }

    /// Returns whether to treat columns with incompatible types as missing (i.e. NULL)
    pub fn ignore_type_conflicts(&self) -> bool {
        self.ignore_type_conflicts
    }

    /// Returns the optional hook for customizing decoding behavior
    pub fn decoder_factory(&self) -> Option<&Arc<dyn DecoderFactory>> {
        self.decoder_factory.as_ref()
    }

    /// Create a decoder for a type.
    ///
    /// This is the standard way to create child decoders from within a decoder
    /// implementation.
    pub fn make_decoder(
        &self,
        field: &FieldRef,
        is_nullable: bool,
    ) -> Result<Box<dyn ArrayDecoder>, ArrowError> {
        make_decoder(self, field, is_nullable)
    }

    /// Create the decoder the reader would use for `field`, ignoring any
    /// [`DecoderFactory`].
    ///
    /// Lets a factory build on the reader's own decoder, including for the field it was
    /// asked about — which [`Self::make_decoder`] cannot do without looping.
    pub fn make_builtin_decoder(
        &self,
        field: &FieldRef,
        is_nullable: bool,
    ) -> Result<Box<dyn ArrayDecoder>, ArrowError> {
        make_builtin_decoder(self, field, is_nullable)
    }
}

fn make_decoder(
    ctx: &DecoderContext,
    field: &FieldRef,
    is_nullable: bool,
) -> Result<Box<dyn ArrayDecoder>, ArrowError> {
    if let Some(factory) = ctx.decoder_factory()
        && let Some(decoder) = factory.make_default_decoder(ctx, field, is_nullable)?
    {
        return Ok(Box::new(CheckedDecoder {
            inner: decoder,
            data_type: field.data_type().clone(),
        }));
    }

    make_builtin_decoder(ctx, field, is_nullable)
}

fn make_builtin_decoder(
    ctx: &DecoderContext,
    field: &FieldRef,
    is_nullable: bool,
) -> Result<Box<dyn ArrayDecoder>, ArrowError> {
    let data_type = field.data_type();

    macro_rules! primitive_decoder {
        ($t:ty, $data_type:expr) => {
            Ok(Box::new(PrimitiveArrayDecoder::<$t>::new(ctx, $data_type)))
        };
    }
    macro_rules! timestamp_decoder {
        ($t:ty, $data_type:expr, $tz:expr) => {{
            Ok(Box::new(TimestampArrayDecoder::<$t, _>::new(
                ctx, $data_type, $tz,
            )))
        }};
    }
    macro_rules! decimal_decoder {
        ($t:ty, $p:expr, $s:expr) => {
            Ok(Box::new(DecimalArrayDecoder::<$t>::new(ctx, $p, $s)))
        };
    }

    downcast_integer! {
        *data_type => (primitive_decoder, data_type),
        DataType::Null => Ok(Box::new(NullArrayDecoder::new(ctx))),
        DataType::Float16 => primitive_decoder!(Float16Type, data_type),
        DataType::Float32 => primitive_decoder!(Float32Type, data_type),
        DataType::Float64 => primitive_decoder!(Float64Type, data_type),
        DataType::Timestamp(TimeUnit::Second, None) => {
            timestamp_decoder!(TimestampSecondType, data_type, Utc)
        },
        DataType::Timestamp(TimeUnit::Millisecond, None) => {
            timestamp_decoder!(TimestampMillisecondType, data_type, Utc)
        },
        DataType::Timestamp(TimeUnit::Microsecond, None) => {
            timestamp_decoder!(TimestampMicrosecondType, data_type, Utc)
        },
        DataType::Timestamp(TimeUnit::Nanosecond, None) => {
            timestamp_decoder!(TimestampNanosecondType, data_type, Utc)
        },
        DataType::Timestamp(TimeUnit::Second, Some(ref tz)) => {
            let tz: Tz = tz.parse()?;
            timestamp_decoder!(TimestampSecondType, data_type, tz)
        },
        DataType::Timestamp(TimeUnit::Millisecond, Some(ref tz)) => {
            let tz: Tz = tz.parse()?;
            timestamp_decoder!(TimestampMillisecondType, data_type, tz)
        },
        DataType::Timestamp(TimeUnit::Microsecond, Some(ref tz)) => {
            let tz: Tz = tz.parse()?;
            timestamp_decoder!(TimestampMicrosecondType, data_type, tz)
        },
        DataType::Timestamp(TimeUnit::Nanosecond, Some(ref tz)) => {
            let tz: Tz = tz.parse()?;
            timestamp_decoder!(TimestampNanosecondType, data_type, tz)
        },
        DataType::Date32 => primitive_decoder!(Date32Type, data_type),
        DataType::Date64 => primitive_decoder!(Date64Type, data_type),
        DataType::Time32(TimeUnit::Second) => primitive_decoder!(Time32SecondType, data_type),
        DataType::Time32(TimeUnit::Millisecond) => primitive_decoder!(Time32MillisecondType, data_type),
        DataType::Time64(TimeUnit::Microsecond) => primitive_decoder!(Time64MicrosecondType, data_type),
        DataType::Time64(TimeUnit::Nanosecond) => primitive_decoder!(Time64NanosecondType, data_type),
        DataType::Duration(TimeUnit::Nanosecond) => primitive_decoder!(DurationNanosecondType, data_type),
        DataType::Duration(TimeUnit::Microsecond) => primitive_decoder!(DurationMicrosecondType, data_type),
        DataType::Duration(TimeUnit::Millisecond) => primitive_decoder!(DurationMillisecondType, data_type),
        DataType::Duration(TimeUnit::Second) => primitive_decoder!(DurationSecondType, data_type),
        DataType::Decimal32(p, s) => decimal_decoder!(Decimal32Type, p, s),
        DataType::Decimal64(p, s) => decimal_decoder!(Decimal64Type, p, s),
        DataType::Decimal128(p, s) => decimal_decoder!(Decimal128Type, p, s),
        DataType::Decimal256(p, s) => decimal_decoder!(Decimal256Type, p, s),
        DataType::Boolean => Ok(Box::new(BooleanArrayDecoder::new(ctx))),
        DataType::Utf8 => Ok(Box::new(StringArrayDecoder::<i32>::new(ctx))),
        DataType::Utf8View => Ok(Box::new(StringViewArrayDecoder::new(ctx))),
        DataType::LargeUtf8 => Ok(Box::new(StringArrayDecoder::<i64>::new(ctx))),
        DataType::List(_) => Ok(Box::new(ListArrayDecoder::<i32>::new(ctx, data_type, is_nullable)?)),
        DataType::LargeList(_) => Ok(Box::new(ListArrayDecoder::<i64>::new(ctx, data_type, is_nullable)?)),
        DataType::ListView(_) => Ok(Box::new(ListViewArrayDecoder::<i32>::new(ctx, data_type, is_nullable)?)),
        DataType::LargeListView(_) => Ok(Box::new(ListViewArrayDecoder::<i64>::new(ctx, data_type, is_nullable)?)),
        DataType::FixedSizeList(_, _) => Ok(Box::new(FixedSizeListArrayDecoder::new(ctx, data_type, is_nullable)?)),
        DataType::Struct(_) => Ok(Box::new(StructArrayDecoder::new(ctx, data_type, is_nullable)?)),
        DataType::Binary => Ok(Box::new(BinaryArrayDecoder::<i32>::default())),
        DataType::LargeBinary => Ok(Box::new(BinaryArrayDecoder::<i64>::default())),
        DataType::FixedSizeBinary(len) => Ok(Box::new(FixedSizeBinaryArrayDecoder::new(len))),
        DataType::BinaryView => Ok(Box::new(BinaryViewDecoder::default())),
        DataType::Map(_, _) => Ok(Box::new(MapArrayDecoder::new(ctx, data_type, is_nullable)?)),
        DataType::RunEndEncoded(ref r, _) => match r.data_type() {
            DataType::Int16 => Ok(Box::new(RunEndEncodedArrayDecoder::<Int16Type>::new(ctx, data_type, is_nullable)?)),
            DataType::Int32 => Ok(Box::new(RunEndEncodedArrayDecoder::<Int32Type>::new(ctx, data_type, is_nullable)?)),
            DataType::Int64 => Ok(Box::new(RunEndEncodedArrayDecoder::<Int64Type>::new(ctx, data_type, is_nullable)?)),
            d => unreachable!("unsupported run end index type: {d}"),
        },
        _ => Err(ArrowError::NotYetImplemented(format!("Support for {data_type} in JSON reader")))
    }
}

#[cfg(test)]
mod tests {
    use arrow_array::cast::AsArray;
    use arrow_array::{
        Array, BooleanArray, Float64Array, GenericListViewArray, Int32Array, ListArray, MapArray,
        NullArray, OffsetSizeTrait, StringArray, StringViewArray, StructArray,
    };
    use arrow_buffer::{ArrowNativeType, NullBuffer, OffsetBuffer, ScalarBuffer};
    use arrow_cast::display::{ArrayFormatter, FormatOptions};
    use arrow_schema::{Field, Fields};
    use serde_json::json;
    use std::fs::File;
    use std::io::{BufReader, Cursor, Seek};
    use std::sync::Mutex;

    use super::*;

    fn do_read(
        buf: &str,
        batch_size: usize,
        coerce_primitive: bool,
        strict_mode: bool,
        schema: SchemaRef,
    ) -> Vec<RecordBatch> {
        let config = ReaderBuilder::new(schema)
            .with_batch_size(batch_size)
            .with_strict_mode(strict_mode)
            .with_coerce_primitive(coerce_primitive);
        do_read_config(buf, config)
    }

    fn do_read_config(buf: &str, builder: ReaderBuilder) -> Vec<RecordBatch> {
        let mut unbuffered = vec![];

        // Test with different batch sizes to test for boundary conditions
        for batch_size in [1, 3, 100, builder.batch_size] {
            unbuffered = builder
                .clone()
                .with_batch_size(batch_size)
                .build(Cursor::new(buf.as_bytes()))
                .unwrap()
                .collect::<Result<Vec<_>, _>>()
                .unwrap();

            for b in unbuffered.iter().take(unbuffered.len() - 1) {
                assert_eq!(b.num_rows(), batch_size)
            }

            // Test with different buffer sizes to test for boundary conditions
            for b in [1, 3, 5] {
                let buffered = builder
                    .clone()
                    .with_batch_size(batch_size)
                    .build(BufReader::with_capacity(b, Cursor::new(buf.as_bytes())))
                    .unwrap()
                    .collect::<Result<Vec<_>, _>>()
                    .unwrap();
                assert_eq!(unbuffered, buffered);
            }
        }

        unbuffered
    }

    #[test]
    fn test_basic() {
        let buf = r#"
        {"a": 1, "b": 2, "c": true, "d": 1}
        {"a": 2E0, "b": 4, "c": false, "d": 2, "e": 254}

        {"b": 6, "a": 2.0, "d": 45}
        {"b": "5", "a": 2}
        {"b": 4e0}
        {"b": 7, "a": null}
        "#;

        let schema = Arc::new(Schema::new(vec![
            Field::new("a", DataType::Int64, true),
            Field::new("b", DataType::Int32, true),
            Field::new("c", DataType::Boolean, true),
            Field::new("d", DataType::Date32, true),
            Field::new("e", DataType::Date64, true),
        ]));

        let mut decoder = ReaderBuilder::new(schema.clone()).build_decoder().unwrap();
        assert!(decoder.is_empty());
        assert_eq!(decoder.len(), 0);
        assert!(!decoder.has_partial_record());
        assert_eq!(decoder.decode(buf.as_bytes()).unwrap(), 221);
        assert!(!decoder.is_empty());
        assert_eq!(decoder.len(), 6);
        assert!(!decoder.has_partial_record());
        let batch = decoder.flush().unwrap().unwrap();
        assert_eq!(batch.num_rows(), 6);
        assert!(decoder.is_empty());
        assert_eq!(decoder.len(), 0);
        assert!(!decoder.has_partial_record());

        let batches = do_read(buf, 1024, false, false, schema);
        assert_eq!(batches.len(), 1);

        let col1 = batches[0].column(0).as_primitive::<Int64Type>();
        assert_eq!(col1.null_count(), 2);
        assert_eq!(col1.values(), &[1, 2, 2, 2, 0, 0]);
        assert!(col1.is_null(4));
        assert!(col1.is_null(5));

        let col2 = batches[0].column(1).as_primitive::<Int32Type>();
        assert_eq!(col2.null_count(), 0);
        assert_eq!(col2.values(), &[2, 4, 6, 5, 4, 7]);

        let col3 = batches[0].column(2).as_boolean();
        assert_eq!(col3.null_count(), 4);
        assert!(col3.value(0));
        assert!(!col3.is_null(0));
        assert!(!col3.value(1));
        assert!(!col3.is_null(1));

        let col4 = batches[0].column(3).as_primitive::<Date32Type>();
        assert_eq!(col4.null_count(), 3);
        assert!(col4.is_null(3));
        assert_eq!(col4.values(), &[1, 2, 45, 0, 0, 0]);

        let col5 = batches[0].column(4).as_primitive::<Date64Type>();
        assert_eq!(col5.null_count(), 5);
        assert!(col5.is_null(0));
        assert!(col5.is_null(2));
        assert!(col5.is_null(3));
        assert_eq!(col5.values(), &[0, 254, 0, 0, 0, 0]);
    }

    #[test]
    fn test_string() {
        let buf = r#"
        {"a": "1", "b": "2"}
        {"a": "hello", "b": "shoo"}
        {"b": "\t😁foo", "a": "\nfoobar\ud83d\ude00\u0061\u0073\u0066\u0067\u00FF"}

        {"b": null}
        {"b": "", "a": null}

        "#;
        let schema = Arc::new(Schema::new(vec![
            Field::new("a", DataType::Utf8, true),
            Field::new("b", DataType::LargeUtf8, true),
        ]));

        let batches = do_read(buf, 1024, false, false, schema);
        assert_eq!(batches.len(), 1);

        let col1 = batches[0].column(0).as_string::<i32>();
        assert_eq!(col1.null_count(), 2);
        assert_eq!(col1.value(0), "1");
        assert_eq!(col1.value(1), "hello");
        assert_eq!(col1.value(2), "\nfoobar😀asfgÿ");
        assert!(col1.is_null(3));
        assert!(col1.is_null(4));

        let col2 = batches[0].column(1).as_string::<i64>();
        assert_eq!(col2.null_count(), 1);
        assert_eq!(col2.value(0), "2");
        assert_eq!(col2.value(1), "shoo");
        assert_eq!(col2.value(2), "\t😁foo");
        assert!(col2.is_null(3));
        assert_eq!(col2.value(4), "");
    }

    #[test]
    fn test_long_string_view_allocation() {
        // The JSON input contains field "a" with different string lengths.
        // According to the implementation in the decoder:
        // - For a string, capacity is only increased if its length > 12 bytes.
        // Therefore, for:
        // Row 1: "short" (5 bytes) -> capacity += 0
        // Row 2: "this is definitely long" (24 bytes) -> capacity += 24
        // Row 3: "hello" (5 bytes) -> capacity += 0
        // Row 4: "\nfoobar😀asfgÿ" (17 bytes) -> capacity += 17
        // Expected total capacity = 24 + 17 = 41
        let expected_capacity: usize = 41;

        let buf = r#"
        {"a": "short", "b": "dummy"}
        {"a": "this is definitely long", "b": "dummy"}
        {"a": "hello", "b": "dummy"}
        {"a": "\nfoobar😀asfgÿ", "b": "dummy"}
        "#;

        let schema = Arc::new(Schema::new(vec![
            Field::new("a", DataType::Utf8View, true),
            Field::new("b", DataType::LargeUtf8, true),
        ]));

        let batches = do_read(buf, 1024, false, false, schema);
        assert_eq!(batches.len(), 1, "Expected one record batch");

        // Get the first column ("a") as a StringViewArray.
        let col_a = batches[0].column(0);
        let string_view_array = col_a
            .as_any()
            .downcast_ref::<StringViewArray>()
            .expect("Column should be a StringViewArray");

        // Retrieve the underlying data buffer from the array.
        // The builder pre-allocates capacity based on the sum of lengths for long strings.
        let data_buffer = string_view_array.to_data().buffers()[0].len();

        // Check that the allocated capacity is at least what we expected.
        // (The actual buffer may be larger than expected due to rounding or internal allocation strategies.)
        assert!(
            data_buffer >= expected_capacity,
            "Data buffer length ({data_buffer}) should be at least {expected_capacity}",
        );

        // Additionally, verify that the decoded values are correct.
        assert_eq!(string_view_array.value(0), "short");
        assert_eq!(string_view_array.value(1), "this is definitely long");
        assert_eq!(string_view_array.value(2), "hello");
        assert_eq!(string_view_array.value(3), "\nfoobar😀asfgÿ");
    }

    /// Test the memory capacity allocation logic when converting numeric types to strings.
    #[test]
    fn test_numeric_view_allocation() {
        // For numeric types, the expected capacity calculation is as follows:
        // Row 1: 123456789  -> Number converts to the string "123456789" (length 9), 9 <= 12, so no capacity is added.
        // Row 2: 1000000000000 -> Treated as an I64 number; its string is "1000000000000" (length 13),
        //                        which is >12 and its absolute value is > 999_999_999_999, so 13 bytes are added.
        // Row 3: 3.1415 -> F32 number, a fixed estimate of 10 bytes is added.
        // Row 4: 2.718281828459045 -> F64 number, a fixed estimate of 10 bytes is added.
        // Total expected capacity = 13 + 10 + 10 = 33 bytes.
        let expected_capacity: usize = 33;

        let buf = r#"
    {"n": 123456789}
    {"n": 1000000000000}
    {"n": 3.1415}
    {"n": 2.718281828459045}
    "#;

        let schema = Arc::new(Schema::new(vec![Field::new("n", DataType::Utf8View, true)]));

        let batches = do_read(buf, 1024, true, false, schema);
        assert_eq!(batches.len(), 1, "Expected one record batch");

        let col_n = batches[0].column(0);
        let string_view_array = col_n
            .as_any()
            .downcast_ref::<StringViewArray>()
            .expect("Column should be a StringViewArray");

        // Check that the underlying data buffer capacity is at least the expected value.
        let data_buffer = string_view_array.to_data().buffers()[0].len();
        assert!(
            data_buffer >= expected_capacity,
            "Data buffer length ({data_buffer}) should be at least {expected_capacity}",
        );

        // Verify that the converted string values are correct.
        // Note: The format of the number converted to a string should match the actual implementation.
        assert_eq!(string_view_array.value(0), "123456789");
        assert_eq!(string_view_array.value(1), "1000000000000");
        assert_eq!(string_view_array.value(2), "3.1415");
        assert_eq!(string_view_array.value(3), "2.718281828459045");
    }

    #[test]
    fn test_string_with_uft8view() {
        let buf = r#"
        {"a": "1", "b": "2"}
        {"a": "hello", "b": "shoo"}
        {"b": "\t😁foo", "a": "\nfoobar\ud83d\ude00\u0061\u0073\u0066\u0067\u00FF"}

        {"b": null}
        {"b": "", "a": null}

        "#;
        let schema = Arc::new(Schema::new(vec![
            Field::new("a", DataType::Utf8View, true),
            Field::new("b", DataType::LargeUtf8, true),
        ]));

        let batches = do_read(buf, 1024, false, false, schema);
        assert_eq!(batches.len(), 1);

        let col1 = batches[0].column(0).as_string_view();
        assert_eq!(col1.null_count(), 2);
        assert_eq!(col1.value(0), "1");
        assert_eq!(col1.value(1), "hello");
        assert_eq!(col1.value(2), "\nfoobar😀asfgÿ");
        assert!(col1.is_null(3));
        assert!(col1.is_null(4));
        assert_eq!(col1.data_type(), &DataType::Utf8View);

        let col2 = batches[0].column(1).as_string::<i64>();
        assert_eq!(col2.null_count(), 1);
        assert_eq!(col2.value(0), "2");
        assert_eq!(col2.value(1), "shoo");
        assert_eq!(col2.value(2), "\t😁foo");
        assert!(col2.is_null(3));
        assert_eq!(col2.value(4), "");
    }

    #[test]
    fn test_complex() {
        let buf = r#"
           {"list": [], "nested": {"a": 1, "b": 2}, "nested_list": {"list2": [{"c": 3}, {"c": 4}]}}
           {"list": [5, 6], "nested": {"a": 7}, "nested_list": {"list2": []}}
           {"list": null, "nested": {"a": null}}
        "#;

        let schema = Arc::new(Schema::new(vec![
            Field::new_list("list", Field::new("element", DataType::Int32, false), true),
            Field::new_struct(
                "nested",
                vec![
                    Field::new("a", DataType::Int32, true),
                    Field::new("b", DataType::Int32, true),
                ],
                true,
            ),
            Field::new_struct(
                "nested_list",
                vec![Field::new_list(
                    "list2",
                    Field::new_struct(
                        "element",
                        vec![Field::new("c", DataType::Int32, false)],
                        false,
                    ),
                    true,
                )],
                true,
            ),
        ]));

        let batches = do_read(buf, 1024, false, false, schema);
        assert_eq!(batches.len(), 1);

        let list = batches[0].column(0).as_list::<i32>();
        assert_eq!(list.len(), 3);
        assert_eq!(list.value_offsets(), &[0, 0, 2, 2]);
        assert_eq!(list.null_count(), 1);
        assert!(list.is_null(2));
        let list_values = list.values().as_primitive::<Int32Type>();
        assert_eq!(list_values.values(), &[5, 6]);

        let nested = batches[0].column(1).as_struct();
        let a = nested.column(0).as_primitive::<Int32Type>();
        assert_eq!(list.null_count(), 1);
        assert_eq!(a.values(), &[1, 7, 0]);
        assert!(list.is_null(2));

        let b = nested.column(1).as_primitive::<Int32Type>();
        assert_eq!(b.null_count(), 2);
        assert_eq!(b.len(), 3);
        assert_eq!(b.value(0), 2);
        assert!(b.is_null(1));
        assert!(b.is_null(2));

        let nested_list = batches[0].column(2).as_struct();
        assert_eq!(nested_list.len(), 3);
        assert_eq!(nested_list.null_count(), 1);
        assert!(nested_list.is_null(2));

        let list2 = nested_list.column(0).as_list::<i32>();
        assert_eq!(list2.len(), 3);
        assert_eq!(list2.null_count(), 1);
        assert_eq!(list2.value_offsets(), &[0, 2, 2, 2]);
        assert!(list2.is_null(2));

        let list2_values = list2.values().as_struct();

        let c = list2_values.column(0).as_primitive::<Int32Type>();
        assert_eq!(c.values(), &[3, 4]);
    }

    #[test]
    fn test_projection() {
        let buf = r#"
           {"list": [], "nested": {"a": 1, "b": 2}, "nested_list": {"list2": [{"c": 3, "d": 5}, {"c": 4}]}}
           {"list": [5, 6], "nested": {"a": 7}, "nested_list": {"list2": []}}
        "#;

        let schema = Arc::new(Schema::new(vec![
            Field::new_struct(
                "nested",
                vec![Field::new("a", DataType::Int32, false)],
                true,
            ),
            Field::new_struct(
                "nested_list",
                vec![Field::new_list(
                    "list2",
                    Field::new_struct(
                        "element",
                        vec![Field::new("d", DataType::Int32, true)],
                        false,
                    ),
                    true,
                )],
                true,
            ),
        ]));

        let batches = do_read(buf, 1024, false, false, schema);
        assert_eq!(batches.len(), 1);

        let nested = batches[0].column(0).as_struct();
        assert_eq!(nested.num_columns(), 1);
        let a = nested.column(0).as_primitive::<Int32Type>();
        assert_eq!(a.null_count(), 0);
        assert_eq!(a.values(), &[1, 7]);

        let nested_list = batches[0].column(1).as_struct();
        assert_eq!(nested_list.num_columns(), 1);
        assert_eq!(nested_list.null_count(), 0);

        let list2 = nested_list.column(0).as_list::<i32>();
        assert_eq!(list2.value_offsets(), &[0, 2, 2]);
        assert_eq!(list2.null_count(), 0);

        let child = list2.values().as_struct();
        assert_eq!(child.num_columns(), 1);
        assert_eq!(child.len(), 2);
        assert_eq!(child.null_count(), 0);

        let c = child.column(0).as_primitive::<Int32Type>();
        assert_eq!(c.values(), &[5, 0]);
        assert_eq!(c.null_count(), 1);
        assert!(c.is_null(1));
    }

    #[test]
    fn test_map() {
        let buf = r#"
           {"map": {"a": ["foo", null]}}
           {"map": {"a": [null], "b": []}}
           {"map": {"c": null, "a": ["baz"]}}
        "#;
        let map = Field::new_map(
            "map",
            Field::MAP_ENTRIES_FIELD_DEFAULT_NAME,
            Field::new(Field::MAP_KEY_FIELD_DEFAULT_NAME, DataType::Utf8, false),
            Field::new_list(
                Field::MAP_VALUE_FIELD_DEFAULT_NAME,
                Field::new("element", DataType::Utf8, true),
                true,
            ),
            false,
            true,
        );

        let schema = Arc::new(Schema::new(vec![map]));

        let batches = do_read(buf, 1024, false, false, schema);
        assert_eq!(batches.len(), 1);

        let map = batches[0].column(0).as_map();
        let map_keys = map.keys().as_string::<i32>();
        let map_values = map.values().as_list::<i32>();
        assert_eq!(map.value_offsets(), &[0, 1, 3, 5]);

        let k: Vec<_> = map_keys.iter().flatten().collect();
        assert_eq!(&k, &["a", "a", "b", "c", "a"]);

        let list_values = map_values.values().as_string::<i32>();
        let lv: Vec<_> = list_values.iter().collect();
        assert_eq!(&lv, &[Some("foo"), None, None, Some("baz")]);
        assert_eq!(map_values.value_offsets(), &[0, 2, 3, 3, 3, 4]);
        assert_eq!(map_values.null_count(), 1);
        assert!(map_values.is_null(3));

        let options = FormatOptions::default().with_null("null");
        let formatter = ArrayFormatter::try_new(map, &options).unwrap();
        assert_eq!(formatter.value(0).to_string(), "{a: [foo, null]}");
        assert_eq!(formatter.value(1).to_string(), "{a: [null], b: []}");
        assert_eq!(formatter.value(2).to_string(), "{c: null, a: [baz]}");
    }

    #[test]
    fn test_map_non_nullable_value() {
        let map = Field::new_map(
            "map",
            Field::MAP_ENTRIES_FIELD_DEFAULT_NAME,
            Field::new(Field::MAP_KEY_FIELD_DEFAULT_NAME, DataType::Utf8, false),
            Field::new(Field::MAP_VALUE_FIELD_DEFAULT_NAME, DataType::Utf8, false),
            false,
            false,
        );
        let schema = Arc::new(Schema::new(vec![map]));
        let buf = r#"{"map": {"key": null}}"#;

        let err = ReaderBuilder::new(schema)
            .build(Cursor::new(buf.as_bytes()))
            .unwrap()
            .read()
            .unwrap_err();

        assert_eq!(
            err.to_string(),
            "Invalid argument error: Found unmasked nulls for non-nullable StructArray field \"value\""
        );
    }

    #[test]
    fn test_not_coercing_primitive_into_string_without_flag() {
        let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Utf8, true)]));

        let buf = r#"{"a": 1}"#;
        let err = ReaderBuilder::new(schema.clone())
            .with_batch_size(1024)
            .build(Cursor::new(buf.as_bytes()))
            .unwrap()
            .read()
            .unwrap_err();

        assert_eq!(
            err.to_string(),
            "Json error: whilst decoding field 'a': expected string got 1"
        );

        let buf = r#"{"a": true}"#;
        let err = ReaderBuilder::new(schema)
            .with_batch_size(1024)
            .build(Cursor::new(buf.as_bytes()))
            .unwrap()
            .read()
            .unwrap_err();

        assert_eq!(
            err.to_string(),
            "Json error: whilst decoding field 'a': expected string got true"
        );
    }

    #[test]
    fn test_coercing_primitive_into_string() {
        let buf = r#"
        {"a": 1, "b": 2, "c": true}
        {"a": 2E0, "b": 4, "c": false}

        {"b": 6, "a": 2.0}
        {"b": "5", "a": 2}
        {"b": 4e0}
        {"b": 7, "a": null}
        "#;

        let schema = Arc::new(Schema::new(vec![
            Field::new("a", DataType::Utf8, true),
            Field::new("b", DataType::Utf8, true),
            Field::new("c", DataType::Utf8, true),
        ]));

        let batches = do_read(buf, 1024, true, false, schema);
        assert_eq!(batches.len(), 1);

        let col1 = batches[0].column(0).as_string::<i32>();
        assert_eq!(col1.null_count(), 2);
        assert_eq!(col1.value(0), "1");
        assert_eq!(col1.value(1), "2E0");
        assert_eq!(col1.value(2), "2.0");
        assert_eq!(col1.value(3), "2");
        assert!(col1.is_null(4));
        assert!(col1.is_null(5));

        let col2 = batches[0].column(1).as_string::<i32>();
        assert_eq!(col2.null_count(), 0);
        assert_eq!(col2.value(0), "2");
        assert_eq!(col2.value(1), "4");
        assert_eq!(col2.value(2), "6");
        assert_eq!(col2.value(3), "5");
        assert_eq!(col2.value(4), "4e0");
        assert_eq!(col2.value(5), "7");

        let col3 = batches[0].column(2).as_string::<i32>();
        assert_eq!(col3.null_count(), 4);
        assert_eq!(col3.value(0), "true");
        assert_eq!(col3.value(1), "false");
        assert!(col3.is_null(2));
        assert!(col3.is_null(3));
        assert!(col3.is_null(4));
        assert!(col3.is_null(5));
    }

    fn test_decimal<T: DecimalType>(data_type: DataType) {
        let buf = r#"
        {"a": 1, "b": 2, "c": 38.30}
        {"a": 2, "b": 4, "c": 123.456}

        {"b": 1337, "a": "2.0452"}
        {"b": "5", "a": "11034.2"}
        {"b": 40}
        {"b": 1234, "a": null}
        "#;

        let schema = Arc::new(Schema::new(vec![
            Field::new("a", data_type.clone(), true),
            Field::new("b", data_type.clone(), true),
            Field::new("c", data_type, true),
        ]));

        let batches = do_read(buf, 1024, true, false, schema);
        assert_eq!(batches.len(), 1);

        let col1 = batches[0].column(0).as_primitive::<T>();
        assert_eq!(col1.null_count(), 2);
        assert!(col1.is_null(4));
        assert!(col1.is_null(5));
        assert_eq!(
            col1.values(),
            &[100, 200, 205, 1103420, 0, 0].map(T::Native::usize_as)
        );

        let col2 = batches[0].column(1).as_primitive::<T>();
        assert_eq!(col2.null_count(), 0);
        assert_eq!(
            col2.values(),
            &[200, 400, 133700, 500, 4000, 123400].map(T::Native::usize_as)
        );

        let col3 = batches[0].column(2).as_primitive::<T>();
        assert_eq!(col3.null_count(), 4);
        assert!(!col3.is_null(0));
        assert!(!col3.is_null(1));
        assert!(col3.is_null(2));
        assert!(col3.is_null(3));
        assert!(col3.is_null(4));
        assert!(col3.is_null(5));
        assert_eq!(
            col3.values(),
            &[3830, 12346, 0, 0, 0, 0].map(T::Native::usize_as)
        );
    }

    #[test]
    fn test_decimal_number_formats() {
        // Rounding half away from zero, exponent notation (a valid JSON
        // number syntax), surrounding whitespace in strings and negative
        // scales are all accepted
        let buf = r#"
        {"a": 0e0, "b": " 1.5 ", "c": 1234.5}
        {"a": 1.5E2, "b": "-0.005", "c": -150}
        {"a": 1e-3, "b": "1e2", "c": 1E2}
        {"a": -0E+0, "b": "+.5", "c": 5}
        "#;
        let schema = Arc::new(Schema::new(vec![
            Field::new("a", DataType::Decimal128(10, 2), true),
            Field::new("b", DataType::Decimal64(18, 2), true),
            Field::new("c", DataType::Decimal128(10, -2), true),
        ]));
        let batches = do_read(buf, 1024, true, false, schema);
        assert_eq!(batches.len(), 1);
        assert_eq!(
            batches[0]
                .column(0)
                .as_primitive::<Decimal128Type>()
                .values(),
            &[0, 15000, 0, 0]
        );
        assert_eq!(
            batches[0]
                .column(1)
                .as_primitive::<Decimal64Type>()
                .values(),
            &[150, -1, 10000, 50]
        );
        assert_eq!(
            batches[0]
                .column(2)
                .as_primitive::<Decimal128Type>()
                .values(),
            &[12, -2, 1, 0]
        );

        // Invalid and out-of-range values are errors (or nulls when type
        // conflicts are ignored), never panics
        let schema = Arc::new(Schema::new(vec![Field::new(
            "a",
            DataType::Decimal128(5, 2),
            true,
        )]));
        let long = "1".repeat(300);
        for (buf, expected) in [
            (
                r#"{"a": "abc"}"#.to_string(),
                "Invalid decimal format: \"abc\"",
            ),
            (
                r#"{"a": 123456789}"#.to_string(),
                "does not fit in Decimal128(5, 2)",
            ),
            (
                r#"{"a": 1e99999}"#.to_string(),
                "does not fit in Decimal128(5, 2)",
            ),
            (
                format!(r#"{{"a": {long}}}"#),
                "does not fit in Decimal128(5, 2)",
            ),
            (
                r#"{"a": 4825037936439135476.2609835314269495255615E-14}"#.to_string(),
                "does not fit in Decimal128(5, 2)",
            ),
        ] {
            let err = ReaderBuilder::new(schema.clone())
                .build(Cursor::new(buf.as_bytes()))
                .unwrap()
                .next()
                .unwrap()
                .unwrap_err()
                .to_string();
            assert!(err.contains(expected), "{buf}: {err}");

            let batch = ReaderBuilder::new(schema.clone())
                .with_ignore_type_conflicts(true)
                .build(Cursor::new(buf.as_bytes()))
                .unwrap()
                .next()
                .unwrap()
                .unwrap();
            assert!(batch.column(0).is_null(0), "{buf}");
        }
    }

    #[test]
    fn test_decimals() {
        test_decimal::<Decimal32Type>(DataType::Decimal32(8, 2));
        test_decimal::<Decimal64Type>(DataType::Decimal64(10, 2));
        test_decimal::<Decimal128Type>(DataType::Decimal128(10, 2));
        test_decimal::<Decimal256Type>(DataType::Decimal256(10, 2));
    }

    fn test_timestamp<T: ArrowTimestampType>() {
        let buf = r#"
        {"a": 1, "b": "2020-09-08T13:42:29.190855+00:00", "c": 38.30, "d": "1997-01-31T09:26:56.123"}
        {"a": 2, "b": "2020-09-08T13:42:29.190855Z", "c": 123.456, "d": 123.456}

        {"b": 1337, "b": "2020-09-08T13:42:29Z", "c": "1997-01-31T09:26:56.123", "d": "1997-01-31T09:26:56.123Z"}
        {"b": 40, "c": "2020-09-08T13:42:29.190855+00:00", "d": "1997-01-31 09:26:56.123-05:00"}
        {"b": 1234, "a": null, "c": "1997-01-31 09:26:56.123Z", "d": "1997-01-31 092656"}
        {"c": "1997-01-31T14:26:56.123-05:00", "d": "1997-01-31"}
        "#;

        let with_timezone = DataType::Timestamp(T::UNIT, Some("+08:00".into()));
        let schema = Arc::new(Schema::new(vec![
            Field::new("a", T::DATA_TYPE, true),
            Field::new("b", T::DATA_TYPE, true),
            Field::new("c", T::DATA_TYPE, true),
            Field::new("d", with_timezone, true),
        ]));

        let batches = do_read(buf, 1024, true, false, schema);
        assert_eq!(batches.len(), 1);

        let unit_in_nanos: i64 = match T::UNIT {
            TimeUnit::Second => 1_000_000_000,
            TimeUnit::Millisecond => 1_000_000,
            TimeUnit::Microsecond => 1_000,
            TimeUnit::Nanosecond => 1,
        };

        let col1 = batches[0].column(0).as_primitive::<T>();
        assert_eq!(col1.null_count(), 4);
        assert!(col1.is_null(2));
        assert!(col1.is_null(3));
        assert!(col1.is_null(4));
        assert!(col1.is_null(5));
        assert_eq!(col1.values(), &[1, 2, 0, 0, 0, 0].map(T::Native::usize_as));

        let col2 = batches[0].column(1).as_primitive::<T>();
        assert_eq!(col2.null_count(), 1);
        assert!(col2.is_null(5));
        assert_eq!(
            col2.values(),
            &[
                1599572549190855000 / unit_in_nanos,
                1599572549190855000 / unit_in_nanos,
                1599572549000000000 / unit_in_nanos,
                40,
                1234,
                0
            ]
        );

        let col3 = batches[0].column(2).as_primitive::<T>();
        assert_eq!(col3.null_count(), 0);
        assert_eq!(
            col3.values(),
            &[
                38,
                123,
                854702816123000000 / unit_in_nanos,
                1599572549190855000 / unit_in_nanos,
                854702816123000000 / unit_in_nanos,
                854738816123000000 / unit_in_nanos
            ]
        );

        let col4 = batches[0].column(3).as_primitive::<T>();

        assert_eq!(col4.null_count(), 0);
        assert_eq!(
            col4.values(),
            &[
                854674016123000000 / unit_in_nanos,
                123,
                854702816123000000 / unit_in_nanos,
                854720816123000000 / unit_in_nanos,
                854674016000000000 / unit_in_nanos,
                854640000000000000 / unit_in_nanos
            ]
        );
    }

    #[test]
    #[cfg_attr(miri, ignore)] // Takes too long
    fn test_timestamps() {
        test_timestamp::<TimestampSecondType>();
        test_timestamp::<TimestampMillisecondType>();
        test_timestamp::<TimestampMicrosecondType>();
        test_timestamp::<TimestampNanosecondType>();
    }

    // A local time in a DST gap or overlap of the target timezone is resolved
    // as described in `arrow_cast::parse::string_to_datetime`: an ambiguous
    // reading takes the later instant, and a nonexistent reading is shifted
    // forward by the length of the gap.
    #[test]
    fn test_timestamp_dst_transitions() {
        let buf = r#"
        {"a": "2024-03-10 02:30:00"}
        {"a": "2024-11-03 01:30:00"}
        "#;
        let data_type = DataType::Timestamp(TimeUnit::Second, Some("America/New_York".into()));
        let schema = Arc::new(Schema::new(vec![Field::new("a", data_type, true)]));

        let batches = do_read(buf, 1024, true, false, schema);
        assert_eq!(batches.len(), 1);

        let col = batches[0].column(0).as_primitive::<TimestampSecondType>();
        assert_eq!(col.null_count(), 0);
        // Nonexistent 02:30 is read as 03:30 EDT (2024-03-10T07:30:00Z), and
        // ambiguous 01:30 takes the later EST instant (2024-11-03T06:30:00Z)
        assert_eq!(col.values(), &[1_710_055_800, 1_730_615_400]);
    }

    fn test_time<T: ArrowTemporalType>() {
        let buf = r#"
        {"a": 1, "b": "09:26:56.123 AM", "c": 38.30}
        {"a": 2, "b": "23:59:59", "c": 123.456}

        {"b": 1337, "b": "6:00 pm", "c": "09:26:56.123"}
        {"b": 40, "c": "13:42:29.190855"}
        {"b": 1234, "a": null, "c": "09:26:56.123"}
        {"c": "14:26:56.123"}
        "#;

        let (DataType::Time32(unit) | DataType::Time64(unit)) = T::DATA_TYPE else {
            unreachable!()
        };

        let unit_in_nanos = match unit {
            TimeUnit::Second => 1_000_000_000,
            TimeUnit::Millisecond => 1_000_000,
            TimeUnit::Microsecond => 1_000,
            TimeUnit::Nanosecond => 1,
        };

        let schema = Arc::new(Schema::new(vec![
            Field::new("a", T::DATA_TYPE, true),
            Field::new("b", T::DATA_TYPE, true),
            Field::new("c", T::DATA_TYPE, true),
        ]));

        let batches = do_read(buf, 1024, true, false, schema);
        assert_eq!(batches.len(), 1);

        let col1 = batches[0].column(0).as_primitive::<T>();
        assert_eq!(col1.null_count(), 4);
        assert!(col1.is_null(2));
        assert!(col1.is_null(3));
        assert!(col1.is_null(4));
        assert!(col1.is_null(5));
        assert_eq!(col1.values(), &[1, 2, 0, 0, 0, 0].map(T::Native::usize_as));

        let col2 = batches[0].column(1).as_primitive::<T>();
        assert_eq!(col2.null_count(), 1);
        assert!(col2.is_null(5));
        assert_eq!(
            col2.values(),
            &[
                34016123000000 / unit_in_nanos,
                86399000000000 / unit_in_nanos,
                64800000000000 / unit_in_nanos,
                40,
                1234,
                0
            ]
            .map(T::Native::usize_as)
        );

        let col3 = batches[0].column(2).as_primitive::<T>();
        assert_eq!(col3.null_count(), 0);
        assert_eq!(
            col3.values(),
            &[
                38,
                123,
                34016123000000 / unit_in_nanos,
                49349190855000 / unit_in_nanos,
                34016123000000 / unit_in_nanos,
                52016123000000 / unit_in_nanos
            ]
            .map(T::Native::usize_as)
        );
    }

    #[test]
    fn test_times() {
        test_time::<Time32MillisecondType>();
        test_time::<Time32SecondType>();
        test_time::<Time64MicrosecondType>();
        test_time::<Time64NanosecondType>();
    }

    fn test_duration<T: ArrowTemporalType>() {
        let buf = r#"
        {"a": 1, "b": "2"}
        {"a": 3, "b": null}
        "#;

        let schema = Arc::new(Schema::new(vec![
            Field::new("a", T::DATA_TYPE, true),
            Field::new("b", T::DATA_TYPE, true),
        ]));

        let batches = do_read(buf, 1024, true, false, schema);
        assert_eq!(batches.len(), 1);

        let col_a = batches[0].column_by_name("a").unwrap().as_primitive::<T>();
        assert_eq!(col_a.null_count(), 0);
        assert_eq!(col_a.values(), &[1, 3].map(T::Native::usize_as));

        let col2 = batches[0].column_by_name("b").unwrap().as_primitive::<T>();
        assert_eq!(col2.null_count(), 1);
        assert_eq!(col2.values(), &[2, 0].map(T::Native::usize_as));
    }

    #[test]
    fn test_durations() {
        test_duration::<DurationNanosecondType>();
        test_duration::<DurationMicrosecondType>();
        test_duration::<DurationMillisecondType>();
        test_duration::<DurationSecondType>();
    }

    #[test]
    fn test_delta_checkpoint() {
        let json = "{\"protocol\":{\"minReaderVersion\":1,\"minWriterVersion\":2}}";
        let schema = Arc::new(Schema::new(vec![
            Field::new_struct(
                "protocol",
                vec![
                    Field::new("minReaderVersion", DataType::Int32, true),
                    Field::new("minWriterVersion", DataType::Int32, true),
                ],
                true,
            ),
            Field::new_struct(
                "add",
                vec![Field::new_map(
                    "partitionValues",
                    "key_value",
                    Field::new("key", DataType::Utf8, false),
                    Field::new("value", DataType::Utf8, true),
                    false,
                    false,
                )],
                true,
            ),
        ]));

        let batches = do_read(json, 1024, true, false, schema);
        assert_eq!(batches.len(), 1);

        let s: StructArray = batches.into_iter().next().unwrap().into();
        let opts = FormatOptions::default().with_null("null");
        let formatter = ArrayFormatter::try_new(&s, &opts).unwrap();
        assert_eq!(
            formatter.value(0).to_string(),
            "{protocol: {minReaderVersion: 1, minWriterVersion: 2}, add: null}"
        );
    }

    #[test]
    fn struct_nullability() {
        let do_test = |child: DataType| {
            // Test correctly enforced nullability
            let non_null = r#"{"foo": {}}"#;
            let schema = Arc::new(Schema::new(vec![Field::new_struct(
                "foo",
                vec![Field::new("bar", child, false)],
                true,
            )]));
            let mut reader = ReaderBuilder::new(schema.clone())
                .build(Cursor::new(non_null.as_bytes()))
                .unwrap();
            assert!(reader.next().unwrap().is_err()); // Should error as not nullable

            let null = r#"{"foo": {bar: null}}"#;
            let mut reader = ReaderBuilder::new(schema.clone())
                .build(Cursor::new(null.as_bytes()))
                .unwrap();
            assert!(reader.next().unwrap().is_err()); // Should error as not nullable

            // Test nulls in nullable parent can mask nulls in non-nullable child
            let null = r#"{"foo": null}"#;
            let mut reader = ReaderBuilder::new(schema)
                .build(Cursor::new(null.as_bytes()))
                .unwrap();
            let batch = reader.next().unwrap().unwrap();
            assert_eq!(batch.num_columns(), 1);
            let foo = batch.column(0).as_struct();
            assert_eq!(foo.len(), 1);
            assert!(foo.is_null(0));
            assert_eq!(foo.num_columns(), 1);

            let bar = foo.column(0);
            assert_eq!(bar.len(), 1);
            // Non-nullable child can still contain null as masked by parent
            assert!(bar.is_null(0));
        };

        do_test(DataType::Boolean);
        do_test(DataType::Int32);
        do_test(DataType::Utf8);
        do_test(DataType::Decimal128(2, 1));
        do_test(DataType::Timestamp(
            TimeUnit::Microsecond,
            Some("+00:00".into()),
        ));
    }

    #[test]
    fn test_truncation() {
        let buf = r#"
        {"i64": 9223372036854775807, "u64": 18446744073709551615 }
        {"i64": "9223372036854775807", "u64": "18446744073709551615" }
        {"i64": -9223372036854775808, "u64": 0 }
        {"i64": "-9223372036854775808", "u64": 0 }
        "#;

        let schema = Arc::new(Schema::new(vec![
            Field::new("i64", DataType::Int64, true),
            Field::new("u64", DataType::UInt64, true),
        ]));

        let batches = do_read(buf, 1024, true, false, schema);
        assert_eq!(batches.len(), 1);

        let i64 = batches[0].column(0).as_primitive::<Int64Type>();
        assert_eq!(i64.values(), &[i64::MAX, i64::MAX, i64::MIN, i64::MIN]);

        let u64 = batches[0].column(1).as_primitive::<UInt64Type>();
        assert_eq!(u64.values(), &[u64::MAX, u64::MAX, u64::MIN, u64::MIN]);
    }

    #[test]
    fn test_timestamp_truncation() {
        let buf = r#"
        {"time": 9223372036854775807 }
        {"time": -9223372036854775808 }
        {"time": 9e5 }
        "#;

        let schema = Arc::new(Schema::new(vec![Field::new(
            "time",
            DataType::Timestamp(TimeUnit::Nanosecond, None),
            true,
        )]));

        let batches = do_read(buf, 1024, true, false, schema);
        assert_eq!(batches.len(), 1);

        let i64 = batches[0]
            .column(0)
            .as_primitive::<TimestampNanosecondType>();
        assert_eq!(i64.values(), &[i64::MAX, i64::MIN, 900000]);
    }

    #[test]
    fn test_timestamp_float_out_of_range() {
        let schema = Arc::new(Schema::new(vec![Field::new(
            "time",
            DataType::Timestamp(TimeUnit::Second, None),
            true,
        )]));

        // -2^63 is the smallest representable value
        let batch = ReaderBuilder::new(schema.clone())
            .build(Cursor::new(
                r#"{"time": -9.223372036854775808e18}"#.as_bytes(),
            ))
            .unwrap()
            .next()
            .unwrap()
            .unwrap();
        let col = batch.column(0).as_primitive::<TimestampSecondType>();
        assert_eq!(col.value(0), i64::MIN);

        for buf in [
            r#"{"time": 1e30}"#,
            r#"{"time": -1e30}"#,
            r#"{"time": 9.223372036854775808e18}"#,
            r#"{"time": 1e999}"#,
        ] {
            let err = ReaderBuilder::new(schema.clone())
                .build(Cursor::new(buf.as_bytes()))
                .unwrap()
                .next()
                .unwrap()
                .unwrap_err()
                .to_string();
            assert!(err.contains("failed to parse"), "{buf}: {err}");

            let batch = ReaderBuilder::new(schema.clone())
                .with_ignore_type_conflicts(true)
                .build(Cursor::new(buf.as_bytes()))
                .unwrap()
                .next()
                .unwrap()
                .unwrap();
            assert!(batch.column(0).is_null(0), "{buf}");
        }
    }

    #[test]
    fn test_strict_mode_no_missing_columns_in_schema() {
        let buf = r#"
        {"a": 1, "b": "2", "c": true}
        {"a": 2E0, "b": "4", "c": false}
        "#;

        let schema = Arc::new(Schema::new(vec![
            Field::new("a", DataType::Int16, false),
            Field::new("b", DataType::Utf8, false),
            Field::new("c", DataType::Boolean, false),
        ]));

        let batches = do_read(buf, 1024, true, true, schema);
        assert_eq!(batches.len(), 1);

        let buf = r#"
        {"a": 1, "b": "2", "c": {"a": true, "b": 1}}
        {"a": 2E0, "b": "4", "c": {"a": false, "b": 2}}
        "#;

        let schema = Arc::new(Schema::new(vec![
            Field::new("a", DataType::Int16, false),
            Field::new("b", DataType::Utf8, false),
            Field::new_struct(
                "c",
                vec![
                    Field::new("a", DataType::Boolean, false),
                    Field::new("b", DataType::Int16, false),
                ],
                false,
            ),
        ]));

        let batches = do_read(buf, 1024, true, true, schema);
        assert_eq!(batches.len(), 1);
    }

    #[test]
    fn test_strict_mode_missing_columns_in_schema() {
        let buf = r#"
        {"a": 1, "b": "2", "c": true}
        {"a": 2E0, "b": "4", "c": false}
        "#;

        let schema = Arc::new(Schema::new(vec![
            Field::new("a", DataType::Int16, true),
            Field::new("c", DataType::Boolean, true),
        ]));

        let err = ReaderBuilder::new(schema)
            .with_batch_size(1024)
            .with_strict_mode(true)
            .build(Cursor::new(buf.as_bytes()))
            .unwrap()
            .read()
            .unwrap_err();

        assert_eq!(
            err.to_string(),
            "Json error: column 'b' missing from schema"
        );

        let buf = r#"
        {"a": 1, "b": "2", "c": {"a": true, "b": 1}}
        {"a": 2E0, "b": "4", "c": {"a": false, "b": 2}}
        "#;

        let schema = Arc::new(Schema::new(vec![
            Field::new("a", DataType::Int16, false),
            Field::new("b", DataType::Utf8, false),
            Field::new_struct("c", vec![Field::new("a", DataType::Boolean, false)], false),
        ]));

        let err = ReaderBuilder::new(schema)
            .with_batch_size(1024)
            .with_strict_mode(true)
            .build(Cursor::new(buf.as_bytes()))
            .unwrap()
            .read()
            .unwrap_err();

        assert_eq!(
            err.to_string(),
            "Json error: whilst decoding field 'c': column 'b' missing from schema"
        );
    }

    fn read_file(path: &str, schema: Option<Schema>) -> Reader<BufReader<File>> {
        let file = File::open(path).unwrap();
        let mut reader = BufReader::new(file);
        let schema = schema.unwrap_or_else(|| {
            let (schema, _) = infer_json_schema(&mut reader, None).unwrap();
            reader.rewind().unwrap();
            schema
        });
        let builder = ReaderBuilder::new(Arc::new(schema)).with_batch_size(64);
        builder.build(reader).unwrap()
    }

    #[test]
    fn test_json_basic() {
        let mut reader = read_file("test/data/basic.json", None);
        let batch = reader.next().unwrap().unwrap();

        assert_eq!(8, batch.num_columns());
        assert_eq!(12, batch.num_rows());

        let schema = reader.schema();
        let batch_schema = batch.schema();
        assert_eq!(schema, batch_schema);

        let a = schema.column_with_name("a").unwrap();
        assert_eq!(0, a.0);
        assert_eq!(&DataType::Int64, a.1.data_type());
        let b = schema.column_with_name("b").unwrap();
        assert_eq!(1, b.0);
        assert_eq!(&DataType::Float64, b.1.data_type());
        let c = schema.column_with_name("c").unwrap();
        assert_eq!(2, c.0);
        assert_eq!(&DataType::Boolean, c.1.data_type());
        let d = schema.column_with_name("d").unwrap();
        assert_eq!(3, d.0);
        assert_eq!(&DataType::Utf8, d.1.data_type());

        let aa = batch.column(a.0).as_primitive::<Int64Type>();
        assert_eq!(1, aa.value(0));
        assert_eq!(-10, aa.value(1));
        let bb = batch.column(b.0).as_primitive::<Float64Type>();
        assert_eq!(2.0, bb.value(0));
        assert_eq!(-3.5, bb.value(1));
        let cc = batch.column(c.0).as_boolean();
        assert!(!cc.value(0));
        assert!(cc.value(10));
        let dd = batch.column(d.0).as_string::<i32>();
        assert_eq!("4", dd.value(0));
        assert_eq!("text", dd.value(8));
    }

    #[test]
    fn test_json_empty_projection() {
        let mut reader = read_file("test/data/basic.json", Some(Schema::empty()));
        let batch = reader.next().unwrap().unwrap();

        assert_eq!(0, batch.num_columns());
        assert_eq!(12, batch.num_rows());
    }

    #[test]
    fn test_json_basic_with_nulls() {
        let mut reader = read_file("test/data/basic_nulls.json", None);
        let batch = reader.next().unwrap().unwrap();

        assert_eq!(4, batch.num_columns());
        assert_eq!(12, batch.num_rows());

        let schema = reader.schema();
        let batch_schema = batch.schema();
        assert_eq!(schema, batch_schema);

        let a = schema.column_with_name("a").unwrap();
        assert_eq!(&DataType::Int64, a.1.data_type());
        let b = schema.column_with_name("b").unwrap();
        assert_eq!(&DataType::Float64, b.1.data_type());
        let c = schema.column_with_name("c").unwrap();
        assert_eq!(&DataType::Boolean, c.1.data_type());
        let d = schema.column_with_name("d").unwrap();
        assert_eq!(&DataType::Utf8, d.1.data_type());

        let aa = batch.column(a.0).as_primitive::<Int64Type>();
        assert!(aa.is_valid(0));
        assert!(!aa.is_valid(1));
        assert!(!aa.is_valid(11));
        let bb = batch.column(b.0).as_primitive::<Float64Type>();
        assert!(bb.is_valid(0));
        assert!(!bb.is_valid(2));
        assert!(!bb.is_valid(11));
        let cc = batch.column(c.0).as_boolean();
        assert!(cc.is_valid(0));
        assert!(!cc.is_valid(4));
        assert!(!cc.is_valid(11));
        let dd = batch.column(d.0).as_string::<i32>();
        assert!(!dd.is_valid(0));
        assert!(dd.is_valid(1));
        assert!(!dd.is_valid(4));
        assert!(!dd.is_valid(11));
    }

    #[test]
    fn test_json_basic_schema() {
        let schema = Schema::new(vec![
            Field::new("a", DataType::Int64, true),
            Field::new("b", DataType::Float32, false),
            Field::new("c", DataType::Boolean, false),
            Field::new("d", DataType::Utf8, false),
        ]);

        let mut reader = read_file("test/data/basic.json", Some(schema.clone()));
        let reader_schema = reader.schema();
        assert_eq!(reader_schema.as_ref(), &schema);
        let batch = reader.next().unwrap().unwrap();

        assert_eq!(4, batch.num_columns());
        assert_eq!(12, batch.num_rows());

        let schema = batch.schema();

        let a = schema.column_with_name("a").unwrap();
        assert_eq!(&DataType::Int64, a.1.data_type());
        let b = schema.column_with_name("b").unwrap();
        assert_eq!(&DataType::Float32, b.1.data_type());
        let c = schema.column_with_name("c").unwrap();
        assert_eq!(&DataType::Boolean, c.1.data_type());
        let d = schema.column_with_name("d").unwrap();
        assert_eq!(&DataType::Utf8, d.1.data_type());

        let aa = batch.column(a.0).as_primitive::<Int64Type>();
        assert_eq!(1, aa.value(0));
        assert_eq!(100000000000000, aa.value(11));
        let bb = batch.column(b.0).as_primitive::<Float32Type>();
        assert_eq!(2.0, bb.value(0));
        assert_eq!(-3.5, bb.value(1));
    }

    #[test]
    fn test_json_basic_schema_projection() {
        let schema = Schema::new(vec![
            Field::new("a", DataType::Int64, true),
            Field::new("c", DataType::Boolean, false),
        ]);

        let mut reader = read_file("test/data/basic.json", Some(schema.clone()));
        let batch = reader.next().unwrap().unwrap();

        assert_eq!(2, batch.num_columns());
        assert_eq!(2, batch.schema().fields().len());
        assert_eq!(12, batch.num_rows());

        assert_eq!(batch.schema().as_ref(), &schema);

        let a = schema.column_with_name("a").unwrap();
        assert_eq!(0, a.0);
        assert_eq!(&DataType::Int64, a.1.data_type());
        let c = schema.column_with_name("c").unwrap();
        assert_eq!(1, c.0);
        assert_eq!(&DataType::Boolean, c.1.data_type());
    }

    #[test]
    fn test_json_arrays() {
        let mut reader = read_file("test/data/arrays.json", None);
        let batch = reader.next().unwrap().unwrap();

        assert_eq!(4, batch.num_columns());
        assert_eq!(3, batch.num_rows());

        let schema = batch.schema();

        let a = schema.column_with_name("a").unwrap();
        assert_eq!(&DataType::Int64, a.1.data_type());
        let b = schema.column_with_name("b").unwrap();
        assert_eq!(
            &DataType::List(Arc::new(Field::new_list_field(DataType::Float64, true))),
            b.1.data_type()
        );
        let c = schema.column_with_name("c").unwrap();
        assert_eq!(
            &DataType::List(Arc::new(Field::new_list_field(DataType::Boolean, true))),
            c.1.data_type()
        );
        let d = schema.column_with_name("d").unwrap();
        assert_eq!(&DataType::Utf8, d.1.data_type());

        let aa = batch.column(a.0).as_primitive::<Int64Type>();
        assert_eq!(1, aa.value(0));
        assert_eq!(-10, aa.value(1));
        assert_eq!(1627668684594000000, aa.value(2));
        let bb = batch.column(b.0).as_list::<i32>();
        let bb = bb.values().as_primitive::<Float64Type>();
        assert_eq!(9, bb.len());
        assert_eq!(2.0, bb.value(0));
        assert_eq!(-6.1, bb.value(5));
        assert!(!bb.is_valid(7));

        let cc = batch
            .column(c.0)
            .as_any()
            .downcast_ref::<ListArray>()
            .unwrap();
        let cc = cc.values().as_boolean();
        assert_eq!(6, cc.len());
        assert!(!cc.value(0));
        assert!(!cc.value(4));
        assert!(!cc.is_valid(5));
    }

    #[test]
    fn test_empty_json_arrays() {
        let json_content = r#"
            {"items": []}
            {"items": null}
            {}
            "#;

        let schema = Arc::new(Schema::new(vec![Field::new(
            "items",
            DataType::List(FieldRef::new(Field::new_list_field(DataType::Null, true))),
            true,
        )]));

        let batches = do_read(json_content, 1024, false, false, schema);
        assert_eq!(batches.len(), 1);

        let col1 = batches[0].column(0).as_list::<i32>();
        assert_eq!(col1.null_count(), 2);
        assert!(col1.value(0).is_empty());
        assert_eq!(col1.value(0).data_type(), &DataType::Null);
        assert!(col1.is_null(1));
        assert!(col1.is_null(2));
    }

    #[test]
    fn test_nested_empty_json_arrays() {
        let json_content = r#"
            {"items": [[],[]]}
            {"items": [[null, null],[null]]}
            "#;

        let schema = Arc::new(Schema::new(vec![Field::new(
            "items",
            DataType::List(FieldRef::new(Field::new_list_field(
                DataType::List(FieldRef::new(Field::new_list_field(DataType::Null, true))),
                true,
            ))),
            true,
        )]));

        let batches = do_read(json_content, 1024, false, false, schema);
        assert_eq!(batches.len(), 1);

        let col1 = batches[0].column(0).as_list::<i32>();
        assert_eq!(col1.null_count(), 0);
        assert_eq!(col1.value(0).len(), 2);
        assert!(col1.value(0).as_list::<i32>().value(0).is_empty());
        assert!(col1.value(0).as_list::<i32>().value(1).is_empty());

        assert_eq!(col1.value(1).len(), 2);
        assert_eq!(col1.value(1).as_list::<i32>().value(0).len(), 2);
        assert_eq!(col1.value(1).as_list::<i32>().value(1).len(), 1);
    }

    #[test]
    fn test_nested_list_json_arrays() {
        let c_field = Field::new_struct("c", vec![Field::new("d", DataType::Utf8, true)], true);
        let a_struct_field = Field::new_struct(
            "a",
            vec![Field::new("b", DataType::Boolean, true), c_field.clone()],
            true,
        );
        let a_field = Field::new("a", DataType::List(Arc::new(a_struct_field.clone())), true);
        let schema = Arc::new(Schema::new(vec![a_field.clone()]));
        let builder = ReaderBuilder::new(schema).with_batch_size(64);
        let json_content = r#"
        {"a": [{"b": true, "c": {"d": "a_text"}}, {"b": false, "c": {"d": "b_text"}}]}
        {"a": [{"b": false, "c": null}]}
        {"a": [{"b": true, "c": {"d": "c_text"}}, {"b": null, "c": {"d": "d_text"}}, {"b": true, "c": {"d": null}}]}
        {"a": null}
        {"a": []}
        {"a": [null]}
        "#;
        let mut reader = builder.build(Cursor::new(json_content)).unwrap();

        // build expected output
        let d = StringArray::from(vec![
            Some("a_text"),
            Some("b_text"),
            None,
            Some("c_text"),
            Some("d_text"),
            None,
            None,
        ]);
        let c = StructArray::new(
            vec![Field::new("d", DataType::Utf8, true)].into(),
            vec![Arc::new(d.clone()) as ArrayRef],
            Some(NullBuffer::from(vec![
                true, true, false, true, true, true, false,
            ])),
        );
        let b = BooleanArray::from(vec![
            Some(true),
            Some(false),
            Some(false),
            Some(true),
            None,
            Some(true),
            None,
        ]);
        let a = StructArray::new(
            vec![Field::new("b", DataType::Boolean, true), c_field.clone()].into(),
            vec![
                Arc::new(b.clone()) as ArrayRef,
                Arc::new(c.clone()) as ArrayRef,
            ],
            Some(NullBuffer::from(vec![
                true, true, true, true, true, true, false,
            ])),
        );
        let a_list = ListArray::new(
            Arc::new(a_struct_field.clone()),
            OffsetBuffer::new(ScalarBuffer::from(vec![0i32, 2, 3, 6, 6, 6, 7])),
            Arc::new(a),
            Some(NullBuffer::from(vec![true, true, true, false, true, true])),
        );

        // compare `a` with result from json reader
        let batch = reader.next().unwrap().unwrap();
        let read = batch.column(0);
        assert_eq!(read.len(), 6);
        // compare the arrays the long way around, to better detect differences
        let read: &ListArray = read.as_list::<i32>();
        let expected = &a_list;
        assert_eq!(read.value_offsets(), &[0, 2, 3, 6, 6, 6, 7]);
        // compare list null buffers
        assert_eq!(read.nulls(), expected.nulls());
        // build struct from list
        let struct_array = read.values().as_struct();
        let expected_struct_array = expected.values().as_struct();

        assert_eq!(7, struct_array.len());
        assert_eq!(1, struct_array.null_count());
        assert_eq!(7, expected_struct_array.len());
        assert_eq!(1, expected_struct_array.null_count());
        // test struct's nulls
        assert_eq!(struct_array.nulls(), expected_struct_array.nulls());
        // test struct's fields
        let read_b = struct_array.column(0);
        assert_eq!(read_b.as_ref(), &b);
        let read_c = struct_array.column(1);
        assert_eq!(read_c.as_struct(), &c);
        let read_c = read_c.as_struct();
        let read_d = read_c.column(0);
        assert_eq!(read_d.as_ref(), &d);

        assert_eq!(read, expected);
    }

    fn assert_read_list_view<O: OffsetSizeTrait>() {
        let field = Arc::new(Field::new("item", DataType::Int32, true));
        let data_type = GenericListViewArray::<O>::DATA_TYPE_CONSTRUCTOR(field.clone());
        let schema = Arc::new(Schema::new(vec![Field::new("lv", data_type, true)]));

        let buf = r#"
        {"lv": [1, 2, 3]}
        {"lv": [4, null]}
        {"lv": null}
        {"lv": [6]}
        {"lv": []}
        "#;

        let batches = do_read(buf, 1024, false, false, schema);
        assert_eq!(batches.len(), 1);
        let batch = &batches[0];
        let col = batch.column(0);
        let list_view = col
            .as_any()
            .downcast_ref::<GenericListViewArray<O>>()
            .unwrap();

        assert_eq!(list_view.len(), 5);

        // Check offsets and sizes
        let expected_offsets: Vec<O> = vec![0, 3, 5, 5, 6]
            .into_iter()
            .map(|v| O::usize_as(v))
            .collect();
        let expected_sizes: Vec<O> = vec![3, 2, 0, 1, 0]
            .into_iter()
            .map(|v| O::usize_as(v))
            .collect();
        assert_eq!(list_view.value_offsets(), &expected_offsets);
        assert_eq!(list_view.value_sizes(), &expected_sizes);

        // Row 0: [1, 2, 3]
        assert!(list_view.is_valid(0));
        let vals = list_view.value(0);
        let ints = vals.as_primitive::<Int32Type>();
        assert_eq!(ints.values(), &[1, 2, 3]);

        // Row 1: [4, null]
        assert!(list_view.is_valid(1));
        let vals = list_view.value(1);
        let ints = vals.as_primitive::<Int32Type>();
        assert_eq!(ints.len(), 2);
        assert_eq!(ints.value(0), 4);
        assert!(ints.is_null(1));

        // Row 2: null
        assert!(list_view.is_null(2));

        // Row 3: [6]
        assert!(list_view.is_valid(3));
        let vals = list_view.value(3);
        let ints = vals.as_primitive::<Int32Type>();
        assert_eq!(ints.values(), &[6]);

        // Row 4: []
        assert!(list_view.is_valid(4));
        let vals = list_view.value(4);
        assert_eq!(vals.len(), 0);
    }

    #[test]
    fn test_read_list_view() {
        assert_read_list_view::<i32>();
        assert_read_list_view::<i64>();
    }

    #[test]
    fn test_read_list_view_rejects_null_non_nullable_child() {
        let field = Arc::new(Field::new("item", DataType::Int32, false));
        for (data_type, array_type) in [
            (DataType::ListView(field.clone()), "ListViewArray"),
            (DataType::LargeListView(field.clone()), "LargeListViewArray"),
        ] {
            let schema = Arc::new(Schema::new(vec![Field::new("lv", data_type, true)]));
            let buf = r#"
            {"lv": [1, 2, 3]}
            {"lv": [4, null]}
            "#;

            let error = ReaderBuilder::new(schema)
                .build(Cursor::new(buf.as_bytes()))
                .unwrap()
                .collect::<Result<Vec<_>, _>>()
                .unwrap_err();

            assert_eq!(
                error.to_string(),
                format!(
                    "Invalid argument error: Non-nullable field of {array_type} \"item\" cannot contain nulls"
                )
            );
        }
    }

    #[test]
    fn test_fixed_size_list() {
        let buf = r#"
        {"a": [1, 2, 3]}
        {"a": [4, 5, 6]}
        {"a": [7, 8, 9]}
        "#;

        let field = Field::new_list_field(DataType::Int32, true);
        let schema = Arc::new(Schema::new(vec![Field::new(
            "a",
            DataType::FixedSizeList(Arc::new(field), 3),
            false,
        )]));

        let batches = do_read(buf, 1024, false, false, schema);
        assert_eq!(batches.len(), 1);

        let col = batches[0].column(0).as_fixed_size_list();
        assert_eq!(col.len(), 3);
        assert_eq!(col.value_length(), 3);

        let values = col.values().as_primitive::<Int32Type>();
        assert_eq!(values.values(), &[1, 2, 3, 4, 5, 6, 7, 8, 9]);
    }

    #[test]
    fn test_fixed_size_list_nullable() {
        let buf = r#"
        {"a": [1, 2]}
        {"a": null}
        {"a": [3, null]}
        "#;

        let field = Field::new_list_field(DataType::Int32, true);
        let schema = Arc::new(Schema::new(vec![Field::new(
            "a",
            DataType::FixedSizeList(Arc::new(field), 2),
            true,
        )]));

        let batches = do_read(buf, 1024, false, false, schema);
        assert_eq!(batches.len(), 1);

        let col = batches[0].column(0).as_fixed_size_list();
        assert_eq!(col.len(), 3);
        assert!(col.is_valid(0));
        assert!(col.is_null(1));
        assert!(col.is_valid(2));

        let values = col.values().as_primitive::<Int32Type>();
        assert_eq!(values.value(0), 1);
        assert_eq!(values.value(1), 2);
        assert_eq!(values.value(4), 3);
        assert!(values.is_null(5));
    }

    #[test]
    fn test_fixed_size_list_zero_size_non_nullable() {
        let buf = r#"
        {"a": []}
        {"a": []}
        {"a": []}
        "#;

        let field = Field::new_list_field(DataType::Int32, true);
        let schema = Arc::new(Schema::new(vec![Field::new(
            "a",
            DataType::FixedSizeList(Arc::new(field), 0),
            false,
        )]));

        let batches = do_read(buf, 1024, false, false, schema);
        assert_eq!(batches.len(), 1);

        let col = batches[0].column(0).as_fixed_size_list();
        assert_eq!(col.len(), 3);
        assert_eq!(col.value_length(), 0);

        let values = col.values().as_primitive::<Int32Type>();
        assert!(values.values().is_empty());
    }

    #[test]
    fn test_fixed_size_list_wrong_size() {
        let buf = r#"{"a": [1, 2, 3]}"#;

        let field = Field::new_list_field(DataType::Int32, true);
        let schema = Arc::new(Schema::new(vec![Field::new(
            "a",
            DataType::FixedSizeList(Arc::new(field), 2),
            false,
        )]));

        let err = ReaderBuilder::new(schema)
            .build(Cursor::new(buf.as_bytes()))
            .unwrap()
            .next()
            .unwrap()
            .unwrap_err();

        assert!(err.to_string().contains("expected 2 but got 3"), "{}", err);
    }

    #[test]
    fn test_fixed_size_list_nested() {
        let buf = r#"
        {"a": [[1, 2], [3, 4]]}
        {"a": [[5, 6], [7, 8]]}
        "#;

        let inner_field = Field::new_list_field(DataType::Int32, true);
        let inner_type = DataType::FixedSizeList(Arc::new(inner_field), 2);
        let outer_field = Arc::new(Field::new_list_field(inner_type.clone(), true));
        let schema = Arc::new(Schema::new(vec![Field::new(
            "a",
            DataType::FixedSizeList(outer_field, 2),
            false,
        )]));

        let batches = do_read(buf, 1024, false, false, schema);
        assert_eq!(batches.len(), 1);

        let col = batches[0].column(0).as_fixed_size_list();
        assert_eq!(col.len(), 2);
        assert_eq!(col.value_length(), 2);

        let inner = col.values().as_fixed_size_list();
        assert_eq!(inner.len(), 4);
        assert_eq!(inner.value_length(), 2);

        let values = inner.values().as_primitive::<Int32Type>();
        assert_eq!(values.values(), &[1, 2, 3, 4, 5, 6, 7, 8]);
    }

    #[test]
    fn test_fixed_size_list_ignore_type_conflicts() {
        let field = Field::new("item", DataType::Int32, true);
        let schema = Arc::new(Schema::new(vec![Field::new(
            "a",
            DataType::FixedSizeList(Arc::new(field), 2),
            true,
        )]));

        let json = vec![
            json!({"a": [1, 2]}),
            json!({"a": "not a list"}),
            json!({"a": 42}),
            json!({"a": [6, 7]}),
        ];

        let mut decoder = ReaderBuilder::new(schema)
            .with_ignore_type_conflicts(true)
            .build_decoder()
            .unwrap();
        decoder.serialize(&json).unwrap();
        let batch = decoder.flush().unwrap().unwrap();

        let col = batch.column(0).as_fixed_size_list();
        assert_eq!(col.len(), 4);
        assert!(col.is_valid(0));
        assert!(col.is_null(1)); // string -> null
        assert!(col.is_null(2)); // number -> null
        assert!(col.is_valid(3));

        let values = col.values().as_primitive::<Int32Type>();
        assert_eq!(values.value(0), 1);
        assert_eq!(values.value(1), 2);
        assert_eq!(values.value(6), 6);
        assert_eq!(values.value(7), 7);
    }

    #[test]
    fn test_skip_empty_lines() {
        let schema = Schema::new(vec![Field::new("a", DataType::Int64, true)]);
        let builder = ReaderBuilder::new(Arc::new(schema)).with_batch_size(64);
        let json_content = "
        {\"a\": 1}
        {\"a\": 2}
        {\"a\": 3}";
        let mut reader = builder.build(Cursor::new(json_content)).unwrap();
        let batch = reader.next().unwrap().unwrap();

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

        let schema = reader.schema();
        let c = schema.column_with_name("a").unwrap();
        assert_eq!(&DataType::Int64, c.1.data_type());
    }

    #[test]
    fn test_with_multiple_batches() {
        let file = File::open("test/data/basic_nulls.json").unwrap();
        let mut reader = BufReader::new(file);
        let (schema, _) = infer_json_schema(&mut reader, None).unwrap();
        reader.rewind().unwrap();

        let builder = ReaderBuilder::new(Arc::new(schema)).with_batch_size(5);
        let mut reader = builder.build(reader).unwrap();

        let mut num_records = Vec::new();
        while let Some(rb) = reader.next().transpose().unwrap() {
            num_records.push(rb.num_rows());
        }

        assert_eq!(vec![5, 5, 2], num_records);
    }

    #[test]
    fn test_timestamp_from_json_seconds() {
        let schema = Schema::new(vec![Field::new(
            "a",
            DataType::Timestamp(TimeUnit::Second, None),
            true,
        )]);

        let mut reader = read_file("test/data/basic_nulls.json", Some(schema));
        let batch = reader.next().unwrap().unwrap();

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

        let schema = reader.schema();
        let batch_schema = batch.schema();
        assert_eq!(schema, batch_schema);

        let a = schema.column_with_name("a").unwrap();
        assert_eq!(
            &DataType::Timestamp(TimeUnit::Second, None),
            a.1.data_type()
        );

        let aa = batch.column(a.0).as_primitive::<TimestampSecondType>();
        assert!(aa.is_valid(0));
        assert!(!aa.is_valid(1));
        assert!(!aa.is_valid(2));
        assert_eq!(1, aa.value(0));
        assert_eq!(1, aa.value(3));
        assert_eq!(5, aa.value(7));
    }

    #[test]
    fn test_timestamp_from_json_milliseconds() {
        let schema = Schema::new(vec![Field::new(
            "a",
            DataType::Timestamp(TimeUnit::Millisecond, None),
            true,
        )]);

        let mut reader = read_file("test/data/basic_nulls.json", Some(schema));
        let batch = reader.next().unwrap().unwrap();

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

        let schema = reader.schema();
        let batch_schema = batch.schema();
        assert_eq!(schema, batch_schema);

        let a = schema.column_with_name("a").unwrap();
        assert_eq!(
            &DataType::Timestamp(TimeUnit::Millisecond, None),
            a.1.data_type()
        );

        let aa = batch.column(a.0).as_primitive::<TimestampMillisecondType>();
        assert!(aa.is_valid(0));
        assert!(!aa.is_valid(1));
        assert!(!aa.is_valid(2));
        assert_eq!(1, aa.value(0));
        assert_eq!(1, aa.value(3));
        assert_eq!(5, aa.value(7));
    }

    #[test]
    fn test_date_from_json_milliseconds() {
        let schema = Schema::new(vec![Field::new("a", DataType::Date64, true)]);

        let mut reader = read_file("test/data/basic_nulls.json", Some(schema));
        let batch = reader.next().unwrap().unwrap();

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

        let schema = reader.schema();
        let batch_schema = batch.schema();
        assert_eq!(schema, batch_schema);

        let a = schema.column_with_name("a").unwrap();
        assert_eq!(&DataType::Date64, a.1.data_type());

        let aa = batch.column(a.0).as_primitive::<Date64Type>();
        assert!(aa.is_valid(0));
        assert!(!aa.is_valid(1));
        assert!(!aa.is_valid(2));
        assert_eq!(1, aa.value(0));
        assert_eq!(1, aa.value(3));
        assert_eq!(5, aa.value(7));
    }

    #[test]
    fn test_time_from_json_nanoseconds() {
        let schema = Schema::new(vec![Field::new(
            "a",
            DataType::Time64(TimeUnit::Nanosecond),
            true,
        )]);

        let mut reader = read_file("test/data/basic_nulls.json", Some(schema));
        let batch = reader.next().unwrap().unwrap();

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

        let schema = reader.schema();
        let batch_schema = batch.schema();
        assert_eq!(schema, batch_schema);

        let a = schema.column_with_name("a").unwrap();
        assert_eq!(&DataType::Time64(TimeUnit::Nanosecond), a.1.data_type());

        let aa = batch.column(a.0).as_primitive::<Time64NanosecondType>();
        assert!(aa.is_valid(0));
        assert!(!aa.is_valid(1));
        assert!(!aa.is_valid(2));
        assert_eq!(1, aa.value(0));
        assert_eq!(1, aa.value(3));
        assert_eq!(5, aa.value(7));
    }

    #[test]
    fn test_json_iterator() {
        let file = File::open("test/data/basic.json").unwrap();
        let mut reader = BufReader::new(file);
        let (schema, _) = infer_json_schema(&mut reader, None).unwrap();
        reader.rewind().unwrap();

        let builder = ReaderBuilder::new(Arc::new(schema)).with_batch_size(5);
        let reader = builder.build(reader).unwrap();
        let schema = reader.schema();
        let (col_a_index, _) = schema.column_with_name("a").unwrap();

        let mut sum_num_rows = 0;
        let mut num_batches = 0;
        let mut sum_a = 0;
        for batch in reader {
            let batch = batch.unwrap();
            assert_eq!(8, batch.num_columns());
            sum_num_rows += batch.num_rows();
            num_batches += 1;
            let batch_schema = batch.schema();
            assert_eq!(schema, batch_schema);
            let a_array = batch.column(col_a_index).as_primitive::<Int64Type>();
            sum_a += (0..a_array.len()).map(|i| a_array.value(i)).sum::<i64>();
        }
        assert_eq!(12, sum_num_rows);
        assert_eq!(3, num_batches);
        assert_eq!(100000000000011, sum_a);
    }

    #[test]
    fn test_decoder_error() {
        let schema = Arc::new(Schema::new(vec![Field::new_struct(
            "a",
            vec![Field::new("child", DataType::Int32, false)],
            true,
        )]));

        let mut decoder = ReaderBuilder::new(schema.clone()).build_decoder().unwrap();
        let _ = decoder.decode(br#"{"a": { "child":"#).unwrap();
        assert!(decoder.tape_decoder.has_partial_row());
        assert_eq!(decoder.tape_decoder.num_buffered_rows(), 1);
        let _ = decoder.flush().unwrap_err();
        assert!(decoder.tape_decoder.has_partial_row());
        assert_eq!(decoder.tape_decoder.num_buffered_rows(), 1);

        let parse_err = |s: &str| {
            ReaderBuilder::new(schema.clone())
                .build(Cursor::new(s.as_bytes()))
                .unwrap()
                .next()
                .unwrap()
                .unwrap_err()
                .to_string()
        };

        let err = parse_err(r#"{"a": 123}"#);
        assert_eq!(
            err,
            "Json error: whilst decoding field 'a': expected { got 123"
        );

        let err = parse_err(r#"{"a": ["bar"]}"#);
        assert_eq!(
            err,
            r#"Json error: whilst decoding field 'a': expected { got ["bar"]"#
        );

        let err = parse_err(r#"{"a": []}"#);
        assert_eq!(
            err,
            "Json error: whilst decoding field 'a': expected { got []"
        );

        let err = parse_err(r#"{"a": [{"child": 234}]}"#);
        assert_eq!(
            err,
            r#"Json error: whilst decoding field 'a': expected { got [{"child": 234}]"#
        );

        let err = parse_err(r#"{"a": [{"child": {"foo": [{"foo": ["bar"]}]}}]}"#);
        assert_eq!(
            err,
            r#"Json error: whilst decoding field 'a': expected { got [{"child": {"foo": [{"foo": ["bar"]}]}}]"#
        );

        let err = parse_err(r#"{"a": true}"#);
        assert_eq!(
            err,
            "Json error: whilst decoding field 'a': expected { got true"
        );

        let err = parse_err(r#"{"a": false}"#);
        assert_eq!(
            err,
            "Json error: whilst decoding field 'a': expected { got false"
        );

        let err = parse_err(r#"{"a": "foo"}"#);
        assert_eq!(
            err,
            "Json error: whilst decoding field 'a': expected { got \"foo\""
        );

        let err = parse_err(r#"{"a": {"child": false}}"#);
        assert_eq!(
            err,
            "Json error: whilst decoding field 'a': whilst decoding field 'child': expected primitive got false"
        );

        let err = parse_err(r#"{"a": {"child": []}}"#);
        assert_eq!(
            err,
            "Json error: whilst decoding field 'a': whilst decoding field 'child': expected primitive got []"
        );

        let err = parse_err(r#"{"a": {"child": [123]}}"#);
        assert_eq!(
            err,
            "Json error: whilst decoding field 'a': whilst decoding field 'child': expected primitive got [123]"
        );

        let err = parse_err(r#"{"a": {"child": [123, 3465346]}}"#);
        assert_eq!(
            err,
            "Json error: whilst decoding field 'a': whilst decoding field 'child': expected primitive got [123, 3465346]"
        );
    }

    #[test]
    fn test_serialize_timestamp() {
        let json = vec![
            json!({"timestamp": 1681319393}),
            json!({"timestamp": "1970-01-01T00:00:00+02:00"}),
        ];
        let schema = Schema::new(vec![Field::new(
            "timestamp",
            DataType::Timestamp(TimeUnit::Second, None),
            true,
        )]);
        let mut decoder = ReaderBuilder::new(Arc::new(schema))
            .build_decoder()
            .unwrap();
        decoder.serialize(&json).unwrap();
        let batch = decoder.flush().unwrap().unwrap();
        assert_eq!(batch.num_rows(), 2);
        assert_eq!(batch.num_columns(), 1);
        let values = batch.column(0).as_primitive::<TimestampSecondType>();
        assert_eq!(values.values(), &[1681319393, -7200]);
    }

    #[test]
    fn test_serialize_decimal() {
        let json = vec![
            json!({"decimal": 1.234}),
            json!({"decimal": "1.234"}),
            json!({"decimal": 1234}),
            json!({"decimal": "1234"}),
        ];
        let schema = Schema::new(vec![Field::new(
            "decimal",
            DataType::Decimal128(10, 3),
            true,
        )]);
        let mut decoder = ReaderBuilder::new(Arc::new(schema))
            .build_decoder()
            .unwrap();
        decoder.serialize(&json).unwrap();
        let batch = decoder.flush().unwrap().unwrap();
        assert_eq!(batch.num_rows(), 4);
        assert_eq!(batch.num_columns(), 1);
        let values = batch.column(0).as_primitive::<Decimal128Type>();
        assert_eq!(values.values(), &[1234, 1234, 1234000, 1234000]);
    }

    #[test]
    fn test_serde_field() {
        let field = Field::new("int", DataType::Int32, true);
        let mut decoder = ReaderBuilder::new_with_field(field)
            .build_decoder()
            .unwrap();
        decoder.serialize(&[1_i32, 2, 3, 4]).unwrap();
        let b = decoder.flush().unwrap().unwrap();
        let values = b.column(0).as_primitive::<Int32Type>().values();
        assert_eq!(values, &[1, 2, 3, 4]);
    }

    #[test]
    fn test_serde_large_numbers() {
        let field = Field::new("int", DataType::Int64, true);
        let mut decoder = ReaderBuilder::new_with_field(field)
            .build_decoder()
            .unwrap();

        decoder.serialize(&[1699148028689_u64, 2, 3, 4]).unwrap();
        let b = decoder.flush().unwrap().unwrap();
        let values = b.column(0).as_primitive::<Int64Type>().values();
        assert_eq!(values, &[1699148028689, 2, 3, 4]);

        let field = Field::new(
            "int",
            DataType::Timestamp(TimeUnit::Microsecond, None),
            true,
        );
        let mut decoder = ReaderBuilder::new_with_field(field)
            .build_decoder()
            .unwrap();

        decoder.serialize(&[1699148028689_u64, 2, 3, 4]).unwrap();
        let b = decoder.flush().unwrap().unwrap();
        let values = b
            .column(0)
            .as_primitive::<TimestampMicrosecondType>()
            .values();
        assert_eq!(values, &[1699148028689, 2, 3, 4]);
    }

    #[test]
    fn test_coercing_primitive_into_string_decoder() {
        let buf = &format!(
            r#"[{{"a": 1, "b": "A", "c": "T"}}, {{"a": 2, "b": "BB", "c": "F"}}, {{"a": {}, "b": 123, "c": false}}, {{"a": {}, "b": 789, "c": true}}]"#,
            (i32::MAX as i64 + 10),
            i64::MAX - 10
        );
        let schema = Schema::new(vec![
            Field::new("a", DataType::Float64, true),
            Field::new("b", DataType::Utf8, true),
            Field::new("c", DataType::Utf8, true),
        ]);
        let json_array: Vec<serde_json::Value> = serde_json::from_str(buf).unwrap();
        let schema_ref = Arc::new(schema);

        // read record batches
        let reader = ReaderBuilder::new(schema_ref.clone()).with_coerce_primitive(true);
        let mut decoder = reader.build_decoder().unwrap();
        decoder.serialize(json_array.as_slice()).unwrap();
        let batch = decoder.flush().unwrap().unwrap();
        assert_eq!(
            batch,
            RecordBatch::try_new(
                schema_ref,
                vec![
                    Arc::new(Float64Array::from(vec![
                        1.0,
                        2.0,
                        (i32::MAX as i64 + 10) as f64,
                        (i64::MAX - 10) as f64
                    ])),
                    Arc::new(StringArray::from(vec!["A", "BB", "123", "789"])),
                    Arc::new(StringArray::from(vec!["T", "F", "false", "true"])),
                ]
            )
            .unwrap()
        );
    }

    #[test]
    fn test_serialize_f32_into_string() {
        // Coercing an f32 into a string column must render the value, not its raw bit pattern.
        let field = Field::new("f", DataType::Utf8, true);
        let mut decoder = ReaderBuilder::new_with_field(field)
            .with_coerce_primitive(true)
            .build_decoder()
            .unwrap();
        decoder.serialize(&[1.5_f32, -2.25_f32]).unwrap();
        let batch = decoder.flush().unwrap().unwrap();
        let values = batch.column(0).as_string::<i32>();
        assert_eq!(values.value(0), "1.5");
        assert_eq!(values.value(1), "-2.25");
    }

    // Parse the given `row` in `struct_mode` as a type given by fields.
    //
    // If as_struct == true, wrap the fields in a Struct field with name "r".
    // If as_struct == false, wrap the fields in a Schema.
    fn _parse_structs(
        row: &str,
        struct_mode: StructMode,
        fields: Fields,
        as_struct: bool,
    ) -> Result<RecordBatch, ArrowError> {
        let builder = if as_struct {
            ReaderBuilder::new_with_field(Field::new("r", DataType::Struct(fields), true))
        } else {
            ReaderBuilder::new(Arc::new(Schema::new(fields)))
        };
        builder
            .with_struct_mode(struct_mode)
            .build(Cursor::new(row.as_bytes()))
            .unwrap()
            .next()
            .unwrap()
    }

    #[test]
    fn test_struct_decoding_list_length() {
        use arrow_array::array;

        let row = "[1, 2]";

        let mut fields = vec![Field::new("a", DataType::Int32, true)];
        let too_few_fields = Fields::from(fields.clone());
        fields.push(Field::new("b", DataType::Int32, true));
        let correct_fields = Fields::from(fields.clone());
        fields.push(Field::new("c", DataType::Int32, true));
        let too_many_fields = Fields::from(fields.clone());

        let parse = |fields: Fields, as_struct: bool| {
            _parse_structs(row, StructMode::ListOnly, fields, as_struct)
        };

        let expected_row = StructArray::new(
            correct_fields.clone(),
            vec![
                Arc::new(array::Int32Array::from(vec![1])),
                Arc::new(array::Int32Array::from(vec![2])),
            ],
            None,
        );
        let row_field = Field::new("r", DataType::Struct(correct_fields.clone()), true);

        assert_eq!(
            parse(too_few_fields.clone(), true).unwrap_err().to_string(),
            "Json error: found extra columns for 1 fields".to_string()
        );
        assert_eq!(
            parse(too_few_fields, false).unwrap_err().to_string(),
            "Json error: found extra columns for 1 fields".to_string()
        );
        assert_eq!(
            parse(correct_fields.clone(), true).unwrap(),
            RecordBatch::try_new(
                Arc::new(Schema::new(vec![row_field])),
                vec![Arc::new(expected_row.clone())]
            )
            .unwrap()
        );
        assert_eq!(
            parse(correct_fields, false).unwrap(),
            RecordBatch::from(expected_row)
        );
        assert_eq!(
            parse(too_many_fields.clone(), true)
                .unwrap_err()
                .to_string(),
            "Json error: found 2 columns for 3 fields".to_string()
        );
        assert_eq!(
            parse(too_many_fields, false).unwrap_err().to_string(),
            "Json error: found 2 columns for 3 fields".to_string()
        );
    }

    #[test]
    fn test_struct_decoding() {
        use arrow_array::builder;

        let nested_object_json = r#"{"a": {"b": [1, 2], "c": {"d": 3}}}"#;
        let nested_list_json = r#"[[[1, 2], {"d": 3}]]"#;
        let nested_mixed_json = r#"{"a": [[1, 2], {"d": 3}]}"#;

        let struct_fields = Fields::from(vec![
            Field::new("b", DataType::new_list(DataType::Int32, true), true),
            Field::new_map(
                "c",
                Field::MAP_ENTRIES_FIELD_DEFAULT_NAME,
                Field::new(Field::MAP_KEY_FIELD_DEFAULT_NAME, DataType::Utf8, false),
                Field::new(Field::MAP_VALUE_FIELD_DEFAULT_NAME, DataType::Int32, true),
                false,
                false,
            ),
        ]);

        let list_array =
            ListArray::from_iter_primitive::<Int32Type, _, _>(vec![Some(vec![Some(1), Some(2)])]);

        let map_array = {
            let mut map_builder = builder::MapBuilder::new(
                None,
                builder::StringBuilder::new(),
                builder::Int32Builder::new(),
            );
            map_builder.keys().append_value("d");
            map_builder.values().append_value(3);
            map_builder.append(true).unwrap();
            map_builder.finish()
        };

        let struct_array = StructArray::new(
            struct_fields.clone(),
            vec![Arc::new(list_array), Arc::new(map_array)],
            None,
        );

        let fields = Fields::from(vec![Field::new("a", DataType::Struct(struct_fields), true)]);
        let schema = Arc::new(Schema::new(fields.clone()));
        let expected = RecordBatch::try_new(schema.clone(), vec![Arc::new(struct_array)]).unwrap();

        let parse = |row: &str, struct_mode: StructMode| {
            _parse_structs(row, struct_mode, fields.clone(), false)
        };

        assert_eq!(
            parse(nested_object_json, StructMode::ObjectOnly).unwrap(),
            expected
        );
        assert_eq!(
            parse(nested_list_json, StructMode::ObjectOnly)
                .unwrap_err()
                .to_string(),
            "Json error: expected { got [[[1, 2], {\"d\": 3}]]".to_owned()
        );
        assert_eq!(
            parse(nested_mixed_json, StructMode::ObjectOnly)
                .unwrap_err()
                .to_string(),
            "Json error: whilst decoding field 'a': expected { got [[1, 2], {\"d\": 3}]".to_owned()
        );

        assert_eq!(
            parse(nested_list_json, StructMode::ListOnly).unwrap(),
            expected
        );
        assert_eq!(
            parse(nested_object_json, StructMode::ListOnly)
                .unwrap_err()
                .to_string(),
            "Json error: expected [ got {\"a\": {\"b\": [1, 2]\"c\": {\"d\": 3}}}".to_owned()
        );
        assert_eq!(
            parse(nested_mixed_json, StructMode::ListOnly)
                .unwrap_err()
                .to_string(),
            "Json error: expected [ got {\"a\": [[1, 2], {\"d\": 3}]}".to_owned()
        );
    }

    // Test cases:
    // [] -> RecordBatch row with no entries.  Schema = [('a', Int32)] -> Error
    // [] -> RecordBatch row with no entries. Schema = [('r', [('a', Int32)])] -> Error
    // [] -> StructArray row with no entries. Fields [('a', Int32')] -> Error
    // [[]] -> RecordBatch row with empty struct entry. Schema = [('r', [('a', Int32)])] -> Error
    #[test]
    fn test_struct_decoding_empty_list() {
        let int_field = Field::new("a", DataType::Int32, true);
        let struct_field = Field::new(
            "r",
            DataType::Struct(Fields::from(vec![int_field.clone()])),
            true,
        );

        let parse = |row: &str, as_struct: bool, field: Field| {
            _parse_structs(
                row,
                StructMode::ListOnly,
                Fields::from(vec![field]),
                as_struct,
            )
        };

        // Missing fields
        assert_eq!(
            parse("[]", true, struct_field.clone())
                .unwrap_err()
                .to_string(),
            "Json error: found 0 columns for 1 fields".to_owned()
        );
        assert_eq!(
            parse("[]", false, int_field.clone())
                .unwrap_err()
                .to_string(),
            "Json error: found 0 columns for 1 fields".to_owned()
        );
        assert_eq!(
            parse("[]", false, struct_field.clone())
                .unwrap_err()
                .to_string(),
            "Json error: found 0 columns for 1 fields".to_owned()
        );
        assert_eq!(
            parse("[[]]", false, struct_field.clone())
                .unwrap_err()
                .to_string(),
            "Json error: whilst decoding field 'r': found 0 columns for 1 fields".to_owned()
        );
    }

    #[test]
    fn test_decode_list_struct_with_wrong_types() {
        let int_field = Field::new("a", DataType::Int32, true);
        let struct_field = Field::new(
            "r",
            DataType::Struct(Fields::from(vec![int_field.clone()])),
            true,
        );

        let parse = |row: &str, as_struct: bool, field: Field| {
            _parse_structs(
                row,
                StructMode::ListOnly,
                Fields::from(vec![field]),
                as_struct,
            )
        };

        // Wrong values
        assert_eq!(
            parse(r#"[["a"]]"#, false, struct_field.clone())
                .unwrap_err()
                .to_string(),
            "Json error: whilst decoding field 'r': whilst decoding field 'a': failed to parse \"a\" as Int32".to_owned()
        );
        assert_eq!(
            parse(r#"[["a"]]"#, true, struct_field.clone())
                .unwrap_err()
                .to_string(),
            "Json error: whilst decoding field 'r': whilst decoding field 'a': failed to parse \"a\" as Int32".to_owned()
        );
        assert_eq!(
            parse(r#"["a"]"#, true, int_field.clone())
                .unwrap_err()
                .to_string(),
            "Json error: whilst decoding field 'a': failed to parse \"a\" as Int32".to_owned()
        );
        assert_eq!(
            parse(r#"["a"]"#, false, int_field.clone())
                .unwrap_err()
                .to_string(),
            "Json error: whilst decoding field 'a': failed to parse \"a\" as Int32".to_owned()
        );
    }

    #[test]
    fn test_type_conflict_nulls() {
        let schema = Schema::new(vec![
            Field::new("null", DataType::Null, true),
            Field::new("bool", DataType::Boolean, true),
            Field::new("primitive", DataType::Int32, true),
            Field::new("numeric", DataType::Decimal128(10, 3), true),
            Field::new("string", DataType::Utf8, true),
            Field::new("string_view", DataType::Utf8View, true),
            Field::new(
                "timestamp",
                DataType::Timestamp(TimeUnit::Second, None),
                true,
            ),
            Field::new(
                "array",
                DataType::List(Arc::new(Field::new("item", DataType::Int32, true))),
                true,
            ),
            Field::new(
                "map",
                DataType::Map(
                    Arc::new(Field::new(
                        Field::MAP_ENTRIES_FIELD_DEFAULT_NAME,
                        DataType::Struct(Fields::from(vec![
                            Field::new(Field::MAP_KEY_FIELD_DEFAULT_NAME, DataType::Utf8, false),
                            Field::new(Field::MAP_VALUE_FIELD_DEFAULT_NAME, DataType::Utf8, true),
                        ])),
                        false, // not nullable
                    )),
                    false, // not sorted
                ),
                true, // nullable
            ),
            Field::new(
                "struct",
                DataType::Struct(Fields::from(vec![Field::new("a", DataType::Int32, true)])),
                true,
            ),
        ]);

        // A compatible value for each schema field above, in schema order
        let json_values = vec![
            json!(null),
            json!(true),
            json!(42),
            json!(1.234),
            json!("hi"),
            json!("ho"),
            json!("1970-01-01T00:00:00+02:00"),
            json!([1, "ho", 3]),
            json!({"k": "value"}),
            json!({"a": 1}),
        ];

        // Create a set of JSON rows that rotates each value past every field
        let json: Vec<_> = (0..json_values.len())
            .map(|i| {
                let pairs = json_values[i..]
                    .iter()
                    .chain(json_values[..i].iter())
                    .zip(&schema.fields)
                    .map(|(v, f)| (f.name().clone(), v.clone()))
                    .collect();
                serde_json::Value::Object(pairs)
            })
            .collect();
        let mut decoder = ReaderBuilder::new(Arc::new(schema))
            .with_ignore_type_conflicts(true)
            .with_coerce_primitive(true)
            .build_decoder()
            .unwrap();
        decoder.serialize(&json).unwrap();
        let batch = decoder.flush().unwrap().unwrap();
        assert_eq!(batch.num_rows(), 10);
        assert_eq!(batch.num_columns(), 10);

        // NOTE: NullArray doesn't materialize any values (they're all NULL by definition)
        let _ = batch
            .column(0)
            .as_any()
            .downcast_ref::<NullArray>()
            .unwrap();

        assert!(
            batch
                .column(1)
                .as_any()
                .downcast_ref::<BooleanArray>()
                .unwrap()
                .iter()
                .eq([
                    Some(true),
                    None,
                    None,
                    None,
                    None,
                    None,
                    None,
                    None,
                    None,
                    None
                ])
        );

        assert!(batch.column(2).as_primitive::<Int32Type>().iter().eq([
            Some(42),
            Some(1),
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None
        ]));

        assert!(batch.column(3).as_primitive::<Decimal128Type>().iter().eq([
            Some(1234),
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            Some(42000)
        ]));

        assert!(
            batch
                .column(4)
                .as_any()
                .downcast_ref::<StringArray>()
                .unwrap()
                .iter()
                .eq([
                    Some("hi"),
                    Some("ho"),
                    Some("1970-01-01T00:00:00+02:00"),
                    None,
                    None,
                    None,
                    None,
                    Some("true"),
                    Some("42"),
                    Some("1.234"),
                ])
        );

        assert!(
            batch
                .column(5)
                .as_any()
                .downcast_ref::<StringViewArray>()
                .unwrap()
                .iter()
                .eq([
                    Some("ho"),
                    Some("1970-01-01T00:00:00+02:00"),
                    None,
                    None,
                    None,
                    None,
                    Some("true"),
                    Some("42"),
                    Some("1.234"),
                    Some("hi"),
                ])
        );

        assert!(
            batch
                .column(6)
                .as_primitive::<TimestampSecondType>()
                .iter()
                .eq([
                    Some(-7200),
                    None,
                    None,
                    None,
                    None,
                    None,
                    Some(42),
                    None,
                    None,
                    None,
                ])
        );

        let arrays = batch
            .column(7)
            .as_any()
            .downcast_ref::<ListArray>()
            .unwrap();
        assert_eq!(
            arrays.nulls(),
            Some(&NullBuffer::from(
                &[
                    true, false, false, false, false, false, false, false, false, false
                ][..]
            ))
        );
        assert_eq!(arrays.offsets()[1], 3);
        let array_values = arrays
            .values()
            .as_any()
            .downcast_ref::<Int32Array>()
            .unwrap();
        assert!(array_values.iter().eq([Some(1), None, Some(3)]));

        let maps = batch.column(8).as_any().downcast_ref::<MapArray>().unwrap();
        assert_eq!(
            maps.nulls(),
            Some(&NullBuffer::from(
                // Both map and struct can parse
                &[
                    true, true, false, false, false, false, false, false, false, false
                ][..]
            ))
        );
        let map_keys = maps.keys().as_any().downcast_ref::<StringArray>().unwrap();
        assert!(map_keys.iter().eq([Some("k"), Some("a")]));
        let map_values = maps
            .values()
            .as_any()
            .downcast_ref::<StringArray>()
            .unwrap();
        assert!(map_values.iter().eq([Some("value"), Some("1")]));

        let structs = batch
            .column(9)
            .as_any()
            .downcast_ref::<StructArray>()
            .unwrap();
        assert_eq!(
            structs.nulls(),
            Some(&NullBuffer::from(
                // Both map and struct can parse
                &[
                    true, false, false, false, false, false, false, false, false, true
                ][..]
            ))
        );
        let struct_fields = structs
            .column(0)
            .as_any()
            .downcast_ref::<Int32Array>()
            .unwrap();
        assert!(struct_fields.slice(0, 2).iter().eq([Some(1), None]));
    }

    #[test]
    fn test_type_conflict_non_nullable() {
        let fields = [
            Field::new("bool", DataType::Boolean, false),
            Field::new("primitive", DataType::Int32, false),
            Field::new("numeric", DataType::Decimal128(10, 3), false),
            Field::new("string", DataType::Utf8, false),
            Field::new("string_view", DataType::Utf8View, false),
            Field::new(
                "timestamp",
                DataType::Timestamp(TimeUnit::Second, None),
                false,
            ),
            Field::new(
                "array",
                DataType::List(Arc::new(Field::new("item", DataType::Int32, true))),
                false,
            ),
            Field::new(
                "fixed_size_list",
                DataType::FixedSizeList(Arc::new(Field::new("item", DataType::Int32, true)), 2),
                false,
            ),
            Field::new(
                "map",
                DataType::Map(
                    Arc::new(Field::new(
                        Field::MAP_ENTRIES_FIELD_DEFAULT_NAME,
                        DataType::Struct(Fields::from(vec![
                            Field::new(Field::MAP_KEY_FIELD_DEFAULT_NAME, DataType::Utf8, false),
                            Field::new(Field::MAP_VALUE_FIELD_DEFAULT_NAME, DataType::Utf8, true),
                        ])),
                        false, // not nullable
                    )),
                    false, // not sorted
                ),
                false, // not nullable
            ),
            Field::new(
                "struct",
                DataType::Struct(Fields::from(vec![Field::new("a", DataType::Int32, true)])),
                false,
            ),
        ];

        // Every field above will have a type conflict with at least one of these values
        let json_values = vec![json!(true), json!({"a": 1})];

        for field in fields {
            let mut decoder = ReaderBuilder::new_with_field(field)
                .with_ignore_type_conflicts(true)
                .build_decoder()
                .unwrap();
            decoder.serialize(&json_values).unwrap();
            decoder
                .flush()
                .expect_err("type conflict on non-nullable type");
        }
    }

    #[test]
    fn test_ignore_type_conflicts_disabled() {
        let fields = [
            Field::new("null", DataType::Null, true),
            Field::new("bool", DataType::Boolean, true),
            Field::new("primitive", DataType::Int32, true),
            Field::new("numeric", DataType::Decimal128(10, 3), true),
            Field::new("string", DataType::Utf8, true),
            Field::new("string_view", DataType::Utf8View, true),
            Field::new(
                "timestamp",
                DataType::Timestamp(TimeUnit::Second, None),
                true,
            ),
            Field::new(
                "array",
                DataType::List(Arc::new(Field::new("item", DataType::Int32, true))),
                true,
            ),
            Field::new(
                "fixed_size_list",
                DataType::FixedSizeList(Arc::new(Field::new("item", DataType::Int32, true)), 2),
                true,
            ),
            Field::new(
                "map",
                DataType::Map(
                    Arc::new(Field::new(
                        Field::MAP_ENTRIES_FIELD_DEFAULT_NAME,
                        DataType::Struct(Fields::from(vec![
                            Field::new(Field::MAP_KEY_FIELD_DEFAULT_NAME, DataType::Utf8, false),
                            Field::new(Field::MAP_VALUE_FIELD_DEFAULT_NAME, DataType::Utf8, true),
                        ])),
                        false, // not nullable
                    )),
                    false, // not sorted
                ),
                true, // not nullable
            ),
            Field::new(
                "struct",
                DataType::Struct(Fields::from(vec![Field::new("a", DataType::Int32, true)])),
                true,
            ),
        ];

        // Every field above will have a type conflict with at least one of these values
        let json_values = vec![json!(true), json!({"a": 1})];

        for field in fields {
            let mut decoder = ReaderBuilder::new_with_field(field)
                .build_decoder()
                .unwrap();
            decoder.serialize(&json_values).unwrap();
            decoder
                .flush()
                .expect_err("type conflict on non-nullable type");
        }
    }

    #[test]
    fn test_read_run_end_encoded() {
        let buf = r#"
        {"a": "x"}
        {"a": "x"}
        {"a": "y"}
        {"a": "y"}
        {"a": "y"}
        "#;

        let ree_type = DataType::RunEndEncoded(
            Arc::new(Field::new(
                Field::REE_RUN_ENDS_FIELD_DEFAULT_NAME,
                DataType::Int32,
                false,
            )),
            Arc::new(Field::new(
                Field::REE_VALUES_FIELD_DEFAULT_NAME,
                DataType::Utf8,
                true,
            )),
        );
        let schema = Arc::new(Schema::new(vec![Field::new("a", ree_type, true)]));
        let batches = do_read(buf, 1024, false, false, schema);
        assert_eq!(batches.len(), 1);

        let col = batches[0].column(0);
        let run_array = col.as_run::<arrow_array::types::Int32Type>();

        // 5 logical values compressed into 2 runs
        assert_eq!(run_array.len(), 5);
        assert_eq!(run_array.run_ends().values(), &[2, 5]);

        let values = run_array.values().as_string::<i32>();
        assert_eq!(values.len(), 2);
        assert_eq!(values.value(0), "x");
        assert_eq!(values.value(1), "y");
    }

    #[test]
    fn test_read_run_end_encoded_consecutive_nulls() {
        let buf = r#"
        {"a": "x"}
        {}
        {}
        {}
        {"a": "y"}
        "#;

        let ree_type = DataType::RunEndEncoded(
            Arc::new(Field::new(
                Field::REE_RUN_ENDS_FIELD_DEFAULT_NAME,
                DataType::Int32,
                false,
            )),
            Arc::new(Field::new(
                Field::REE_VALUES_FIELD_DEFAULT_NAME,
                DataType::Utf8,
                true,
            )),
        );
        let schema = Arc::new(Schema::new(vec![Field::new("a", ree_type, true)]));
        let batches = do_read(buf, 1024, false, false, schema);
        assert_eq!(batches.len(), 1);

        let col = batches[0].column(0);
        let run_array = col.as_run::<arrow_array::types::Int32Type>();

        // 5 logical values: "x", null, null, null, "y" → 3 runs
        assert_eq!(run_array.len(), 5);
        assert_eq!(run_array.run_ends().values(), &[1, 4, 5]);

        let values = run_array.values().as_string::<i32>();
        assert_eq!(values.len(), 3);
        assert_eq!(values.value(0), "x");
        assert!(values.is_null(1));
        assert_eq!(values.value(2), "y");
    }

    #[test]
    fn test_read_run_end_encoded_nullability() {
        for field_nullable in [false, true] {
            for values_nullable in [false, true] {
                let ree_type = DataType::RunEndEncoded(
                    Arc::new(Field::new(
                        Field::REE_RUN_ENDS_FIELD_DEFAULT_NAME,
                        DataType::Int32,
                        false,
                    )),
                    Arc::new(Field::new(
                        Field::REE_VALUES_FIELD_DEFAULT_NAME,
                        DataType::Utf8,
                        values_nullable,
                    )),
                );
                let schema = Arc::new(Schema::new(vec![Field::new("a", ree_type, field_nullable)]));

                for buf in [
                    r#"{"a": "x"}
                    {"a": null}
                    {"a": "y"}"#,
                    r#"{"a": "x"}
                    {}
                    {"a": "y"}"#,
                ] {
                    let mut decoder = ReaderBuilder::new(schema.clone()).build_decoder().unwrap();
                    let result = decoder.decode(buf.as_bytes()).and_then(|_| decoder.flush());

                    if field_nullable && values_nullable {
                        result.expect("REE field and values are both nullable");
                    } else {
                        let err = result.expect_err(
                            "REE nulls require both the field and values to be nullable",
                        );
                        assert!(
                            err.to_string().contains(
                                "Encountered nulls in non-nullable values of RunEndEncoded"
                            ),
                            "unexpected error: {err}"
                        );
                    }
                }
            }
        }
    }

    #[test]
    fn test_read_run_end_encoded_all_unique() {
        let buf = r#"
        {"a": 1}
        {"a": 2}
        {"a": 3}
        "#;

        let ree_type = DataType::RunEndEncoded(
            Arc::new(Field::new(
                Field::REE_RUN_ENDS_FIELD_DEFAULT_NAME,
                DataType::Int32,
                false,
            )),
            Arc::new(Field::new(
                Field::REE_VALUES_FIELD_DEFAULT_NAME,
                DataType::Int32,
                true,
            )),
        );
        let schema = Arc::new(Schema::new(vec![Field::new("a", ree_type, true)]));
        let batches = do_read(buf, 1024, false, false, schema);
        assert_eq!(batches.len(), 1);

        let col = batches[0].column(0);
        let run_array = col.as_run::<arrow_array::types::Int32Type>();

        // No compression: 3 unique values → 3 runs
        assert_eq!(run_array.len(), 3);
        assert_eq!(run_array.run_ends().values(), &[1, 2, 3]);
    }

    #[test]
    fn test_read_run_end_encoded_int16_run_ends() {
        let buf = r#"
        {"a": "x"}
        {"a": "x"}
        {"a": "y"}
        "#;

        let ree_type = DataType::RunEndEncoded(
            Arc::new(Field::new(
                Field::REE_RUN_ENDS_FIELD_DEFAULT_NAME,
                DataType::Int16,
                false,
            )),
            Arc::new(Field::new(
                Field::REE_VALUES_FIELD_DEFAULT_NAME,
                DataType::Utf8,
                true,
            )),
        );
        let schema = Arc::new(Schema::new(vec![Field::new("a", ree_type, true)]));
        let batches = do_read(buf, 1024, false, false, schema);
        assert_eq!(batches.len(), 1);

        let col = batches[0].column(0);
        let run_array = col.as_run::<arrow_array::types::Int16Type>();

        assert_eq!(run_array.len(), 3);
        assert_eq!(run_array.run_ends().values(), &[2i16, 3]);
    }

    #[test]
    fn test_read_nested_run_end_encoded() {
        let buf = r#"
        {"a": "x"}
        {"a": "x"}
        {"a": "y"}
        "#;

        // The outer REE compresses whole rows, while the inner REE compresses the
        // repeated string values produced by decoding those rows.
        let inner_type = DataType::RunEndEncoded(
            Arc::new(Field::new(
                Field::REE_RUN_ENDS_FIELD_DEFAULT_NAME,
                DataType::Int64,
                false,
            )),
            Arc::new(Field::new(
                Field::REE_VALUES_FIELD_DEFAULT_NAME,
                DataType::Utf8,
                true,
            )),
        );
        let outer_type = DataType::RunEndEncoded(
            Arc::new(Field::new(
                Field::REE_RUN_ENDS_FIELD_DEFAULT_NAME,
                DataType::Int64,
                false,
            )),
            Arc::new(Field::new(
                Field::REE_VALUES_FIELD_DEFAULT_NAME,
                inner_type,
                true,
            )),
        );
        let schema = Arc::new(Schema::new(vec![Field::new("a", outer_type, true)]));
        let batches = do_read(buf, 1024, false, false, schema);
        assert_eq!(batches.len(), 1);

        let col = batches[0].column(0);
        let outer = col.as_run::<arrow_array::types::Int64Type>();
        // Three logical rows compress to two outer runs: ["x", "x"] and ["y"].
        assert_eq!(outer.len(), 3);
        assert_eq!(outer.run_ends().values(), &[2, 3]);

        let nested = outer.values().as_run::<arrow_array::types::Int64Type>();
        // The physical values of the outer REE are themselves a two-element REE.
        assert_eq!(nested.len(), 2);
        assert_eq!(nested.run_ends().values(), &[1, 2]);

        let nested_values = nested.values().as_string::<i32>();
        assert_eq!(nested_values.len(), 2);
        assert_eq!(nested_values.value(0), "x");
        assert_eq!(nested_values.value(1), "y");
    }

    #[test]
    fn test_flatten_top_level_arrays() {
        let buf = r#"
            [
                {"a": 1},
                {"a": 2}
            ]
            {"a": 3}
            [{"a": 4}, {"a": 5}, {"a": 6}]
            "#;

        let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Int32, true)]));
        let batches = do_read_config(buf, ReaderBuilder::new(schema).with_flatten(true));
        assert_eq!(batches.len(), 1);

        let col = batches[0].column(0);
        let col = col.as_any().downcast_ref::<Int32Array>().unwrap();
        assert_eq!(col.len(), 6);
        for i in 0..6 {
            assert_eq!(col.value(i), (i as i32) + 1);
        }
    }

    /// Declining everything must be indistinguishable from no factory at all.
    #[test]
    fn test_decoder_factory_declining_everything_is_transparent() {
        #[derive(Debug, Default)]
        struct DeclineAll {
            seen: Mutex<Vec<FieldRef>>,
        }

        impl DecoderFactory for DeclineAll {
            fn make_default_decoder(
                &self,
                _ctx: &DecoderContext,
                field: &FieldRef,
                _is_nullable: bool,
            ) -> Result<Option<Box<dyn ArrayDecoder>>, ArrowError> {
                self.seen.lock().unwrap().push(field.clone());
                Ok(None)
            }
        }

        let schema = Arc::new(Schema::new(vec![
            Field::new("a", DataType::Int32, true),
            Field::new("b", DataType::Utf8, true),
            Field::new_list("c", Field::new("item", DataType::Float64, true), true),
        ]));
        let buf = r#"{"a": 1, "b": "x", "c": [1.5, 2.5]}
        {"a": null, "c": []}
        "#;

        let read = |factory: Option<Arc<dyn DecoderFactory>>| {
            let mut builder = ReaderBuilder::new(schema.clone());
            if let Some(factory) = factory {
                builder = builder.with_decoder_factory(factory);
            }
            builder
                .build(Cursor::new(buf.as_bytes()))
                .unwrap()
                .next()
                .unwrap()
                .unwrap()
        };

        let factory = Arc::new(DeclineAll::default());
        assert_eq!(read(None), read(Some(factory.clone())));

        // Consulted for the root struct and every leaf, with names attached; the
        // root is the documented synthesized nameless struct
        let seen = factory.seen.lock().unwrap();
        assert!(
            matches!(seen[0].data_type(), DataType::Struct(_)),
            "{seen:?}"
        );
        assert_eq!(seen[0].name(), "", "root field should be nameless");

        let by_name: Vec<(&str, &DataType)> = seen
            .iter()
            .map(|f| (f.name().as_str(), f.data_type()))
            .collect();
        for expected in [
            ("a", &DataType::Int32),
            ("b", &DataType::Utf8),
            ("item", &DataType::Float64),
        ] {
            assert!(
                by_name.contains(&expected),
                "{expected:?} missing from {by_name:?}"
            );
        }
    }

    /// A custom decoder's output is not trusted: the arrays built from it are
    /// constructed without further validation, so a mismatch must be caught here.
    #[test]
    fn test_decoder_factory_output_is_validated() {
        /// Returns either the wrong data type or the wrong number of rows
        struct Bad {
            wrong_type: bool,
        }

        impl ArrayDecoder for Bad {
            fn decode(&mut self, _tape: &Tape<'_>, pos: &[u32]) -> Result<ArrayRef, ArrowError> {
                Ok(match self.wrong_type {
                    true => Arc::new(StringViewArray::from(vec!["x"; pos.len()])) as ArrayRef,
                    false => Arc::new(StringArray::from(Vec::<&str>::new())),
                })
            }
        }

        #[derive(Debug)]
        struct BadFactory {
            wrong_type: bool,
        }

        impl DecoderFactory for BadFactory {
            fn make_default_decoder(
                &self,
                _ctx: &DecoderContext,
                field: &FieldRef,
                _is_nullable: bool,
            ) -> Result<Option<Box<dyn ArrayDecoder>>, ArrowError> {
                match field.data_type() {
                    DataType::Utf8 => Ok(Some(Box::new(Bad {
                        wrong_type: self.wrong_type,
                    }))),
                    _ => Ok(None),
                }
            }
        }

        let read = |wrong_type: bool| {
            let schema = Arc::new(Schema::new(vec![Field::new("s", DataType::Utf8, true)]));
            ReaderBuilder::new(schema)
                .with_decoder_factory(Arc::new(BadFactory { wrong_type }))
                .build(Cursor::new(br#"{"s": "hello"}"#.as_slice()))
                .unwrap()
                .next()
                .unwrap()
                .unwrap_err()
                .to_string()
        };

        let err = read(true);
        assert!(
            err.contains("returned Utf8View for a field of type Utf8"),
            "{err}"
        );

        let err = read(false);
        assert!(err.contains("returned 0 values for 1 rows"), "{err}");
    }

    #[test]
    fn test_non_finite_floats() {
        let schema = infer_json_schema_from_iterator(std::iter::once(Ok(json!({
            "metadata": {"year": 0.0}, "values": [0.0], "text": ""
        }))))
        .unwrap();
        let builder = ReaderBuilder::new(Arc::new(schema));
        let input = r#"{"metadata":{"year":NaN},"values":[Infinity,-Infinity,null],"text":"NaN \"Infinity\""}"#;
        let mut reader = builder.clone().build(input.as_bytes()).unwrap();
        assert!(reader.next().unwrap().is_err());

        // Existing helper covers small I/O buffers and multiple batch sizes.
        let builder = builder.with_allow_non_finite_floats(true);
        let batches = do_read_config(&format!("{input}\n{input}"), builder.clone());
        let batch = &batches[0];
        let year = batch
            .column_by_name("metadata")
            .unwrap()
            .as_struct()
            .column_by_name("year")
            .unwrap()
            .as_primitive::<Float64Type>();
        assert_eq!(year.null_count(), 0);
        assert!(year.values().iter().all(|value| value.is_nan()));
        let lists = batch.column_by_name("values").unwrap().as_list::<i32>();
        let text = batch.column_by_name("text").unwrap().as_string::<i32>();
        for row in 0..batch.num_rows() {
            let values = lists.value(row);
            let values = values.as_primitive::<Float64Type>();
            assert_eq!(values.value(0), f64::INFINITY);
            assert_eq!(values.value(1), f64::NEG_INFINITY);
            assert!(values.is_null(2));
            assert_eq!(text.value(row), r#"NaN "Infinity""#);
        }
    }
}