genson-core 1.0.2

use super::*;
use serde_json::json;

#[test]
fn test_normalise_record() {
    let schema = json!({
        "type": "record",
        "name": "doc",
        "fields": [
            {"name": "id", "type": "string"},
            {"name": "labels", "type": {"type": "map", "values": "string"}},
        ]
    });

    let cfg = NormaliseConfig::default();
    let input = json!({"id": 42}); // id is number, labels missing
    let normalised = normalise_value(input, &schema, &cfg, None);

    assert_eq!(normalised, json!({"id": "42", "labels": Value::Null}));
}

#[test]
fn test_normalise_array_union() {
    let schema = json!(["null", {"type": "array", "items": "string"}]);
    let cfg = NormaliseConfig::default();

    let input = json!("hello"); // scalar string
    let normalised = normalise_value(input, &schema, &cfg, None);

    assert_eq!(normalised, json!(["hello"]));
}

#[test]
fn test_empty_map_to_null() {
    let schema = json!({"type": "map", "values": "string"});
    let cfg = NormaliseConfig::default();

    let input = json!({});
    let normalised = normalise_value(input, &schema, &cfg, None);

    assert_eq!(normalised, Value::Null);
}

#[test]
fn test_empty_map_preserved_if_flag_off() {
    let schema = json!({"type": "map", "values": "string"});
    let cfg = NormaliseConfig {
        empty_as_null: false,
        ..NormaliseConfig::default()
    };

    let input = json!({});
    let normalised = normalise_value(input, &schema, &cfg, None);

    assert_eq!(normalised, json!({}));
}

#[test]
fn test_string_coercion_toggle() {
    let schema = json!({
        "type": "record",
        "name": "doc",
        "fields": [
            {"name": "int_field", "type": "int"},
            {"name": "bool_field", "type": "boolean"},
        ]
    });

    let input = json!({
        "int_field": "42",
        "bool_field": "true"
    });

    // Default: coerce_string = false
    let cfg_no_coerce = NormaliseConfig {
        empty_as_null: true,
        coerce_string: false,
        ..NormaliseConfig::default()
    };
    let norm_no_coerce = normalise_value(input.clone(), &schema, &cfg_no_coerce, None);
    assert_eq!(
        norm_no_coerce,
        json!({
            "int_field": Value::Null,     // stays null because string not coerced
            "bool_field": Value::Null     // same here
        })
    );

    // With coerce_string = true
    let cfg_coerce = NormaliseConfig {
        empty_as_null: true,
        coerce_string: true,
        ..NormaliseConfig::default()
    };
    let norm_coerce = normalise_value(input, &schema, &cfg_coerce, None);
    assert_eq!(
        norm_coerce,
        json!({
            "int_field": 42,
            "bool_field": true
        })
    );
}

#[test]
fn test_normalise_map_of_records() {
    // Schema: map<string, record{language:string, value:string}>
    let schema = json!({
        "type": "map",
        "values": {
            "type": "object",
            "properties": {
                "language": { "type": "string" },
                "value": { "type": "string" }
            },
            "required": ["language", "value"]
        }
    });

    // Input data
    let input = json!({
        "en": { "language": "en", "value": "Hello" },
        "fr": { "language": "fr", "value": "Bonjour" }
    });

    let cfg = NormaliseConfig::default();

    let normalised = normalise_value(input, &schema, &cfg, None);

    // Expect same shape back (since it's already valid against schema)
    let expected = json!({
        "en": { "language": "en", "value": "Hello" },
        "fr": { "language": "fr", "value": "Bonjour" }
    });

    assert_eq!(normalised, expected);
}

#[test]
fn test_normalise_map_of_records_with_null() {
    // Same schema as before
    let schema = json!({
        "type": "map",
        "values": {
            "type": "object",
            "properties": {
                "language": { "type": "string" },
                "value": { "type": "string" }
            },
            "required": ["language", "value"]
        }
    });

    // Input with a null value (should normalise to null for that entry)
    let input = json!({
        "en": { "language": "en", "value": "Hello" },
        "fr": null
    });

    let cfg = NormaliseConfig::default();

    let normalised = normalise_value(input, &schema, &cfg, None);

    let expected = json!({
        "en": { "language": "en", "value": "Hello" },
        "fr": null
    });

    assert_eq!(normalised, expected);
}

