//! The writer stack as one call: ParquetWriter → RollingFileWriter →
//! DataFileWriter → `fast_append` commit.

use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{Duration, Instant};

use tokio::sync::{Semaphore, mpsc};

use arrow_array::{Array, LargeStringArray, RecordBatch, StringArray};
use arrow_schema::SchemaRef as ArrowSchemaRef;
use bytes::Bytes;
use iceberg::Catalog;
use iceberg::spec::{
    DataContentType, DataFile, DataFileFormat, Literal, PartitionKey, Struct,
    Transform,
};
use iceberg::table::Table as IceTable;
use iceberg::transaction::{ApplyTransactionAction, Transaction};
use iceberg::writer::base_writer::data_file_writer::DataFileWriterBuilder;
use iceberg::spec::SchemaRef as IceSchemaRef;
use iceberg::writer::file_writer::{ParquetWriter, ParquetWriterBuilder};
use iceberg::writer::file_writer::location_generator::{
    DefaultFileNameGenerator, DefaultLocationGenerator,
};
use iceberg::writer::file_writer::rolling_writer::RollingFileWriterBuilder;
use iceberg::writer::{IcebergWriter, IcebergWriterBuilder};
use parquet::arrow::ArrowWriter;
use parquet::basic::Compression;
use parquet::file::properties::WriterProperties;
use parquet::schema::types::ColumnPath;
use rayon::prelude::*;

use crate::bridge::recast;
use crate::error::{Result, SkadeError};
use crate::read::arrow_schema_of;

/// Tunable Parquet `WriterProperties` for skade writes — the knobs that live in
/// skade (compression, per-column bloom filters, row-group size, dictionary)
/// rather than in the catalog. These complement nornir's sort/`SortOrder`
/// file-level data-skipping with **intra-file row-group skipping** on point
/// lookups: a bloom filter on a high-cardinality column (e.g. `symbol`, `sha`)
/// lets a reader skip row groups whose chunk cannot contain the probed value.
///
/// Build with [`WriteProps::new`] (a chosen [`Compression`]) then the builder
/// setters, or `WriteProps::default()` for the bare Parquet defaults
/// (uncompressed, dictionary on, no bloom, default row-group size).
///
/// ```
/// # use skade::{WriteProps, Compression};
/// let props = WriteProps::new(Compression::ZSTD(Default::default()))
///     .bloom_columns(["symbol", "sha"])   // point-lookup row-group skipping
///     .row_group_size(128 * 1024)         // smaller groups → finer skipping
///     .dictionary(true);
/// # let _ = props;
/// ```
#[derive(Debug, Clone)]
pub struct WriteProps {
    /// Parquet compression codec (`UNCOMPRESSED` / `SNAPPY` / `ZSTD(level)` / …).
    pub compression: Compression,
    /// Columns to enable a per-column bloom filter on (by leaf column name).
    /// Empty (the default) writes no bloom filters.
    pub bloom_columns: Vec<String>,
    /// `set_max_row_group_size` when `Some`; otherwise the Parquet default
    /// (1,048,576 rows). Smaller groups → finer row-group skipping, more metadata.
    pub row_group_size: Option<usize>,
    /// Whether dictionary encoding is enabled (Parquet default: `true`).
    pub dictionary: bool,
}

impl Default for WriteProps {
    fn default() -> Self {
        WriteProps {
            compression: Compression::UNCOMPRESSED,
            bloom_columns: Vec::new(),
            row_group_size: None,
            dictionary: true,
        }
    }
}

impl WriteProps {
    /// A `WriteProps` with the given compression codec and otherwise Parquet
    /// defaults (dictionary on, no bloom, default row-group size).
    pub fn new(compression: Compression) -> Self {
        WriteProps {
            compression,
            ..Self::default()
        }
    }

    /// Enable a bloom filter on each named column (builder form). Replaces any
    /// previously set list.
    pub fn bloom_columns<I, S>(mut self, columns: I) -> Self
    where
        I: IntoIterator<Item = S>,
        S: Into<String>,
    {
        self.bloom_columns = columns.into_iter().map(Into::into).collect();
        self
    }

    /// Set `set_max_row_group_size` (builder form).
    pub fn row_group_size(mut self, rows: usize) -> Self {
        self.row_group_size = Some(rows);
        self
    }

    /// Set dictionary encoding on/off (builder form).
    pub fn dictionary(mut self, enabled: bool) -> Self {
        self.dictionary = enabled;
        self
    }

