//! Schema bridges: widen → Iceberg roundtrip → un-widen, bit-exact for
//! unsigned values above the signed max; windowed Parquet reads.

use std::sync::Arc;

use anyhow::Result;
use skade::arrow_array::{Int64Array, RecordBatch, UInt32Array, UInt64Array};
use skade::arrow_schema::{DataType, Field, Schema};

fn unsigned_schema() -> Schema {
    Schema::new(vec![
        Field::new("u32", DataType::UInt32, false),
        Field::new("u64", DataType::UInt64, false),
    ])
}

fn unsigned_batch() -> Result<RecordBatch> {
    // Values past the signed max — the cases plain casts would null/refuse.
    Ok(RecordBatch::try_new(
        Arc::new(unsigned_schema()),
        vec![
            Arc::new(UInt32Array::from(vec![0u32, 7, u32::MAX, u32::MAX - 1])),
            Arc::new(UInt64Array::from(vec![
                0u64,
                7,
                u64::MAX,
                (i64::MAX as u64) + 1,
            ])),
        ],
    )?)
}

#[test]
fn widen_unwiden_is_bit_exact() -> Result<()> {
    let original = unsigned_batch()?;

    let widened = skade::widen_for_iceberg(&original)?;
    assert_eq!(widened.column(0).data_type(), &DataType::Int32);
    assert_eq!(widened.column(1).data_type(), &DataType::Int64);
    // u64::MAX reinterprets to -1.
    let w64 = widened
        .column(1)
        .as_any()
        .downcast_ref::<Int64Array>()
        .unwrap();
    assert_eq!(w64.value(2), -1);

    let back = skade::unwiden(&[widened], original.schema())?;
    assert_eq!(back.len(), 1);
    assert_eq!(
        &back[0], &original,
        "widen→unwiden roundtrip must be bit-exact"
    );
    Ok(())
}

/// Unsigned columns survive a full Iceberg write/read cycle via widen → store
/// (signed) → scan → unwiden.
#[tokio::test]
async fn unsigned_roundtrip_through_iceberg() -> Result<()> {
    let tmp = tempfile::tempdir()?;
    let wh = skade::open(tmp.path().join("lake")).await?;

    let original = unsigned_batch()?;
    let widened = skade::widen_for_iceberg(&original)?;

    let mut t = wh.create_table("idx", widened.schema().as_ref()).await?;
    t.append(&[widened]).await?;

    let stored = t.read().await?;
    let back = skade::unwiden(&stored, original.schema())?;
    let total: usize = back.iter().map(|b| b.num_rows()).sum();
    assert_eq!(total, original.num_rows());

    // Sort-insensitive check: collect u64 column values from all batches.
    let mut got: Vec<u64> = back
        .iter()
        .flat_map(|b| {
            b.column(1)
                .as_any()
                .downcast_ref::<UInt64Array>()
                .unwrap()
                .values()
                .to_vec()
        })
        .collect();
    got.sort_unstable();
    let mut want = vec![0u64, 7, u64::MAX, (i64::MAX as u64) + 1];
    want.sort_unstable();
    assert_eq!(got, want);
    Ok(())
}

#[test]
fn windowed_parquet_reads() -> Result<()> {
    use skade::parquet::arrow::ArrowWriter;
    use skade::parquet::file::properties::WriterProperties;

    let tmp = tempfile::tempdir()?;
    let path = tmp.path().join("rows.parquet");

    // 10 row groups × 100 rows.
    let schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Int64, false)]));
    let props = WriterProperties::builder()
        .set_max_row_group_size(100)
        .build();
    let file = std::fs::File::create(&path)?;
    let mut w = ArrowWriter::try_new(file, schema.clone(), Some(props))?;
    let ids: Vec<i64> = (0..1000).collect();
    w.write(&RecordBatch::try_new(
        schema,
        vec![Arc::new(Int64Array::from(ids))],
    )?)?;
    w.close()?;

    let (ice, file_schema, rg_rows) = skade::parquet_layout(&path)?;
    assert_eq!(ice.as_struct().fields().len(), 1);
    assert_eq!(file_schema.fields().len(), 1);
    assert_eq!(rg_rows, vec![100usize; 10]);

    // ~300-row windows over the first 800 rows → 8 row groups in 3 windows.
    let windows = skade::rowgroup_windows(&rg_rows, 300, 800);
    assert_eq!(windows.len(), 3);
    assert_eq!(windows.iter().map(|w| w.len()).sum::<usize>(), 8);

    let mut total = 0usize;
    for w in &windows {
        let batches = skade::read_row_groups(&path, w, 64, 4)?;
        total += batches.iter().map(|b| b.num_rows()).sum::<usize>();
    }
    assert_eq!(total, 800);
    Ok(())
}