/// A small deterministic generator for the property test below.
struct Rng(u64);

impl Rng {
    fn below(&mut self, n: u64) -> u64 {
        self.0 ^= self.0 << 13;
        self.0 ^= self.0 >> 7;
        self.0 ^= self.0 << 17;
        self.0 % n
    }
}

/// A random document nesting records in records and arrays. A record's keys extend its
/// own key by a letter (`a`, `ab`, `abc`), so rows share keys where they share a path but
/// no two paths do: genson's Avro schema names a record after its key, and two records of
/// one name but different fields are a case `normalise` does not handle and this test is
/// not about. An array's items are records, or else all of one scalar type.
fn random_document(rng: &mut Rng, key: &str, depth: u32) -> Value {
    let pick = if depth == 0 { 3 + rng.below(4) } else { rng.below(7) };
    match pick {
        0 | 1 => random_record(rng, key, depth),
        2 => {
            let scalar = [json!(1), json!("s"), json!(true)][rng.below(3) as usize].clone();
            let records = rng.below(2) == 0;
            let items = (0..rng.below(4)).map(|_| {
                if records {
                    random_record(rng, key, depth)
                } else {
                    scalar.clone()
                }
            });
            Value::Array(items.collect())
        }
        3 => Value::Null,
        4 => json!(rng.below(10)),
        5 => json!("s"),
        _ => json!(true),
    }
}

fn random_record(rng: &mut Rng, key: &str, depth: u32) -> Value {
    let n = 1 + rng.below(3);
    let fields = (0..n).map(|_| {
        let key = format!("{key}{}", ["a", "b", "c"][rng.below(3) as usize]);
        let value = random_document(rng, &key, depth - 1);
        (key, value)
    });
    Value::Object(fields.collect())
}

/// Every key `random_document` can write below the root, to prune by.
fn random_document_keys() -> Vec<String> {
    let mut keys = vec![String::new()];
    let mut all = Vec::new();
    for _ in 0..4 {
        keys = keys
            .iter()
            .flat_map(|k| ["a", "b", "c"].map(|l| format!("{k}{l}")))
            .collect();
        all.extend(keys.iter().cloned());
    }
    all
}

/// The input value at `path` from a row's root. A value that is not an array where the
/// schema has one is normalised as an array of itself, so its index 0 is itself.
fn at<'a>(value: &'a Value, path: &[PathSegment]) -> Option<&'a Value> {
    path.iter().try_fold(value, |v, segment| match segment {
        PathSegment::Key(k) => v.get(k),
        PathSegment::Index(0) if !v.is_array() => Some(v),
        PathSegment::Index(i) => v.get(i),
    })
}

/// For random rows and prune names: every pruned value is the input as it was at its path
/// (taking a record's fields out of its input leaves the records that are pruned whole), and
/// names that no record holds prune nothing and change nothing.
#[cfg(feature = "avro")]
#[test]
fn test_pruned_values_are_the_input_at_their_paths() {
    use crate::{infer_json_schema_from_strings, SchemaInferenceConfig};
    let mut rng = Rng(0x2545_f491_4f6c_dd1d);
    let cfg = NormaliseConfig::default();
    let mut pruned_seen = 0;
    for _ in 0..500 {
        let rows: Vec<Value> = (0..1 + rng.below(6))
            .map(|_| json!({"r": random_document(&mut rng, "", 4)}))
            .collect();
        let strings: Vec<String> = rows.iter().map(Value::to_string).collect();
        let config = SchemaInferenceConfig {
            avro: true,
            ..Default::default()
        };
        let Ok(inferred) = infer_json_schema_from_strings(&strings, config) else {
            continue;
        };
        let prune: HashSet<String> = random_document_keys()
            .into_iter()
            .filter(|_| rng.below(8) == 0)
            .collect();

        let (out, pruned) = normalise_values_pruned(rows.clone(), &inferred.schema, &cfg, &prune);
        assert_eq!(out.len(), rows.len());
        for p in &pruned {
            assert_eq!(at(&rows[p.row], &p.path), Some(&p.value), "{strings:?} pruning {prune:?}");
        }
        pruned_seen += pruned.len();

        let absent: HashSet<String> = ["zz".to_string()].into();
        let (unpruned, none) = normalise_values_pruned(rows.clone(), &inferred.schema, &cfg, &absent);
        assert!(none.is_empty());
        assert_eq!(unpruned, normalise_values(rows, &inferred.schema, &cfg), "{strings:?}");
    }
    assert!(pruned_seen > 100, "the rows should prune something: {pruned_seen}");
}