    /// Translate into Parquet [`WriterProperties`] used by both the iceberg
    /// `ParquetWriterBuilder` (single-file [`append`] path) and the in-memory
    /// `ArrowWriter` ([`ingest_parallel`]/[`ingest_pipelined`] encode path), so
    /// every write surface honours the same knobs.
    fn to_writer_properties(&self) -> WriterProperties {
        let mut b = WriterProperties::builder()
            .set_compression(self.compression)
            .set_dictionary_enabled(self.dictionary);
        if let Some(n) = self.row_group_size {
            // `set_max_row_group_row_count` is the un-deprecated name for the
            // old `set_max_row_group_size` (rows, not bytes) in parquet 58.
            b = b.set_max_row_group_row_count(Some(n));
        }
        for col in &self.bloom_columns {
            b = b.set_column_bloom_filter_enabled(ColumnPath::from(col.as_str()), true);
        }
        b.build()
    }
}

/// Process-unique data-file name prefix so repeated commits (and concurrent
/// writers in one process) never collide on file names.
fn unique_prefix() -> String {
    static SEQ: AtomicU64 = AtomicU64::new(0);
    let nanos = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_nanos() as u64)
        .unwrap_or(0);
    format!(
        "skade-{}-{nanos:x}-{}",
        std::process::id(),
        SEQ.fetch_add(1, Ordering::Relaxed)
    )
}

/// Throughput of an [`ingest`] run.
#[derive(Debug, Clone)]
pub struct IngestStats {
    pub rows: u64,
    pub commits: u64,
    pub elapsed: Duration,
}

impl IngestStats {
    pub fn rows_per_sec(&self) -> f64 {
        let s = self.elapsed.as_secs_f64();
        if s > 0.0 { self.rows as f64 / s } else { 0.0 }
    }
    pub fn commits_per_sec(&self) -> f64 {
        let s = self.elapsed.as_secs_f64();
        if s > 0.0 {
            self.commits as f64 / s
        } else {
            0.0
        }
    }
}

/// Derive the single [`PartitionKey`] for `batches` from `table`'s default
/// partition spec, or `None` for an unpartitioned table (the common case).
///
/// Supports **identity** transforms over **string** columns (`Utf8`/`LargeUtf8`).
/// Every row across every batch must share the same partition value — an
/// [`append`] commits one group of data files as one snapshot, so it writes a
/// single partition. A batch that spans partitions, or a non-identity/non-string
/// partition column, errors loudly rather than mis-tagging the files (which
/// would silently break partition pruning on read).
fn partition_key_for(table: &IceTable, batches: &[RecordBatch]) -> Result<Option<PartitionKey>> {
    let meta = table.metadata();
    let spec = meta.default_partition_spec();
    if spec.fields().is_empty() {
        return Ok(None);
    }
    let schema = meta.current_schema();
    let mut lits: Vec<Option<Literal>> = Vec::with_capacity(spec.fields().len());
    for f in spec.fields() {
        if f.transform != Transform::Identity {
            return Err(SkadeError::other(format!(
                "skade::append: only identity partition transforms are supported (got {:?} on `{}`)",
                f.transform, f.name
            )));
        }
        let src = schema.field_by_id(f.source_id).ok_or_else(|| {
            SkadeError::other(format!(
                "partition source field id {} not in schema",
                f.source_id
            ))
        })?;
        // Single partition value across every batch (None ⇒ all batches empty).
        let mut value: Option<String> = None;
        for b in batches {
            let col = b.column_by_name(&src.name).ok_or_else(|| {
                SkadeError::other(format!(
                    "partition column `{}` missing from batch",
                    src.name
                ))
            })?;
            let read = |i: usize| -> Result<&str> {
                if let Some(a) = col.as_any().downcast_ref::<StringArray>() {
                    Ok(a.value(i))
                } else if let Some(a) = col.as_any().downcast_ref::<LargeStringArray>() {
                    Ok(a.value(i))
                } else {
                    Err(SkadeError::other(format!(
                        "partition column `{}` is not a string (only Utf8/LargeUtf8 identity partitions)",
                        src.name
                    )))
                }
            };
            for i in 0..col.len() {
                let v = read(i)?;
                match &value {
                    None => value = Some(v.to_string()),
                    Some(v0) if v0 != v => {
                        return Err(SkadeError::other(format!(
                            "skade::append: batch spans multiple `{}` partitions (`{v0}` vs `{v}`); \
                             single-partition appends only",
                            src.name
                        )));
                    }
                    _ => {}
                }
            }
        }
        match value {
            Some(v) => lits.push(Some(Literal::string(v))),
            None => return Ok(None),
        }
    }
    Ok(Some(PartitionKey::new(
        spec.as_ref().clone(),
        schema.clone(),
        Struct::from_iter(lits),
    )))
}

