//! [`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::{Schema as ArrowSchema, SchemaRef as ArrowSchemaRef};
use iceberg::table::Table as IceTable;
use iceberg::{Catalog as _, TableIdent};
use skade_katalog::RedbCatalog;

use crate::error::Result;
use crate::read::{
    DeltaPlan, ScanFilter, arrow_schema_of, read_all, read_delta, read_equality_deletes,
    read_filtered, read_limited, scan_count,
};
use parquet::basic::Compression;

use crate::write::{
    IngestStats, WriteProps, append_props, ingest_parallel_props, ingest_pipelined_props,
    ingest_props,
};

/// 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,
    /// Parquet `WriterProperties` knobs (compression, bloom filters, row-group
    /// size, dictionary) applied to this handle's writes. Default: uncompressed,
    /// dictionary on, no bloom, default row-group size.
    write_props: WriteProps,
}

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

    /// Set the Parquet compression codec for this handle's writes (builder form):
    /// `table.compression(Compression::ZSTD(Default::default()))`. Compression
    /// trades CPU for smaller files + less read I/O; in `ingest_parallel` it runs
    /// inside the all-core encode stage, so it's parallelised for free.
    pub fn compression(mut self, compression: Compression) -> Self {
        self.write_props.compression = compression;
        self
    }

    /// Set the compression codec in place (non-consuming).
    pub fn set_compression(&mut self, compression: Compression) {
        self.write_props.compression = compression;
    }

    /// Replace this handle's full [`WriteProps`] (compression + bloom filters +
    /// row-group size + dictionary) — builder form. Use this to enable
    /// point-lookup row-group skipping via bloom filters on chosen columns.
    ///
    /// ```no_run
    /// # use skade::{Table, WriteProps, Compression};
    /// # fn f(t: Table) -> Table {
    /// t.write_props(
    ///     WriteProps::new(Compression::ZSTD(Default::default()))
    ///         .bloom_columns(["symbol", "sha"])
    ///         .row_group_size(128 * 1024),
    /// )
    /// # }
    /// ```
    pub fn write_props(mut self, props: WriteProps) -> Self {
        self.write_props = props;
        self
    }

    /// Replace this handle's [`WriteProps`] in place (non-consuming).
    pub fn set_write_props(&mut self, props: WriteProps) {
        self.write_props = props;
    }

    /// This handle's current [`WriteProps`].
    pub fn props(&self) -> &WriteProps {
        &self.write_props
    }

    /// 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
    }

    pub(crate) fn catalog(&self) -> &Arc<RedbCatalog> {
        &self.catalog
    }

    pub(crate) fn set_inner(&mut self, inner: IceTable) {
        self.inner = 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_props(
            self.catalog.as_ref(),
            &self.inner,
            batches,
            &self.write_props,
        )
        .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_props(
            self.catalog.as_ref(),
            self.inner.clone(),
            batches,
            batches_per_commit,
            &self.write_props,
        )
        .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_props(
            self.catalog.as_ref(),
            self.inner.clone(),
            groups,
            files_per_commit,
            &self.write_props,
        )
        .await?;
        self.inner = table;
        Ok(stats)
    }

    /// Bulk ingest via the **no-barrier gatling pipeline**: encoders run ahead of
    /// one writer through a bounded channel (depth `channel_depth`), so all-core
    /// encode overlaps the writer's commit and memory stays bounded — vs
    /// [`ingest_parallel`](Self::ingest_parallel) which encodes everything first.
    /// Per-file partition; commits every `files_per_commit` files.
    pub async fn ingest_pipelined(
        &mut self,
        groups: Vec<Vec<RecordBatch>>,
        files_per_commit: usize,
        channel_depth: usize,
    ) -> Result<IngestStats> {
        let (table, stats) = ingest_pipelined_props(
            self.catalog.as_ref(),
            self.inner.clone(),
            groups,
            files_per_commit,
            channel_depth,
            &self.write_props,
        )
        .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
    }

    /// Like [`read`](Self::read) but projects only `columns` — column pruning is
    /// pushed into the scan, so only those columns are read from Parquet.
    pub async fn read_columns(&self, columns: &[&str]) -> Result<Vec<RecordBatch>> {
        crate::read::read_columns(&self.inner, columns).await
    }

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

    /// **Filtered / pushdown read.** Like [`read`](Self::read) but pushes a
    /// backend-neutral [`ScanFilter`] into the scan planner (partition / file /
    /// row-group pruning) and projects only `columns` (empty = all columns,
    /// preserving order). See [`read_filtered`] for the pruning semantics: it
    /// prunes at file granularity, not per row, so the result may still contain
    /// rows the predicate would reject — keep a residual per-row guard if you
    /// need exact filtering.
    ///
    /// ```no_run
    /// # use skade::{Table, ScanFilter};
    /// # async fn run(t: &Table) -> skade::Result<()> {
    /// // Read only the `znippy` partition, all columns.
    /// let rows = t.read_filtered(&ScanFilter::eq("repo", "znippy"), &[]).await?;
    /// # let _ = rows; Ok(()) }
    /// ```
    pub async fn read_filtered(
        &self,
        filter: &ScanFilter,
        columns: &[&str],
    ) -> Result<Vec<RecordBatch>> {
        read_filtered(&self.inner, filter, columns).await
    }

    /// Plan-time data-skipping stats (see [`crate::read::plan_stats`]): how many
    /// data files / rows survive pruning for `filter` (`None` = the whole table),
    /// computed from the scan plan without reading data. Diff a filtered plan
    /// against the `None` baseline to get files SKIPPED — the headline metric for
    /// the Iceberg manifest-pruning win.
    pub async fn plan_stats(
        &self,
        filter: Option<&ScanFilter>,
    ) -> Result<crate::read::ScanPlanStats> {
        crate::read::plan_stats(&self.inner, filter).await
    }

    /// **Limit / early-break streaming read.** Like [`read`](Self::read) but
    /// stops once `max_rows` rows are in hand instead of materializing the whole
    /// table — it drives the scan's Arrow stream and cancels the rest of the
    /// scan, so on a big table a preview reads only the first data file(s).
    /// Returns at most `max_rows` worth of rows (the last batch may carry the
    /// count slightly over; truncate if you need an exact count).
    /// `max_rows == 0` falls back to a full [`read`](Self::read). See
    /// [`read_limited`].
    pub async fn read_limited(&self, max_rows: usize) -> Result<Vec<RecordBatch>> {
        read_limited(&self.inner, max_rows).await
    }

    /// **Incremental read** — the rows appended between snapshots `from` and `to`
    /// (only the data files those snapshots added, not the whole table). See
    /// [`read_delta`] for the manifest-diff approach and its limitations.
    /// `from = None` reads from the beginning up to `to`. Returns the batches
    /// plus the [`DeltaPlan`] describing what was read.
    pub async fn read_delta(
        &self,
        from: Option<i64>,
        to: i64,
    ) -> Result<(Vec<RecordBatch>, DeltaPlan)> {
        read_delta(&self.inner, from, to).await
    }

    /// Read the deleted-row identities for a [`DeltaPlan`]'s equality-delete
    /// files (one `(equality_ids, batches)` per file). See
    /// [`read_equality_deletes`].
    pub async fn read_equality_deletes(
        &self,
        plan: &DeltaPlan,
    ) -> Result<Vec<(Vec<i32>, Vec<RecordBatch>)>> {
        read_equality_deletes(&self.inner, plan).await
    }

    /// The current snapshot id (the natural `to` for [`read_delta`]), or `None`
    /// if the table has no snapshots yet.
    pub fn current_snapshot_id(&self) -> Option<i64> {
        self.inner.metadata().current_snapshot().map(|s| s.snapshot_id())
    }

    /// 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(())
    }

    /// **Additive schema evolution.** Ensure the table carries every top-level
    /// column in `desired` (an Arrow schema, same mapping as
    /// [`crate::arrow_to_iceberg`]); any column the table is *missing* is added
    /// as an **optional** Iceberg column via an add-column metadata commit. This
    /// is idempotent and purely additive:
    ///
    /// * **Idempotent** — if every `desired` column is already present (by name)
    ///   it is a no-op (no catalog write); calling it twice in a row does nothing
    ///   the second time.
    /// * **Additive only** — columns are only *added*. Columns present in the
    ///   table but absent from `desired` are left untouched; nothing is ever
    ///   dropped or renamed. Type changes on existing columns are ignored (the
    ///   stored field wins) — this method only closes the "table is missing a
    ///   column the writer now produces" gap.
    /// * **Field-id stable** — every existing column keeps the exact field id it
    ///   was created with (the evolved schema reuses the table's current
    ///   `NestedField`s verbatim); each new column gets a fresh id above the
    ///   table's current `highest_field_id`. Old data files are untouched
    ///   (add-column is metadata-only), so rows written before the evolution
    ///   read back `null` for the new column, and rows written after carry it.
    ///
    /// Use this before [`append`](Self::append) when the writer always builds the
    /// full (current) batch but the table may have been created by an older
    /// binary with fewer columns — it evolves the stale table forward so the
    /// append's recast finds every column. After it returns, this handle points
    /// at the evolved table (no separate [`refresh`](Self::refresh) needed).
    ///
    /// New columns are added **optional** regardless of the `desired` field's
    /// nullability, because existing rows have no value for them (Iceberg
    /// add-column cannot make a column required without a default).
    ///
    /// ```no_run
    /// # use skade::arrow_schema::{DataType, Field, Schema};
    /// # async fn run(t: &mut skade::Table) -> skade::Result<()> {
    /// // Table was created with [id, name]; writer now also produces `score`.
    /// let desired = Schema::new(vec![
    ///     Field::new("id", DataType::Int64, false),
    ///     Field::new("name", DataType::Utf8, false),
    ///     Field::new("score", DataType::Float64, true), // new column
    /// ]);
    /// t.ensure_schema(&desired).await?; // adds `score`; no-op if already present
    /// # Ok(()) }
    /// ```
    pub async fn ensure_schema(&mut self, desired: &ArrowSchema) -> Result<()> {
        use std::collections::HashSet;

        use iceberg::TableUpdate;
        use iceberg::spec::{NestedField, NestedFieldRef, Schema};

        let current = self.inner.metadata().current_schema();
        let have: HashSet<&str> =
            current.as_struct().fields().iter().map(|f| f.name.as_str()).collect();

        // Names in `desired` the table does not yet have, in `desired` order.
        let missing_names: Vec<&str> = desired
            .fields()
            .iter()
            .map(|f| f.name().as_str())
            .filter(|n| !have.contains(n))
            .collect();
        if missing_names.is_empty() {
            // Steady state: every desired column is already present — no commit.
            return Ok(());
        }

        // Derive Iceberg types for the *desired* columns once, reusing the exact
        // Arrow→Iceberg mapping the table was created with. (Field ids here are
        // a throwaway `1..N`; we re-stamp ids below.)
        let desired_ice = crate::bridge::arrow_to_iceberg(desired)?;
        let desired_by_name: std::collections::HashMap<&str, &NestedFieldRef> = desired_ice
            .as_struct()
            .fields()
            .iter()
            .map(|f| (f.name.as_str(), f))
            .collect();

        // Evolved schema = the table's current fields verbatim (ids preserved) …
        let mut fields: Vec<NestedFieldRef> = current.as_struct().fields().to_vec();
        // … plus each missing column as a fresh OPTIONAL field above the current
        // highest id (add-column has no value for pre-existing rows).
        for (next_id, name) in (current.highest_field_id() + 1..).zip(missing_names.iter()) {
            let ty = desired_by_name
                .get(name)
                .map(|f| f.field_type.as_ref().clone())
                .ok_or_else(|| {
                    crate::error::SkadeError::other(format!(
                        "ensure_schema: desired column `{name}` vanished from derived schema"
                    ))
                })?;
            fields.push(Arc::new(NestedField::optional(next_id, *name, ty)));
        }

        // A new schema id (current + 1) so `AddSchema` registers a new schema
        // rather than colliding with the existing one; `-1` in SetCurrentSchema
        // = "the schema just added" (iceberg metadata-builder convention).
        let new_schema_id = current.schema_id() + 1;
        let evolved = Schema::builder()
            .with_schema_id(new_schema_id)
            .with_fields(fields)
            .build()?;

        let updates = vec![
            TableUpdate::AddSchema { schema: evolved },
            TableUpdate::SetCurrentSchema { schema_id: -1 },
        ];
        let ident = self.inner.identifier().clone();
        self.inner = self.catalog.commit_table(ident, Vec::new(), updates).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?)
        }
    }
}