/// `normalise_values_pruned` (by its compiled plan) and the reference that reads the schema
/// for every value, on the same rows, schema, options and prune names: equal rows and
/// pruned values, or the same panic. Returns how many values were pruned.
fn assert_plan_matches(
    rows: &[Value],
    schema: &Value,
    cfg: &NormaliseConfig,
    prune: &HashSet<String>,
) -> usize {
    use std::panic::{catch_unwind, AssertUnwindSafe};
    let message = |e: Box<dyn std::any::Any + Send>| {
        e.downcast_ref::<&str>()
            .map(|s| s.to_string())
            .or_else(|| e.downcast_ref::<String>().cloned())
    };
    let plan = catch_unwind(AssertUnwindSafe(|| {
        normalise_values_pruned(rows.to_vec(), schema, cfg, prune)
    }))
    .map_err(message);
    let reference = catch_unwind(AssertUnwindSafe(|| {
        normalise_values_pruned_by_schema(rows.to_vec(), schema, cfg, prune)
    }))
    .map_err(message);
    assert_eq!(
        plan, reference,
        "rows {rows:?}\nschema {schema}\nprune {prune:?}"
    );
    plan.map_or(0, |(_, pruned)| pruned.len())
}

/// Every option combination `assert_plan_matches` is run with.
fn normalise_configs() -> Vec<NormaliseConfig> {
    let mut configs = Vec::new();
    for map_encoding in [
        MapEncoding::Mapping,
        MapEncoding::Entries,
        MapEncoding::KeyValueEntries,
    ] {
        for (empty_as_null, coerce_string) in [(true, false), (false, true)] {
            for wrap_root in [None, Some("w".to_string())] {
                configs.push(NormaliseConfig {
                    empty_as_null,
                    coerce_string,
                    map_encoding,
                    wrap_root,
                });
            }
        }
    }
    configs
}

/// For random rows and schemas inferred from them (records, arrays, maps, promoted scalars,
/// unions), with and without prune names, under every option: the compiled plan normalises
/// and prunes as reading the schema does, also for rows of another batch than the schema's.
#[cfg(feature = "avro")]
#[test]
fn test_plan_matches_the_schema_reading_path() {
    use crate::{infer_json_schema_from_strings, SchemaInferenceConfig};
    let mut rng = Rng(0x9e37_79b9_7f4a_7c15);
    let configs = normalise_configs();
    let keys = random_document_keys();
    let mut previous: Vec<Value> = Vec::new();
    let mut pruned_seen = 0;
    for case in 0..2000 {
        let rows: Vec<Value> = (0..1 + rng.below(6))
            .map(|_| {
                let mut row = json!({"r": random_document(&mut rng, "", 4)});
                // Strings that coerce, floats beside integers, and an empty map
                match rng.below(5) {
                    0 => row["n"] = json!(["1", "x", "2.5", "true"][rng.below(4) as usize]),
                    1 => {
                        row["n"] = [json!(1.5), json!(2), json!(-3)][rng.below(3) as usize].clone()
                    }
                    2 => row["m"] = json!({}),
                    _ => {}
                }
                row
            })
            .collect();
        let strings: Vec<String> = rows.iter().map(Value::to_string).collect();
        let config = SchemaInferenceConfig {
            avro: true,
            map_threshold: [20, 2, 0][case % 3],
            unify_maps: case % 2 == 0,
            wrap_scalars: case % 4 != 3,
            ..Default::default()
        };
        let Ok(inferred) = infer_json_schema_from_strings(&strings, config) else {
            continue;
        };
        let prune: HashSet<String> = keys.iter().filter(|_| rng.below(8) == 0).cloned().collect();
        let cfg = &configs[case % configs.len()];
        for prune in [HashSet::new(), prune] {
            pruned_seen += assert_plan_matches(&rows, &inferred.schema, cfg, &prune);
            assert_plan_matches(&previous, &inferred.schema, cfg, &prune);
        }
        previous = rows;
    }
    assert!(
        pruned_seen > 100,
        "the rows should prune something: {pruned_seen}"
    );
}