/// Write `batches` into one group of Parquet data files and commit them as a
/// single `fast_append` snapshot. Batches are [`recast`] to the table's
/// field-id Arrow schema first (covers `Utf8 → LargeUtf8` etc.), so any
/// column-compatible batches work. For a partitioned table the (single)
/// partition key is derived from the batch via [`partition_key_for`]. Returns
/// the updated table.
pub async fn append(
    catalog: &dyn Catalog,
    table: &IceTable,
    batches: &[RecordBatch],
) -> Result<IceTable> {
    append_props(catalog, table, batches, &WriteProps::default()).await
}

/// Like [`append`] but with an explicit Parquet `Compression` codec
/// (`UNCOMPRESSED` / `SNAPPY` / `ZSTD(level)` / …). skade defaults to
/// uncompressed; compression trades CPU for smaller files + less read I/O.
/// Shorthand for [`append_props`] with compression-only [`WriteProps`].
pub async fn append_with(
    catalog: &dyn Catalog,
    table: &IceTable,
    batches: &[RecordBatch],
    compression: Compression,
) -> Result<IceTable> {
    append_props(catalog, table, batches, &WriteProps::new(compression)).await
}

/// Like [`append`] but with full [`WriteProps`] control — compression plus
/// per-column bloom filters, row-group size, and dictionary encoding. Use this
/// to enable point-lookup row-group skipping (`bloom_columns`).
pub async fn append_props(
    catalog: &dyn Catalog,
    table: &IceTable,
    batches: &[RecordBatch],
    props: &WriteProps,
) -> Result<IceTable> {
    let target = arrow_schema_of(table)?;
    let batches = recast(batches, target)?;
    let partition = partition_key_for(table, &batches)?;

    let schema = table.metadata().current_schema().clone();
    let data_location = format!("{}/data", table.metadata().location());
    let location_gen = DefaultLocationGenerator::with_data_location(data_location);
    let file_name_gen =
        DefaultFileNameGenerator::new(unique_prefix(), None, DataFileFormat::Parquet);

    let pw = ParquetWriterBuilder::new(props.to_writer_properties(), schema);
    let rolling = RollingFileWriterBuilder::new_with_default_file_size(
        pw,
        table.file_io().clone(),
        location_gen,
        file_name_gen,
    );
    let mut writer = DataFileWriterBuilder::new(rolling).build(partition).await?;
    for b in &batches {
        writer.write(b.clone()).await?;
    }
    let data_files = writer.close().await?;

    let tx = Transaction::new(table);
    let action = tx.fast_append().add_data_files(data_files);
    let tx = action.apply(tx)?;
    Ok(tx.commit(catalog).await?)
}

/// Ingest all batches, committing every `batches_per_commit` batches via
/// [`append`]. Returns the final table + throughput.
pub async fn ingest(
    catalog: &dyn Catalog,
    table: IceTable,
    batches: impl IntoIterator<Item = RecordBatch>,
    batches_per_commit: usize,
) -> Result<(IceTable, IngestStats)> {
    ingest_props(catalog, table, batches, batches_per_commit, &WriteProps::default()).await
}

/// Like [`ingest`] with an explicit Parquet `Compression` codec.
pub async fn ingest_with(
    catalog: &dyn Catalog,
    table: IceTable,
    batches: impl IntoIterator<Item = RecordBatch>,
    batches_per_commit: usize,
    compression: Compression,
) -> Result<(IceTable, IngestStats)> {
    ingest_props(
        catalog,
        table,
        batches,
        batches_per_commit,
        &WriteProps::new(compression),
    )
    .await
}

