//! The read primitive: full table → Arrow in one call.

use std::sync::Arc;

use arrow_array::RecordBatch;
use arrow_schema::SchemaRef as ArrowSchemaRef;
use futures::TryStreamExt;
use iceberg::table::Table as IceTable;

use crate::error::Result;

/// The Arrow schema (with Iceberg field-id metadata) of `table`'s current
/// schema — the schema batches must carry to be written, and the schema scans
/// come back with.
pub fn arrow_schema_of(table: &IceTable) -> Result<ArrowSchemaRef> {
    let s = iceberg::arrow::schema_to_arrow_schema(table.metadata().current_schema())?;
    Ok(Arc::new(s))
}

/// Full-scan `table` (current snapshot, all columns) into Arrow record batches:
/// `scan().select_all().build()?.to_arrow()` as one ergonomic call.
pub async fn read_all(table: &IceTable) -> Result<Vec<RecordBatch>> {
    let stream = table.scan().select_all().build()?.to_arrow().await?;
    Ok(stream.try_collect().await?)
}

/// Full-scan `table` and count rows without materializing the batches.
pub async fn scan_count(table: &IceTable) -> Result<u64> {
    let stream = table.scan().select_all().build()?.to_arrow().await?;
    Ok(stream
        .try_fold(0u64, |acc, batch| async move { Ok(acc + batch.num_rows() as u64) })
        .await?)
}