/// Schemas inference does not write, for the kinds of node normalisation treats specially:
/// names and other types it passes values through, records without `fields` or with
/// fields lacking a name or type, unions of only `null` and empty ones (which panic on a
/// value as before), and promoted records.
#[test]
fn test_plan_matches_on_unusual_schemas() {
    let schemas = [
        json!("null"),
        json!("Named"),
        json!({"type": "string"}),
        json!({"type": "fixed", "size": 2}),
        json!(7),
        json!({"type": "record"}),
        json!({"type": "record", "fields": {"a": "int"}}),
        json!({"type": "record", "fields": [
            {"name": "a"},
            {"type": "int"},
            "x",
            {"name": 1, "type": "int"},
            {"name": "b", "type": "long"},
        ]}),
        json!(["null"]),
        json!([]),
        json!(["string", "int"]),
        json!(["int", "null", "string"]),
        json!({"type": "array"}),
        json!({"type": "map"}),
        json!({"type": "array", "items": {"type": "map", "values": ["null", "double"]}}),
        json!({"type": "record", "name": "p", "fields": [
            {"name": "v__string", "type": ["null", "string"]},
            {"name": "v__int", "type": ["null", "long"]},
            {"name": "v__float", "type": ["null", "double"]},
            {"name": "v__boolean", "type": ["null", "boolean"]},
        ]}),
        json!({"type": "record", "name": "r", "fields": [
            {"name": "a", "type": {"type": "map", "values": {
                "type": "record",
                "name": "q",
                "fields": [
                    {"name": "x", "type": ["null", "string"]},
                    {"name": "y", "type": {"type": "array", "items": "boolean"}},
                ],
            }}},
            {"name": "b", "type": ["null", {"type": "array", "items": "int"}]},
        ]}),
    ];
    let values = [
        json!(null),
        json!(1),
        json!(-2),
        json!(1.5),
        json!("s"),
        json!("12"),
        json!("true"),
        json!(true),
        json!([]),
        json!([1, "a", null]),
        json!({}),
        json!({"a": 1, "b": [2], "v__string": "s"}),
        json!({"a": {"k": {"x": 1, "y": true}, "e": {}}, "b": 3}),
        json!({"a": "scalar", "b": null}),
    ];
    let prunes: [HashSet<String>; 3] = [
        HashSet::new(),
        ["x".to_string(), "v__int".to_string()].into(),
        ["b".to_string(), "a".to_string()].into(),
    ];
    for schema in &schemas {
        for cfg in &normalise_configs() {
            for prune in &prunes {
                // One value at a time, so that a value that panics leaves the others tested
                for value in &values {
                    assert_plan_matches(std::slice::from_ref(value), schema, cfg, prune);
                }
                assert_plan_matches(&values, schema, cfg, prune);
            }
        }
    }
}