/// Like [`ingest`] with full [`WriteProps`] control (compression + bloom +
/// row-group size + dictionary). Each commit group is written via
/// [`append_props`].
pub async fn ingest_props(
    catalog: &dyn Catalog,
    table: IceTable,
    batches: impl IntoIterator<Item = RecordBatch>,
    batches_per_commit: usize,
    props: &WriteProps,
) -> Result<(IceTable, IngestStats)> {
    let bpc = batches_per_commit.max(1);
    let mut table = table;
    let (mut rows, mut commits) = (0u64, 0u64);
    let mut group: Vec<RecordBatch> = Vec::with_capacity(bpc);
    let start = Instant::now();
    for batch in batches {
        rows += batch.num_rows() as u64;
        group.push(batch);
        if group.len() >= bpc {
            table = append_props(catalog, &table, &group, props).await?;
            commits += 1;
            group.clear();
        }
    }
    if !group.is_empty() {
        table = append_props(catalog, &table, &group, props).await?;
        commits += 1;
    }
    Ok((
        table,
        IngestStats {
            rows,
            commits,
            elapsed: start.elapsed(),
        },
    ))
}

/// Parallel bulk ingest — the gatling topology: **all cores encode** (rayon),
/// **one** sequential writer commits.
///
/// Each inner `Vec<RecordBatch>` is encoded into one in-memory Parquet file
/// across the rayon pool (pure CPU, no I/O), then a single writer streams the
/// files to the table's `FileIO` and commits every `files_per_commit` files via
/// `fast_append`. Single-partition per call (same rule as [`append`]); for a
/// partitioned table every file is tagged with the derived partition key.
///
/// The encode phase holds all files in memory before the write phase, so size
/// `groups` to your memory budget for very large loads (chunk across calls).
pub async fn ingest_parallel(
    catalog: &dyn Catalog,
    table: IceTable,
    groups: Vec<Vec<RecordBatch>>,
    files_per_commit: usize,
) -> Result<(IceTable, IngestStats)> {
    ingest_parallel_props(catalog, table, groups, files_per_commit, &WriteProps::default()).await
}

/// Like [`ingest_parallel`] with an explicit Parquet `Compression` codec — the
/// compression runs inside the all-core encode stage, so it's parallelised free.
pub async fn ingest_parallel_with(
    catalog: &dyn Catalog,
    table: IceTable,
    groups: Vec<Vec<RecordBatch>>,
    files_per_commit: usize,
    compression: Compression,
) -> Result<(IceTable, IngestStats)> {
    ingest_parallel_props(
        catalog,
        table,
        groups,
        files_per_commit,
        &WriteProps::new(compression),
    )
    .await
}

/// Like [`ingest_parallel`] with full [`WriteProps`] control — compression,
/// per-column bloom filters, row-group size, and dictionary. All of it (bloom
/// hashing included) runs inside the all-core encode stage, so it's parallelised
/// for free across the rayon pool.
pub async fn ingest_parallel_props(
    catalog: &dyn Catalog,
    table: IceTable,
    groups: Vec<Vec<RecordBatch>>,
    files_per_commit: usize,
    props: &WriteProps,
) -> Result<(IceTable, IngestStats)> {
    let fpc = files_per_commit.max(1);
    let target = arrow_schema_of(&table)?;
    let start = Instant::now();

    // Recast each group to the table's field-id Arrow schema (so the Parquet
    // columns carry the field ids the scan maps back through).
    let groups: Vec<Vec<RecordBatch>> = groups
        .into_iter()
        .map(|g| recast(&g, target.clone()))
        .collect::<Result<_>>()?;

    // One partition key for the whole call (single-partition rule), derived from
    // every batch — `None` for an unpartitioned table.
    let all: Vec<RecordBatch> = groups.iter().flatten().cloned().collect();
    let partition = partition_key_for(&table, &all)?;
    drop(all);

    // ── Encode stage: all cores, pure CPU, no I/O. ──
    let mut encoded: Vec<(usize, Vec<u8>, u64)> = groups
        .into_par_iter()
        .enumerate()
        .map(|(seq, g)| {
            let (bytes, rows) = encode_parquet(target.clone(), &g, props)?;
            Ok::<_, SkadeError>((seq, bytes, rows))
        })
        .collect::<Result<Vec<_>>>()?;
    encoded.sort_by_key(|(seq, _, _)| *seq);

    // ── Writer stage: exactly ONE sequential writer + committer. ──
    let prefix = unique_prefix();
    let data_dir = format!("{}/data", table.metadata().location());
    let ice_schema = table.metadata().current_schema().clone();
    let mut table = table;
    let (mut rows, mut commits) = (0u64, 0u64);
    let mut pending: Vec<DataFile> = Vec::with_capacity(fpc);
    for (seq, bytes, record_count) in encoded {
        let path = format!("{data_dir}/{prefix}-{seq:08}.parquet");
        let bytes = Bytes::from(bytes);
        let len = bytes.len() as u64;
        let data_file = build_data_file(&ice_schema, &bytes, path.clone(), record_count, len, partition.as_ref())?;
        table
            .file_io()
            .new_output(&path)?
            .write(bytes)
            .await?;
        rows += record_count;
        pending.push(data_file);
        if pending.len() >= fpc {
            table = commit_files(catalog, &table, std::mem::take(&mut pending)).await?;
            commits += 1;
        }
    }
    if !pending.is_empty() {
        table = commit_files(catalog, &table, pending).await?;
        commits += 1;
    }
    Ok((
        table,
        IngestStats {
            rows,
            commits,
            elapsed: start.elapsed(),
        },
    ))
}

