//! [`Table`] — a catalog-bound Iceberg table handle with ergonomic append,
//! read-to-Arrow, and per-table SQL.

use std::sync::Arc;

use arrow_array::RecordBatch;
use arrow_schema::SchemaRef as ArrowSchemaRef;
use iceberg::table::Table as IceTable;
use iceberg::{Catalog as _, TableIdent};
use nornir_catalog::RedbCatalog;

use crate::error::Result;
use crate::read::{arrow_schema_of, read_all, scan_count};
use crate::write::{append, ingest, ingest_parallel, IngestStats};

/// A handle to one Iceberg table inside a [`crate::Warehouse`]. Writes go
/// through the catalog (`fast_append` snapshots); the handle tracks the table
/// state of its **own** last commit — call [`Table::refresh`] to pick up
/// commits made through other handles.
pub struct Table {
    catalog: Arc<RedbCatalog>,
    inner: IceTable,
}

impl Table {
    pub(crate) fn new(catalog: Arc<RedbCatalog>, inner: IceTable) -> Self {
        Table { catalog, inner }
    }

    /// The table's identifier (`namespace` + name).
    pub fn ident(&self) -> &TableIdent {
        self.inner.identifier()
    }

    /// The underlying `iceberg::table::Table` (escape hatch: `scan()`,
    /// metadata, snapshots, `file_io()`, …).
    pub fn inner(&self) -> &IceTable {
        &self.inner
    }

    /// The Arrow schema (with Iceberg field-id metadata) of the current schema.
    pub fn arrow_schema(&self) -> Result<ArrowSchemaRef> {
        arrow_schema_of(&self.inner)
    }

    /// Append `batches` as one `fast_append` snapshot (one commit). Batches are
    /// recast to the table schema automatically (see [`crate::recast`]).
    pub async fn append(&mut self, batches: &[RecordBatch]) -> Result<()> {
        self.inner = append(self.catalog.as_ref(), &self.inner, batches).await?;
        Ok(())
    }

    /// Bulk-ingest batches, committing every `batches_per_commit` batches.
    /// Returns throughput stats.
    pub async fn ingest(
        &mut self,
        batches: impl IntoIterator<Item = RecordBatch>,
        batches_per_commit: usize,
    ) -> Result<IngestStats> {
        let (table, stats) = ingest(
            self.catalog.as_ref(),
            self.inner.clone(),
            batches,
            batches_per_commit,
        )
        .await?;
        self.inner = table;
        Ok(stats)
    }

    /// Parallel bulk ingest — each `Vec<RecordBatch>` becomes one Parquet file,
    /// encoded across **all cores** (rayon) and committed by one sequential
    /// writer every `files_per_commit` files (the gatling topology). Single
    /// partition per call. Returns throughput stats.
    pub async fn ingest_parallel(
        &mut self,
        groups: Vec<Vec<RecordBatch>>,
        files_per_commit: usize,
    ) -> Result<IngestStats> {
        let (table, stats) =
            ingest_parallel(self.catalog.as_ref(), self.inner.clone(), groups, files_per_commit)
                .await?;
        self.inner = table;
        Ok(stats)
    }

    /// Full-scan the current snapshot into Arrow record batches.
    pub async fn read(&self) -> Result<Vec<RecordBatch>> {
        read_all(&self.inner).await
    }

    /// Row count of the current snapshot (full scan, batches not materialized).
    pub async fn count(&self) -> Result<u64> {
        scan_count(&self.inner).await
    }

    /// Reload the table from the catalog (pick up commits made elsewhere).
    pub async fn refresh(&mut self) -> Result<()> {
        self.inner = self.catalog.load_table(self.inner.identifier()).await?;
        Ok(())
    }
}

#[cfg(feature = "sql")]
mod sql {
    use super::*;
    use datafusion::prelude::SessionContext;
    use iceberg_datafusion::IcebergStaticTableProvider;

    impl Table {
        /// Run one SQL statement over **this** table, registered under its bare
        /// name (`SELECT … FROM <name> …`), against the handle's current
        /// snapshot. For multi-table SQL use [`crate::Warehouse::sql`].
        pub async fn sql(&self, query: &str) -> Result<Vec<RecordBatch>> {
            let ctx = SessionContext::new();
            let provider =
                IcebergStaticTableProvider::try_new_from_table(self.inner.clone()).await?;
            ctx.register_table(self.ident().name(), Arc::new(provider))?;
            Ok(ctx.sql(query).await?.collect().await?)
        }
    }
}