/// Random JSON text for the typed decoder: objects with keys from a small set (some
/// promoted, some repeated or escaped), arrays, numbers in every form `serde_json` reads
/// differently, escaped strings, and whitespace of every kind between tokens.
#[cfg(feature = "parquet")]
fn random_text(rng: &mut Rng, depth: u32, out: &mut String) {
    let ws = |rng: &mut Rng, out: &mut String| {
        if rng.below(4) == 0 {
            out.push_str([" ", "\n", "\t", "\r", "  "][rng.below(5) as usize]);
        }
    };
    let keys = [
        "a",
        "b",
        "c",
        "id",
        "v__string",
        "v__int",
        "v__float",
        "v__boolean",
        "v",
    ];
    ws(rng, out);
    // Now and then something `serde_json` rejects, making the row invalid
    if rng.below(60) == 0 {
        out.push_str(
            [
                "0123",
                "-012",
                "00",
                "1.",
                ".5",
                "+1",
                "1e",
                "--1",
                "nul",
                "tru",
                "\"\\x\"",
                "\"\\ud800\"",
                "\"\\udc00x\"",
                "\"a\tb\"",
                "[1,]",
                "{\"a\":1,}",
                "{\"a\" 1}",
                "1e999",
            ][rng.below(18) as usize],
        );
        return;
    }
    let pick = if depth == 0 {
        3 + rng.below(6)
    } else {
        rng.below(9)
    };
    match pick {
        0 | 1 => {
            out.push('{');
            let n = rng.below(5);
            for i in 0..n {
                if i > 0 {
                    out.push(',');
                }
                ws(rng, out);
                let key = keys[rng.below(keys.len() as u64) as usize];
                match rng.below(12) {
                    0 => out.push_str(&format!("\"\\u{:04x}{}\"", key.as_bytes()[0], &key[1..])),
                    _ => out.push_str(&format!("\"{key}\"")),
                }
                ws(rng, out);
                out.push(':');
                random_text(rng, depth - 1, out);
            }
            ws(rng, out);
            out.push('}');
        }
        2 => {
            out.push('[');
            let n = rng.below(4);
            for i in 0..n {
                if i > 0 {
                    out.push(',');
                }
                random_text(rng, depth - 1, out);
            }
            ws(rng, out);
            out.push(']');
        }
        3 => out.push_str("null"),
        4 => out.push_str(["true", "false"][rng.below(2) as usize]),
        5 | 6 => out.push_str(
            [
                "0",
                "-0",
                "12",
                "-12",
                "1.5",
                "-0.0",
                "1e2",
                "-1.25E-3",
                "2.5e+3",
                "123456789012345678901234",
                "9223372036854775807",
                "9223372036854775808",
                "-9223372036854775808",
                "-9223372036854775809",
                "18446744073709551615",
                "18446744073709551616",
                "999999999999999999",
                "1000000000000000000",
                "0.1",
                "100",
            ][rng.below(20) as usize],
        ),
        _ => out.push_str(
            [
                "\"s\"",
                "\"\"",
                "\"a\\nb\"",
                "\"\\u00e9t\\u00e9\"",
                "\"\\ud83d\\ude00\"",
                "\"q\\\"q\"",
                "\"12\"",
                "\"-3\"",
                "\"2.5\"",
                "\"true\"",
                "\"0\"",
                "\"nan\"",
                "\"inf\"",
                "\"1e999\"",
                "\"caf\u{e9}\"",
            ][rng.below(15) as usize],
        ),
    }
    ws(rng, out);
}

/// The typed rows the `Value` path gives: each row parsed (`null` if invalid), normalised
/// and decoded with `values_to_struct_array`; `Err` if any of it panics or fails.
#[cfg(feature = "parquet")]
fn typed_by_value(
    rows: &[Option<String>],
    plan: &NormalisePlan<'_>,
    fields: &arrow::datatypes::Fields,
) -> Result<(arrow::array::StructArray, Vec<PrunedValue>), String> {
    use std::panic::{catch_unwind, AssertUnwindSafe};
    catch_unwind(AssertUnwindSafe(|| {
        let mut values = Vec::new();
        let mut pruned = Vec::new();
        for (i, row) in rows.iter().enumerate() {
            let Some(row) = row else {
                values.push(Value::Null);
                continue;
            };
            let v = serde_json::from_str(row).unwrap_or(Value::Null);
            let (normalised, mut row_pruned) = plan.normalise(v);
            row_pruned.iter_mut().for_each(|p| p.row = i);
            pruned.extend(row_pruned);
            values.push(normalised);
        }
        crate::parquet::values_to_struct_array(&values, fields).map(|a| (a, pruned))
    }))
    .unwrap_or_else(|_| Err("panicked".to_string()))
}