/// One Parquet file encoded in memory by a worker, handed to the lone writer.
struct EncodedFile {
    seq: usize,
    bytes: Vec<u8>,
    record_count: u64,
    partition: Option<iceberg::spec::PartitionKey>,
}

/// Bulk ingest with the **no-barrier gatling pipeline**: encoders (up to one per
/// core, semaphore-bounded) run **ahead** of one sequential writer through a
/// bounded channel, so the all-core Parquet encode **overlaps** the writer's
/// FileIO + commit, and memory is bounded to `channel_depth` in-flight files
/// (unlike [`ingest_parallel`], which encodes *all* files before writing).
///
/// Each inner `Vec<RecordBatch>` becomes one file; the partition key is derived
/// per file (so a partitioned table is tagged correctly even though files are
/// written in completion order, not input order). Commits every
/// `files_per_commit` files via `fast_append`.
pub async fn ingest_pipelined(
    catalog: &dyn Catalog,
    table: IceTable,
    groups: Vec<Vec<RecordBatch>>,
    files_per_commit: usize,
    channel_depth: usize,
    compression: Compression,
) -> Result<(IceTable, IngestStats)> {
    ingest_pipelined_props(
        catalog,
        table,
        groups,
        files_per_commit,
        channel_depth,
        &WriteProps::new(compression),
    )
    .await
}

/// Like [`ingest_pipelined`] with full [`WriteProps`] control — compression,
/// per-column bloom filters, row-group size, and dictionary, all applied inside
/// the overlapped all-core encode stage.
pub async fn ingest_pipelined_props(
    catalog: &dyn Catalog,
    table: IceTable,
    groups: Vec<Vec<RecordBatch>>,
    files_per_commit: usize,
    channel_depth: usize,
    props: &WriteProps,
) -> Result<(IceTable, IngestStats)> {
    let fpc = files_per_commit.max(1);
    let target = arrow_schema_of(&table)?;
    let start = Instant::now();
    // Immutable handle for the encoders (schema + partition spec don't change
    // during the ingest); the writer advances its own `table` via commits.
    let meta = Arc::new(table.clone());
    // Shared, owned write props for the spawned encoder tasks.
    let props = Arc::new(props.clone());
    let workers = std::thread::available_parallelism()
        .map(|n| n.get())
        .unwrap_or(1)
        .max(1);
    let (tx, mut rx) = mpsc::channel::<EncodedFile>(channel_depth.max(1));
    let sem = Arc::new(Semaphore::new(workers));

    // ── Encode stage: bounded fan-out; CPU on spawn_blocking, send when done. ──
    let mut encoders = Vec::with_capacity(groups.len());
    for (seq, group) in groups.into_iter().enumerate() {
        let (tx, sem, target, meta, props) = (
            tx.clone(),
            sem.clone(),
            target.clone(),
            meta.clone(),
            props.clone(),
        );
        encoders.push(tokio::spawn(async move {
            let _permit = sem.acquire_owned().await.expect("encode semaphore closed");
            let ef = tokio::task::spawn_blocking(move || -> Result<EncodedFile> {
                let batches = recast(&group, target.clone())?;
                let partition = partition_key_for(&meta, &batches)?;
                let (bytes, record_count) = encode_parquet(target, &batches, &props)?;
                Ok(EncodedFile {
                    seq,
                    bytes,
                    record_count,
                    partition,
                })
            })
            .await
            .expect("encode task panicked")?;
            // Backpressure: blocks here when the writer is `channel_depth` behind.
            let _ = tx.send(ef).await;
            Ok::<(), SkadeError>(())
        }));
    }
    drop(tx); // channel closes once every encoder's clone is dropped

    // ── Writer stage: exactly ONE task does all FileIO writes + commits. ──
    let prefix = unique_prefix();
    let data_dir = format!("{}/data", table.metadata().location());
    let ice_schema = table.metadata().current_schema().clone();
    let mut table = table;
    let (mut rows, mut commits) = (0u64, 0u64);
    let mut pending: Vec<DataFile> = Vec::with_capacity(fpc);
    while let Some(ef) = rx.recv().await {
        let path = format!("{data_dir}/{prefix}-{:08}.parquet", ef.seq);
        let bytes = Bytes::from(ef.bytes);
        let len = bytes.len() as u64;
        // Derive per-column bounds from this file's Parquet footer (see
        // `ingest_parallel_props`): identical stats path to `append`, so the
        // hand-committed DataFile is file-prunable.
        let data_file = build_data_file(
            &ice_schema,
            &bytes,
            path.clone(),
            ef.record_count,
            len,
            ef.partition.as_ref(),
        )?;
        table
            .file_io()
            .new_output(&path)?
            .write(bytes)
            .await?;
        rows += ef.record_count;
        pending.push(data_file);
        if pending.len() >= fpc {
            table = commit_files(catalog, &table, std::mem::take(&mut pending)).await?;
            commits += 1;
        }
    }
    if !pending.is_empty() {
        table = commit_files(catalog, &table, pending).await?;
        commits += 1;
    }
    // Surface any encoder error (the channel already let the writer drain).
    for e in encoders {
        e.await.expect("encoder task panicked")?;
    }
    Ok((
        table,
        IngestStats {
            rows,
            commits,
            elapsed: start.elapsed(),
        },
    ))
}