/// For random JSON text (valid, invalid and null rows), schemas inferred from the valid
/// rows (wrapped, with a field beside them or not, or as they are; with all of them or all
/// but the last) and every option and prune set: the typed decoder gives the arrays and
/// pruned values the `Value` path gives, or declines (`None`) where that path fails or
/// might differ, and declines rarely.
#[cfg(feature = "parquet")]
#[test]
fn test_typed_decode_matches_the_value_path() {
    use crate::{infer_json_schema_from_strings, SchemaInferenceConfig};
    let mut rng = Rng(0xd1b5_4a32_d192_ed03);
    let configs: Vec<NormaliseConfig> = normalise_configs()
        .into_iter()
        .filter(|c| c.map_encoding != MapEncoding::Entries)
        .collect();
    let prunes: [HashSet<String>; 4] = [
        HashSet::new(),
        ["v__string".to_string()].into(),
        ["a".to_string(), "v__int".to_string()].into(),
        ["b".to_string()].into(),
    ];
    let (mut decoded, mut declined, mut failing, mut pruned_seen) = (0, 0, 0, 0);
    let mut decoded_unwrapped = 0;
    for case in 0..2500 {
        // Mostly with each row wrapped as `w`, else the rows as they are: then mostly objects,
        // for a record at the root
        let wrap = case % 5 != 4;
        let mut rows: Vec<Option<String>> = Vec::new();
        for _ in 0..1 + rng.below(6) {
            let mut text = String::new();
            if !wrap && rng.below(8) != 0 {
                text.push_str("{\"a\":");
                random_text(&mut rng, 3, &mut text);
                text.push_str(",\"v\":");
                random_text(&mut rng, 3, &mut text);
                text.push('}');
            } else {
                random_text(&mut rng, 4, &mut text);
            }
            let text = match rng.below(14) {
                0 => text[..text.len() / 2].to_string(),
                1 => format!("{text} x"),
                2 => format!("{text}\n{text}"),
                _ => text,
            };
            rows.push(if rng.below(15) == 0 { None } else { Some(text) });
        }
        // Now and then the schema is inferred without the last row, which then need not fit
        // it, or with a field beside the wrapped row
        let inferred_from = match case % 7 {
            3 => &rows[..rows.len() - 1],
            _ => &rows[..],
        };
        let beside = case % 6 == 1;
        let strings: Vec<String> = inferred_from
            .iter()
            .flatten()
            .filter_map(|r| serde_json::from_str::<Value>(r).ok())
            .map(|v| match (wrap, beside) {
                (true, true) => json!({"w": v, "a": "x"}),
                (true, false) => json!({"w": v}),
                (false, _) => v,
            })
            .map(|v| v.to_string())
            .collect();
        if strings.is_empty() {
            continue;
        }
        let config = SchemaInferenceConfig {
            avro: true,
            map_threshold: [20, 2, 0][case % 3],
            unify_maps: case % 2 == 0,
            wrap_scalars: case % 4 != 3,
            ..Default::default()
        };
        let Ok(inferred) = infer_json_schema_from_strings(&strings, config) else {
            continue;
        };
        let cfg = &configs[case % configs.len()];
        let cfg = NormaliseConfig {
            wrap_root: wrap.then(|| "w".to_string()),
            ..cfg.clone()
        };
        if inferred.schema.get("type") != Some(&json!("record")) {
            continue;
        }
        let prune = &prunes[case % prunes.len()];
        let mut out_schema = inferred.schema.clone();
        prune_schema(&mut out_schema, prune);
        let native = cfg.map_encoding == MapEncoding::Mapping;
        let Ok(fields) = crate::parquet::avro_record_fields(&out_schema, native) else {
            continue;
        };
        let plan = NormalisePlan::new(&inferred.schema, &cfg, prune);
        let reference = typed_by_value(&rows, &plan, &fields);
        let borrowed: Vec<Option<&str>> = rows.iter().map(Option::as_deref).collect();
        let typed = plan
            .to_arrow(&fields)
            .and_then(|p| p.normalise(&borrowed, 0));
        match (typed, reference) {
            (Some(typed), Ok(reference)) => {
                assert_eq!(
                    typed.0, reference.0,
                    "rows {rows:?}\nschema {}",
                    inferred.schema
                );
                assert_eq!(
                    typed.1, reference.1,
                    "rows {rows:?}\nschema {}",
                    inferred.schema
                );
                pruned_seen += typed.1.len();
                decoded += 1;
                decoded_unwrapped += usize::from(!wrap);
            }
            (Some(typed), Err(e)) => {
                panic!(
                    "decoded {typed:?} where the Value path fails ({e}): rows {rows:?}\nschema {}",
                    inferred.schema
                )
            }
            (None, Ok(_)) => declined += 1,
            (None, Err(_)) => failing += 1,
        }
    }
    eprintln!("typed decode: decoded {decoded} ({decoded_unwrapped} unwrapped), declined {declined}, failing {failing}, pruned {pruned_seen}");
    assert!(
        decoded > 500,
        "decoded {decoded}, declined {declined}, failing {failing}"
    );
    assert!(
        declined * 20 < decoded,
        "decoded {decoded}, declined {declined}"
    );
    assert!(pruned_seen > 20, "pruned {pruned_seen}");
    assert!(
        decoded_unwrapped > 50,
        "decoded {decoded_unwrapped} unwrapped"
    );
}

/// Rows nested 127 deep are read, and 128 deep are invalid (null), as `serde_json` reads
/// them: in a string field, the one array written out and the other row null.
#[cfg(feature = "parquet")]
#[test]
fn test_typed_decode_depth_limit() {
    let schema = json!({
        "type": "record",
        "name": "document",
        "fields": [{"name": "w", "type": ["null", "string"]}]
    });
    let cfg = NormaliseConfig {
        wrap_root: Some("w".to_string()),
        ..Default::default()
    };
    let prune = HashSet::new();
    let plan = NormalisePlan::new(&schema, &cfg, &prune);
    let fields = crate::parquet::avro_record_fields(&schema, false).unwrap();
    let nested = |depth: usize| Some(format!("{}1{}", "[".repeat(depth), "]".repeat(depth)));
    let rows = vec![nested(127), nested(128)];
    let borrowed: Vec<Option<&str>> = rows.iter().map(Option::as_deref).collect();
    let typed = plan
        .to_arrow(&fields)
        .and_then(|p| p.normalise(&borrowed, 0))
        .unwrap();
    assert_eq!(typed, typed_by_value(&rows, &plan, &fields).unwrap());
    let w = typed
        .0
        .column(0)
        .as_any()
        .downcast_ref::<arrow::array::StringArray>()
        .unwrap();
    assert_eq!(w.value(0), nested(127).unwrap());
    assert!(arrow::array::Array::is_null(w, 1));
}

/// A record pruned by a field after one that could panic: where that earlier field holds a
/// value (inference only saw it null), the `Value` path panics before it reaches the
/// pruning, so the typed decoder declines; where it is null, both prune the record. A field
/// of an empty union panics even on null.
#[cfg(feature = "parquet")]
#[test]
fn test_typed_decode_declines_a_prune_after_a_panic() {
    let schema = |a_type: Value| {
        json!({
            "type": "record",
            "name": "document",
            "fields": [
                {"name": "a", "type": a_type},
                {"name": "b", "type": ["null", {
                    "type": "record",
                    "name": "b",
                    "fields": [
                        {"name": "x", "type": ["null", "long"]},
                        {"name": "y", "type": ["null", "string"]}
                    ]
                }]}
            ]
        })
    };
    let cfg = NormaliseConfig::default();
    let prune: HashSet<String> = ["x".to_string()].into();
    let row = |text: &str| vec![Some(text.to_string())];
    let (held, null) = (
        row(r#"{"a": 1, "b": {"x": 1}}"#),
        row(r#"{"a": null, "b": {"x": 1}}"#),
    );
    for (a_type, null_prunes) in [(json!(["null"]), true), (json!([]), false)] {
        let schema = schema(a_type);
        let plan = NormalisePlan::new(&schema, &cfg, &prune);
        let mut out_schema = schema.clone();
        prune_schema(&mut out_schema, &prune);
        let fields = crate::parquet::avro_record_fields(&out_schema, false).unwrap();
        let decode = |rows: &[Option<String>]| {
            let borrowed: Vec<Option<&str>> = rows.iter().map(Option::as_deref).collect();
            plan.to_arrow(&fields)
                .and_then(|p| p.normalise(&borrowed, 0))
        };
        assert!(typed_by_value(&held, &plan, &fields).is_err());
        assert!(decode(&held).is_none());
        if null_prunes {
            let typed = decode(&null).unwrap();
            assert_eq!(typed.1.len(), 1);
            assert_eq!(typed, typed_by_value(&null, &plan, &fields).unwrap());
        } else {
            assert!(typed_by_value(&null, &plan, &fields).is_err());
            assert!(decode(&null).is_none());
        }
    }
}