/// Encode a group of batches into one in-memory Parquet file — CPU only, the
/// part that fans out across cores. `schema` must carry the iceberg field ids.
fn encode_parquet(
    schema: ArrowSchemaRef,
    batches: &[RecordBatch],
    props: &WriteProps,
) -> Result<(Vec<u8>, u64)> {
    let mut buf: Vec<u8> = Vec::new();
    let mut rows = 0u64;
    {
        let mut w = ArrowWriter::try_new(&mut buf, schema, Some(props.to_writer_properties()))?;
        for b in batches {
            rows += b.num_rows() as u64;
            w.write(b)?;
        }
        w.close()?;
    }
    Ok((buf, rows))
}

/// Build the `DataFile` for one already-encoded in-memory Parquet file, with the
/// per-column Iceberg statistics (`lower_bounds`/`upper_bounds`, `value_counts`,
/// `null_value_counts`, `column_sizes`, `split_offsets`) derived from the file's
/// own Parquet footer via [`ParquetWriter::data_file_builder_from_parquet_bytes`]
/// — the SAME statistics path the streaming [`append`]/`DataFileWriter` close
/// uses. This is what makes the fast `ingest_parallel`/`ingest_pipelined` paths
/// produce file-prunable data files (matching `append`); previously they emitted
/// only `record_count` + `file_size`, so the scan planner could never skip them.
///
/// `ice_schema` is the table's current Iceberg schema (carries field ids).
/// Bounds for column types the iceberg stats path can't serialize are simply
/// omitted by that path (no bound = no pruning = correct), never guessed.
fn build_data_file(
    ice_schema: &IceSchemaRef,
    bytes: &Bytes,
    path: String,
    record_count: u64,
    file_size: u64,
    partition: Option<&PartitionKey>,
) -> Result<DataFile> {
    let mut b =
        ParquetWriter::data_file_builder_from_parquet_bytes(ice_schema.clone(), bytes, path)
            .map_err(SkadeError::other)?;
    // The stats helper already sets content/format/record_count/file_size, but we
    // re-assert record_count + file_size from the writer's own accounting (the
    // FileIO-written byte length and the rows we counted) so they always match
    // the bytes actually committed, independent of footer parsing.
    b.content(DataContentType::Data)
        .file_format(DataFileFormat::Parquet)
        .record_count(record_count)
        .file_size_in_bytes(file_size);
    if let Some(pk) = partition {
        // Override the empty partition struct the stats helper defaults to.
        b.partition(pk.data().clone());
        b.partition_spec_id(pk.spec().spec_id());
    }
    b.build().map_err(SkadeError::other)
}

/// One `fast_append` commit of already-written data files.
async fn commit_files(
    catalog: &dyn Catalog,
    table: &IceTable,
    files: Vec<DataFile>,
) -> Result<IceTable> {
    let tx = Transaction::new(table);
    let action = tx.fast_append().add_data_files(files);
    let tx = action.apply(tx)?;
    Ok(tx.commit(catalog).await?)
}
