geopackage 0.2.0

//! The feature/attribute write path: [`FeatureWriter`] and the batched
//! [`Layer::write_all`] helper.
//!
//! # Transaction shape
//!
//! [`Layer::writer`] returns a [`FeatureWriter`] that **owns its transaction**
//! (opened with rusqlite's `unchecked_transaction`, so it works on the shared
//! `&Connection` the read path already uses). Writes stage into that
//! transaction; [`FeatureWriter::commit`] flushes the `gpkg_contents`
//! `last_change` and bounding box, then commits. Dropping a writer without
//! committing rolls the transaction back. rusqlite types never appear in the
//! public API: geometry is `impl geo_traits::GeometryTrait<T = f64>` and
//! non-geometry values are the crate's [`Value`] enum.
//!
//! An owned transaction (rather than a caller-passed transaction object) keeps
//! the escape-hatch `rusqlite::Transaction` out of the public surface and lets
//! the writer maintain the running bounding-box fold and `last_change` at one
//! commit point. The raw connection ([`crate::GeoPackage::connection`]) remains
//! available for callers who want to drive their own transaction.
//!
//! # Bounding box and `last_change`
//!
//! The writer seeds a bounding-box fold from the existing `gpkg_contents` row
//! and unions each written geometry's XY envelope into it (a cheap running
//! fold, never a rescan). Deletes do not shrink the box: an over-estimate is
//! spec-legal, and shrinking would need a rescan. On commit, a non-empty fold
//! is written back and `last_change` is refreshed to the strict 1.4 datetime
//! form via SQLite's `strftime` (matching the normative column default).
//!
//! # Envelopes and Z/M
//!
//! Every written geometry gets a GPB envelope (XY, or XYZ when it carries Z),
//! so a reader, and the rtree triggers that ask for four bounds a row, never
//! have to decode the WKB body to get them; encoding is delegated to
//! [`geopackage_core::geometry::encode_gpb`]. A geometry's `z`/`m` presence is
//! validated against the column's [`ZmFlag`] before encoding, so a violation
//! is a typed [`Error::ZmViolation`] rather than a malformed row.
//!
//! # Spatial indexes
//!
//! Individual `insert`/`update`/`delete` calls, and the per-batch
//! [`Layer::write_all`] path, go through ordinary SQL, so a table that already
//! carries the rtree triggers has its index maintained by those triggers (the
//! `ST_*` functions are registered on every connection).
//!
//! [`Layer::write_all`] additionally takes the bulk path when it writes a large
//! batch into an indexed layer: it drops the triggers, inserts the rows without
//! per-row index maintenance, brings the index up to date in one operation, and
//! reinstalls the triggers. How the index is brought up to date depends on the
//! size of the write against the size of the index, and is chosen once the rows
//! are written and both counts are known: a write large enough to be worth it
//! rebuilds the index outright (see [`crate::bulk`]), and a smaller one adds the
//! new entries to the existing index instead. The threshold at which the whole
//! path engages, and forcing it either way, are controlled by
//! [`BulkIndexOptions`] via [`Layer::write_all_with`].
//!
//! # Atomicity of the bulk path
//!
//! The bulk `write_all` is a single transaction: dropping the triggers, every
//! row insert, the `gpkg_contents` flush, the index work at the end (rebuild or
//! append), and reinstalling the triggers all commit together. A crash or an
//! error at any point rolls the whole thing back to the state before the call,
//! so the rows can never be committed against an index that was not brought up
//! to date with them.
//!
//! This was not always so. The rebuild used to run in its own transaction
//! because it built the index in an `ATTACH`ed scratch database and `ATTACH`
//! requires autocommit, which left a window where a crash committed the rows but
//! not the index. Building the tree directly ([`crate::packed`]) removed the
//! `ATTACH` and with it the window. [`Layer::spatial_index_status`] and
//! [`Layer::repair_spatial_index`] still exist and still recover a
//! [`crate::SpatialIndexStatus::Stale`] index, since a file can arrive from
//! anywhere, but this path no longer produces one.

use geo_traits::{Dimensions, GeometryTrait};
use geopackage_core::geometry::encode_gpb;
#[cfg(feature = "arrow")]
use geopackage_core::geometry::encode_gpb_from_wkb;
use geopackage_core::ident::quote;
use geopackage_core::triggers;
use geopackage_core::types::ZmFlag;
use rusqlite::types::Value as SqlValue;
use rusqlite::{Connection, OptionalExtension, Transaction, params_from_iter};

use crate::bulk::{self, BulkIndexOptions};

/// How large a `write_all` must be, relative to the rows already in a spatial
/// index, before the bulk path rebuilds that index instead of adding its
/// entries to it.
///
/// A write of at least `existing / MERGE_REBUILD_RATIO` new entries takes the
/// rebuild. See [`rebuild_beats_append`] for the measurements behind the value.
const MERGE_REBUILD_RATIO: usize = 10;
use crate::index::drop_all_rtree_triggers;
use crate::value::value_to_sql;
use crate::{Error, Layer, Result, Value};

/// A new row for [`Layer::write_all`]: an optional explicit feature id, an
/// optional geometry, and the non-geometry values in column order.
///
/// Construct with [`NewFeature::new`] (a geometry) or
/// [`NewFeature::attributes`] (none); set an explicit id with
/// [`NewFeature::with_fid`].
#[derive(Debug, Clone)]
pub struct NewFeature<G> {
    /// Explicit feature id, or `None` to let SQLite assign one.
    pub fid: Option<i64>,
    /// The geometry, or `None` for a NULL geometry / attribute row.
    pub geometry: Option<G>,
    /// The non-geometry column values, in the layer's value-column order.
    pub values: Vec<Value>,
}

impl<G> NewFeature<G> {
    /// A feature with a geometry and its values (auto-assigned id).
    pub fn new(geometry: G, values: Vec<Value>) -> Self {
        Self {
            fid: None,
            geometry: Some(geometry),
            values,
        }
    }

    /// A row with no geometry (a NULL geometry, or an attribute row).
    pub fn attributes(values: Vec<Value>) -> Self {
        Self {
            fid: None,
            geometry: None,
            values,
        }
    }

    /// Set an explicit feature id.
    #[must_use]
    pub fn with_fid(mut self, fid: i64) -> Self {
        self.fid = Some(fid);
        self
    }
}

/// A row that [`Layer::write_all`] and its bulk counterpart can write.
///
/// Implemented by [`NewFeature`], whose geometry is an object to be encoded, and
/// by the columnar write path, whose geometry is already ISO WKB and only needs
/// a header. Both paths share the batching, the bulk-index decision and the
/// transaction handling; they differ only in how one row reaches the database,
/// which is what this trait names.
pub(crate) trait WritableRow {
    /// Write this row through `writer`, returning its assigned feature id and
    /// the XY envelope of its geometry, or `None` when it has no indexable
    /// geometry.
    fn write(self, writer: &mut FeatureWriter<'_>) -> Result<(i64, Option<[f64; 4]>)>;
}

impl<G: GeometryTrait<T = f64>> WritableRow for NewFeature<G> {
    fn write(self, writer: &mut FeatureWriter<'_>) -> Result<(i64, Option<[f64; 4]>)> {
        match &self.geometry {
            Some(geometry) => writer.insert_returning_envelope(self.fid, geometry, &self.values),
            None => writer
                .insert_row(self.fid, &self.values)
                .map(|fid| (fid, None)),
        }
    }
}

/// The geometry column a [`FeatureWriter`] targets.
#[derive(Debug)]
struct GeomTarget {
    name: String,
    quoted_name: String,
    srs_id: i32,
    z: ZmFlag,
    m: ZmFlag,
}

/// A running union of written XY envelopes, seeded from the existing
/// `gpkg_contents` bounding box. Bounds are stored as
/// `[min_x, max_x, min_y, max_y]` (the shape [`encode_gpb`] returns).
#[derive(Debug, Clone, Copy)]
struct BboxFold {
    min_x: f64,
    max_x: f64,
    min_y: f64,
    max_y: f64,
    seen: bool,
}

impl BboxFold {
    fn new() -> Self {
        Self {
            min_x: f64::INFINITY,
            max_x: f64::NEG_INFINITY,
            min_y: f64::INFINITY,
            max_y: f64::NEG_INFINITY,
            seen: false,
        }
    }

    fn seed(&mut self, existing: Option<[f64; 4]>) {
        if let Some([min_x, max_x, min_y, max_y]) = existing {
            self.min_x = min_x;
            self.max_x = max_x;
            self.min_y = min_y;
            self.max_y = max_y;
            self.seen = true;
        }
    }

    fn add(&mut self, [min_x, max_x, min_y, max_y]: [f64; 4]) {
        self.min_x = self.min_x.min(min_x);
        self.max_x = self.max_x.max(max_x);
        self.min_y = self.min_y.min(min_y);
        self.max_y = self.max_y.max(max_y);
        self.seen = true;
    }

    fn bounds(&self) -> Option<[f64; 4]> {
        self.seen
            .then_some([self.min_x, self.max_x, self.min_y, self.max_y])
    }
}

/// A prepared-statement writer over one layer, owning a transaction.
///
/// Obtain one with [`Layer::writer`]. Each `insert`/`update`/`delete` stages
/// into the writer's transaction using rusqlite's per-connection statement
/// cache (so repeated calls reuse the compiled statement); [`Self::commit`]
/// flushes catalogue metadata and commits. Dropping a writer without committing
/// rolls its transaction back. The `gpkg_contents` bounding box is grown by a
/// running fold over written geometry envelopes and `last_change` is refreshed
/// on commit.
pub struct FeatureWriter<'conn> {
    tx: Transaction<'conn>,
    table_name: String,
    quoted_table: String,
    /// The primary-key expression: the quoted pk column, or `rowid`.
    pk_expr: String,
    /// Quoted non-geometry column names, in value order.
    value_columns: Vec<String>,
    geometry: Option<GeomTarget>,
    bbox: BboxFold,
    /// The four possible `INSERT` statements, by whether the row carries an
    /// explicit feature id and whether it carries a geometry.
    ///
    /// Built once per writer rather than per row. Composing one costs a `Vec` of
    /// column names, a `String` per placeholder and two joins, which is around
    /// seventeen allocations for a fifteen-column table, and it was happening on
    /// every insert to produce one of four fixed strings.
    insert_sql: [String; 4],
    /// Any insert, or any update/delete that changed a row (drives
    /// `last_change`).
    dirty: bool,
    /// A geometry was written (drives the bounding-box flush).
    bbox_dirty: bool,
}

impl<'a> Layer<'a> {
    /// Begin a write transaction over this layer, returning a [`FeatureWriter`].
    ///
    /// The writer owns the transaction: stage rows with its `insert`/`update`/
    /// `delete` methods, then call [`FeatureWriter::commit`]. Dropping the
    /// writer without committing rolls the transaction back.
    pub fn writer(&self) -> Result<FeatureWriter<'a>> {
        let conn: &Connection = self.gpkg().connection();
        let tx = conn.unchecked_transaction()?;
        let existing = read_contents_bbox(&tx, self.table_name())?;

        let pk_name = self.primary_key_column();
        let pk_expr = match pk_name {
            Some(pk) => quote(pk)?,
            None => "rowid".to_owned(),
        };
        // The read path's value columns include the primary key; the write path
        // treats the primary key as a separate `fid` and the geometry through
        // its own column, so exclude both here.
        let value_columns = self
            .value_columns()
            .iter()
            .filter(|c| Some(c.name.as_str()) != pk_name)
            .map(|c| quote(&c.name))
            .collect::<std::result::Result<Vec<_>, _>>()?;
        let geometry = match self.geometry_column() {
            Some(g) => Some(GeomTarget {
                name: g.column_name.clone(),
                quoted_name: quote(&g.column_name)?,
                srs_id: g.srs_id,
                z: g.z,
                m: g.m,
            }),
            None => None,
        };
        let mut bbox = BboxFold::new();
        bbox.seed(existing);

        let mut writer = FeatureWriter {
            tx,
            table_name: self.table_name().to_owned(),
            quoted_table: quote(self.table_name())?,
            pk_expr,
            value_columns,
            geometry,
            bbox,
            insert_sql: [const { String::new() }; 4],
            dirty: false,
            bbox_dirty: false,
        };
        writer.insert_sql = [
            writer.build_insert_sql(false, false),
            writer.build_insert_sql(true, false),
            writer.build_insert_sql(false, true),
            writer.build_insert_sql(true, true),
        ];
        Ok(writer)
    }

    /// Write every item of `features` in batches, each batch its own committed
    /// transaction.
    ///
    /// `batch_size` bounds how many rows share a transaction (`0` writes them
    /// all in a single transaction). Returns the assigned feature ids in order.
    /// Batches commit independently: an error part-way leaves already-committed
    /// batches in place, so pass `0` when you need all-or-nothing.
    ///
    /// Rows with `geometry: Some(_)` go through [`FeatureWriter::insert`]; rows
    /// with `None` through [`FeatureWriter::insert_row`].
    ///
    /// When the layer carries a spatial index and the write is at least
    /// [`DEFAULT_BULK_THRESHOLD`](bulk::DEFAULT_BULK_THRESHOLD) rows, it takes
    /// the bulk path instead, maintaining the index in one operation at the end
    /// rather than row by row through the triggers;
    /// [`Self::write_all_with`] tunes or forces that choice.
    pub fn write_all<G, I>(&self, features: I, batch_size: usize) -> Result<Vec<i64>>
    where
        G: GeometryTrait<T = f64>,
        I: IntoIterator<Item = NewFeature<G>>,
    {
        self.write_all_with(features, batch_size, BulkIndexOptions::default())
    }

    /// [`Self::write_all`] with an explicit [`BulkIndexOptions`] controlling the
    /// bulk-vs-triggered index-build choice.
    ///
    /// The bulk path is taken when the layer has a spatial index, the write
    /// reaches `options.bulk_threshold` rows, and it is large enough relative to
    /// the rows already indexed to be worth rebuilding the index rather than
    /// appending to it. The size condition is settled from the iterator's
    /// `size_hint` where that is possible, and by buffering up to
    /// `bulk_threshold` rows where it is not, so an iterator that does not know
    /// its own length still reaches the path.
    /// [`BulkIndexOptions::always_bulk`] drops the size condition;
    /// [`BulkIndexOptions::never_bulk`] disables the path.
    pub fn write_all_with<G, I>(
        &self,
        features: I,
        batch_size: usize,
        options: BulkIndexOptions,
    ) -> Result<Vec<i64>>
    where
        G: GeometryTrait<T = f64>,
        I: IntoIterator<Item = NewFeature<G>>,
    {
        self.write_all_impl(features, batch_size, options, bulk::no_fault)
    }

    /// The `write_all_with` core, taking a [`bulk::TestFault`] so that a test
    /// can force the index build to fail after the rows have been staged.
    pub(crate) fn write_all_impl<R, I>(
        &self,
        features: I,
        batch_size: usize,
        options: BulkIndexOptions,
        fault: bulk::TestFault,
    ) -> Result<Vec<i64>>
    where
        R: WritableRow,
        I: IntoIterator<Item = R>,
    {
        let mut iter = features.into_iter();
        let (bulk, buffered) = self.bulk_write_engages(&mut iter, options)?;
        // Rows pulled to reach the decision are put back in front of the rest,
        // so either path sees the same sequence it would have seen.
        let features = buffered.into_iter().chain(iter);
        if bulk {
            self.write_all_bulk(features, options, fault)
        } else {
            self.write_all_batched(features, batch_size)
        }
    }

    /// Whether this write is large enough to take the bulk path, along with any
    /// rows that had to be pulled from `features` to decide.
    ///
    /// This is the one decision that has to be made before a row is written,
    /// because the bulk path drops the RTree triggers first and dropping them
    /// for a handful of rows would cost more in schema churn than it saves.
    /// Whether the index is then rebuilt or appended to is settled after the
    /// write, from the exact counts (see [`rebuild_beats_append`]).
    ///
    /// The size condition is answered from [`Iterator::size_hint`] whenever the
    /// hint settles it, which covers every `Vec`-like source at no cost. An
    /// iterator that does not know its own length, which is most iterators that
    /// are not backed by a collection, reports a lower bound of `0` and would
    /// otherwise never reach the threshold however many rows it went on to
    /// yield. For those the rows themselves are the only evidence available, so
    /// they are buffered until either the threshold is reached, which is all the
    /// proof the decision needs, or the iterator ends first, which gives an exact
    /// count that is known to be below it.
    ///
    /// Buffering is therefore bounded by `options.bulk_threshold` rows and never
    /// by the length of the input. Raising the threshold raises that bound for
    /// unsized iterators.
    fn bulk_write_engages<R, I>(
        &self,
        features: &mut I,
        options: BulkIndexOptions,
    ) -> Result<(bool, Vec<R>)>
    where
        I: Iterator<Item = R>,
    {
        let threshold = options.bulk_threshold;
        // `never_bulk`: the caller has ruled the bulk path out, so there is
        // nothing to decide and nothing to buffer deciding it.
        if threshold == usize::MAX {
            return Ok((false, Vec::new()));
        }
        let (lower, upper) = features.size_hint();
        // An upper bound below the threshold settles it without touching the
        // database, which is the common case for a small write from a `Vec`.
        if upper.is_some_and(|upper| upper < threshold) {
            return Ok((false, Vec::new()));
        }
        // A lower bound that already clears the threshold settles it the other
        // way, again without pulling a row. `has_spatial_index` already requires
        // a geometry column and a single-column primary key, which are the other
        // two things the bulk path needs, so it is the whole availability test.
        if lower >= threshold {
            return Ok((self.has_spatial_index()?, Vec::new()));
        }
        // Neither bound settles it, so buffer up to the threshold. Testing the
        // layer first means an unindexed layer, which can never take the bulk
        // path, does not buffer rows only to discard the decision.
        if !self.has_spatial_index()? {
            return Ok((false, Vec::new()));
        }
        let mut buffered = Vec::new();
        while buffered.len() < threshold {
            let Some(feature) = features.next() else {
                return Ok((false, buffered));
            };
            buffered.push(feature);
        }
        Ok((true, buffered))
    }

    /// The per-batch triggered write path: one committed transaction per
    /// `batch_size` rows (`0` = a single transaction for the whole iterator).
    fn write_all_batched<R, I>(&self, features: I, batch_size: usize) -> Result<Vec<i64>>
    where
        R: WritableRow,
        I: IntoIterator<Item = R>,
    {
        let mut fids = Vec::new();
        let mut iter = features.into_iter();
        let mut batch = self.writer()?;
        let mut in_batch = 0usize;
        let mut wrote_any = false;
        for feature in iter.by_ref() {
            let (fid, _) = feature.write(&mut batch)?;
            fids.push(fid);
            wrote_any = true;
            in_batch += 1;
            if batch_size != 0 && in_batch >= batch_size {
                batch.commit()?;
                batch = self.writer()?;
                in_batch = 0;
                wrote_any = false;
            }
        }
        if wrote_any || in_batch > 0 {
            batch.commit()?;
        } else {
            // Nothing written into the final (fresh) writer: roll it back.
            drop(batch);
        }
        Ok(fids)
    }

    /// The bulk write path: drop the rtree triggers, insert every row in one
    /// transaction (no per-row index maintenance, but `gpkg_contents` bbox and
    /// `last_change` are still maintained by the writer commit), then bring the
    /// index up to date and reinstall the triggers.
    ///
    /// On any failure after the triggers are dropped, the index is restored to a
    /// consistent, trigger-maintained state before the error is returned.
    fn write_all_bulk<R, I>(
        &self,
        features: I,
        options: BulkIndexOptions,
        fault: bulk::TestFault,
    ) -> Result<Vec<i64>>
    where
        R: WritableRow,
        I: IntoIterator<Item = R>,
    {
        let geom = self
            .geometry_column()
            .ok_or_else(|| Error::NoGeometryColumn {
                table_name: self.table_name().to_owned(),
            })?;
        let pk = self
            .primary_key_column()
            .ok_or_else(|| Error::NoPrimaryKey {
                table_name: self.table_name().to_owned(),
            })?;
        let table = self.table_name();
        let column = &geom.column_name;
        let rtree = triggers::rtree_table_name(table, column);

        (|| -> Result<Vec<i64>> {
            let mut fids = Vec::new();
            let mut entries = Vec::new();
            let mut writer = self.writer()?;
            // Inside the writer's transaction, so a failure or a crash rolls the
            // trigger drop back along with everything else.
            drop_all_rtree_triggers(writer.connection(), table, column)?;

            // Both counts describe the table as it was before this write, and
            // both are read here rather than earlier because dropping the
            // triggers is this transaction's first write statement and so the
            // point at which SQLite hands it the write lock. Read before that,
            // there is a window in which another connection commits a row
            // between the count and the lock, and `table_was_empty` in
            // particular cannot survive that: the entry set would then be
            // missing that row, and the gate cannot notice, because it checks
            // the built index against that same set.
            let conn = writer.connection();
            // Envelopes computed while encoding can be reused as the RTree entry
            // set only if this write accounts for every indexable row in the
            // table. An empty table before the write is the cheap, sufficient
            // proof; otherwise `fill_index` re-derives the set with its own
            // `ST_*` scan.
            let table_was_empty = bulk::table_row_count(conn, table)? == 0;
            let indexed = rtree_entry_count(conn, &rtree)?;

            for feature in features {
                let (fid, envelope) = feature.write(&mut writer)?;
                if let Some(envelope) = envelope {
                    entries.push((fid, envelope));
                }
                fids.push(fid);
            }
            // Flush the catalogue metadata but keep the transaction open, so the
            // rows and the index commit together.
            let tx = writer.flush()?;

            // Reinstalling the trigger set is the last thing either branch does,
            // and it happens inside the same transaction as the drop, so a
            // failure anywhere rolls both back together.
            let reinstall = |conn: &Connection| -> Result<()> {
                for sql in triggers::create_triggers_sql(table, column, pk)? {
                    conn.execute_batch(&sql)?;
                }
                Ok(())
            };

            if rebuild_beats_append(entries.len(), indexed) {
                let precomputed = table_was_empty.then_some(entries);
                bulk::fill_index_in_transaction(
                    &tx,
                    table,
                    column,
                    pk,
                    &rtree,
                    options,
                    precomputed,
                    fault,
                    reinstall,
                )?;
            } else {
                append_entries(&tx, &rtree, &entries)?;
                // The rebuild branch hands this to `fill_index`, which calls it
                // at the equivalent point. Calling it here as well is what lets
                // a test fail this branch too, once its index work is done.
                fault(&tx, &rtree)?;
                reinstall(&tx)?;
            }
            tx.commit()?;
            Ok(fids)
        })()
    }
}

/// Whether a bulk `write_all` that produced `new_entries` index entries against
/// an index already holding `indexed` of them should rebuild that index rather
/// than append the new entries to it.
///
/// This is decided after the rows are written rather than before, so both counts
/// are exact. Deciding it up front meant guessing the size of the write from
/// [`Iterator::size_hint`], which can only ever supply a lower bound, and an
/// iterator that supplies none at all could not be placed on either side of the
/// ratio.
///
/// An empty index is the clear case: there is nothing to preserve, so the
/// rebuild is a straight win. A populated index is a trade, because a rebuild
/// pays for the rows already in it. Measured at 1M and 100k existing rows, a
/// rebuild costs roughly 1.5 us per row of the *total* table while an append
/// costs roughly 18 to 40 us per *new* row, rising with table size as the index
/// deepens. Rebuilding therefore wins once the new entries are somewhere between
/// 5% and 10% of the existing ones. [`MERGE_REBUILD_RATIO`] takes the
/// conservative end: at 100k existing rows a 10k-row append measured 187 ms
/// against 149 ms rebuilt, and at 1M existing a 100k-row append measured 2938 ms
/// against 1783 ms.
fn rebuild_beats_append(new_entries: usize, indexed: usize) -> bool {
    if indexed == 0 {
        return true;
    }
    new_entries >= indexed / MERGE_REBUILD_RATIO
}

/// Add one RTree entry per newly written row, leaving the existing index in
/// place.
///
/// This is the work the `_insert` trigger would have done had it still been
/// installed: the same `INSERT OR REPLACE`, the same values, in the same row
/// order, so the index this leaves behind is the one a triggered write would
/// have produced. The envelopes were computed while encoding the geometries, and
/// a row whose geometry is NULL or empty contributed none, which is exactly the
/// trigger's `NEW.geom NOT NULL AND NOT ST_IsEmpty(NEW.geom)` condition.
///
/// Nothing gates the result. The bulk build is gated because it writes a tree by
/// hand into an on-disk format SQLite does not document as an interface; these
/// inserts go through the RTree module itself and need no more checking than the
/// triggers do.
fn append_entries(conn: &Connection, rtree: &str, entries: &[(i64, [f64; 4])]) -> Result<()> {
    if entries.is_empty() {
        return Ok(());
    }
    let sql = format!(
        "INSERT OR REPLACE INTO {} VALUES (?1, ?2, ?3, ?4, ?5)",
        quote(rtree)?
    );
    let mut stmt = conn.prepare_cached(&sql)?;
    for &(fid, [min_x, max_x, min_y, max_y]) in entries {
        stmt.execute(rusqlite::params![fid, min_x, max_x, min_y, max_y])?;
    }
    Ok(())
}

/// The number of entries currently in the RTree `rtree`.
fn rtree_entry_count(conn: &Connection, rtree: &str) -> Result<usize> {
    let count: i64 = conn.query_row(
        &format!("SELECT count(*) FROM {}", quote(rtree)?),
        [],
        |r| r.get(0),
    )?;
    Ok(usize::try_from(count).unwrap_or(usize::MAX))
}

impl<'conn> FeatureWriter<'conn> {
    /// Insert a feature with a geometry, returning its feature id.
    ///
    /// `fid` is `None` to let SQLite assign the id (returned), or `Some(id)` for
    /// an explicit id. `values` must have one entry per non-geometry column, in
    /// the layer's value-column order.
    ///
    /// # Errors
    ///
    /// - [`Error::NoGeometryColumn`] if the layer has no geometry column (use
    ///   [`Self::insert_row`]).
    /// - [`Error::ZmViolation`] if the geometry's `z`/`m` presence breaks the
    ///   column's constraint.
    /// - [`Error::ValueCountMismatch`] if `values` has the wrong length.
    pub fn insert<G: GeometryTrait<T = f64>>(
        &mut self,
        fid: Option<i64>,
        geometry: &G,
        values: &[Value],
    ) -> Result<i64> {
        self.insert_returning_envelope(fid, geometry, values)
            .map(|(assigned, _)| assigned)
    }

    /// [`Self::insert`], additionally returning the geometry's XY envelope
    /// (`[min_x, max_x, min_y, max_y]`, `None` for an empty geometry) so the
    /// bulk write path can accumulate RTree entries without a second `ST_*`
    /// scan of the table.
    pub(crate) fn insert_returning_envelope<G: GeometryTrait<T = f64>>(
        &mut self,
        fid: Option<i64>,
        geometry: &G,
        values: &[Value],
    ) -> Result<(i64, Option<[f64; 4]>)> {
        self.check_values(values)?;
        let (blob, xy) = self.encode_geometry(geometry)?;
        let sql = self.insert_sql(fid.is_some(), true);
        let mut binds: Vec<SqlValue> = Vec::with_capacity(values.len() + 2);
        if let Some(id) = fid {
            binds.push(SqlValue::Integer(id));
        }
        binds.extend(values.iter().map(value_to_sql));
        binds.push(SqlValue::Blob(blob));
        let assigned = self.exec_insert(sql, &binds, fid)?;
        if let Some(envelope) = xy {
            self.bbox.add(envelope);
            self.bbox_dirty = true;
        }
        self.dirty = true;
        Ok((assigned, xy))
    }

    /// Insert a feature whose geometry is already ISO WKB, with its non-geometry
    /// values already prepared as bindings.
    ///
    /// The counterpart of [`Self::insert_returning_envelope`] for the columnar
    /// path, and the reason it takes bindings rather than [`Value`]s: an Arrow
    /// batch already holds every string and blob contiguously, so a binding that
    /// borrows from it costs nothing, where building a `Value` would allocate
    /// per cell and copy. Only `DATE` and `DATETIME` have to be owned, because
    /// they are formatted rather than copied.
    ///
    /// The WKB is parsed once, for the envelope the header and the index both
    /// need, and to reject a body that is not ISO WKB.
    ///
    /// # Errors
    ///
    /// As [`Self::insert_returning_envelope`], plus [`Error::Core`] if the bytes
    /// are not a geometry the `wkb` reader accepts.
    #[cfg(feature = "arrow")]
    pub(crate) fn insert_wkb_bound(
        &mut self,
        fid: Option<i64>,
        wkb: &[u8],
        values: &[rusqlite::types::ToSqlOutput<'_>],
    ) -> Result<(i64, Option<[f64; 4]>)> {
        use rusqlite::types::{ToSqlOutput, Value as SqlV, ValueRef};

        self.check_value_count(values.len())?;
        let geom = self
            .geometry
            .as_ref()
            .ok_or_else(|| Error::NoGeometryColumn {
                table_name: self.table_name.clone(),
            })?;
        let encoded = encode_gpb_from_wkb(wkb, geom.srs_id).map_err(|e| Error::Core(e.into()))?;
        let has_z = matches!(encoded.dimensions, Dimensions::Xyz | Dimensions::Xyzm);
        let has_m = matches!(encoded.dimensions, Dimensions::Xym | Dimensions::Xyzm);
        self.check_zm("z", geom.z, has_z, &geom.name)?;
        self.check_zm("m", geom.m, has_m, &geom.name)?;

        let sql = self.insert_sql(fid.is_some(), true);
        let mut binds: Vec<ToSqlOutput<'_>> = Vec::with_capacity(values.len() + 2);
        if let Some(id) = fid {
            binds.push(ToSqlOutput::Borrowed(ValueRef::Integer(id)));
        }
        binds.extend(values.iter().cloned());
        // The blob was just built, so it moves into the binding rather than
        // being borrowed from something that has to outlive the statement.
        binds.push(ToSqlOutput::Owned(SqlV::Blob(encoded.blob)));
        let assigned = self.exec_insert_bound(sql, &binds, fid)?;
        if let Some(envelope) = encoded.xy_envelope {
            self.bbox.add(envelope);
            self.bbox_dirty = true;
        }
        self.dirty = true;
        Ok((assigned, encoded.xy_envelope))
    }

    /// [`Self::insert_wkb_bound`] for a row with no geometry.
    #[cfg(feature = "arrow")]
    pub(crate) fn insert_row_bound(
        &mut self,
        fid: Option<i64>,
        values: &[rusqlite::types::ToSqlOutput<'_>],
    ) -> Result<i64> {
        use rusqlite::types::{ToSqlOutput, ValueRef};

        self.check_value_count(values.len())?;
        let sql = self.insert_sql(fid.is_some(), false);
        let mut binds: Vec<ToSqlOutput<'_>> = Vec::with_capacity(values.len() + 1);
        if let Some(id) = fid {
            binds.push(ToSqlOutput::Borrowed(ValueRef::Integer(id)));
        }
        binds.extend(values.iter().cloned());
        let assigned = self.exec_insert_bound(sql, &binds, fid)?;
        self.dirty = true;
        Ok(assigned)
    }

    /// Insert a row with no geometry (a NULL geometry on a feature table, or an
    /// attribute row), returning its feature id.
    ///
    /// # Errors
    ///
    /// [`Error::ValueCountMismatch`] if `values` has the wrong length.
    pub fn insert_row(&mut self, fid: Option<i64>, values: &[Value]) -> Result<i64> {
        self.check_values(values)?;
        let sql = self.insert_sql(fid.is_some(), false);
        let mut binds: Vec<SqlValue> = Vec::with_capacity(values.len() + 1);
        if let Some(id) = fid {
            binds.push(SqlValue::Integer(id));
        }
        binds.extend(values.iter().map(value_to_sql));
        let assigned = self.exec_insert(sql, &binds, fid)?;
        self.dirty = true;
        Ok(assigned)
    }

    /// Update the feature `fid`, setting its geometry and values. Returns
    /// whether a row matched.
    ///
    /// # Errors
    ///
    /// As [`Self::insert`].
    pub fn update<G: GeometryTrait<T = f64>>(
        &mut self,
        fid: i64,
        geometry: &G,
        values: &[Value],
    ) -> Result<bool> {
        self.check_values(values)?;
        let (blob, xy) = self.encode_geometry(geometry)?;
        let sql = self.update_sql(true);
        let mut binds: Vec<SqlValue> = Vec::with_capacity(values.len() + 2);
        binds.extend(values.iter().map(value_to_sql));
        binds.push(SqlValue::Blob(blob));
        binds.push(SqlValue::Integer(fid));
        let matched = self.exec_update(&sql, &binds)?;
        if matched {
            if let Some(envelope) = xy {
                self.bbox.add(envelope);
                self.bbox_dirty = true;
            }
            self.dirty = true;
        }
        Ok(matched)
    }

    /// Update the feature `fid`'s non-geometry values, leaving the geometry
    /// untouched. Returns whether a row matched.
    ///
    /// # Errors
    ///
    /// [`Error::ValueCountMismatch`] if `values` has the wrong length.
    pub fn update_row(&mut self, fid: i64, values: &[Value]) -> Result<bool> {
        self.check_values(values)?;
        let sql = self.update_sql(false);
        let mut binds: Vec<SqlValue> = Vec::with_capacity(values.len() + 1);
        binds.extend(values.iter().map(value_to_sql));
        binds.push(SqlValue::Integer(fid));
        let matched = self.exec_update(&sql, &binds)?;
        if matched {
            self.dirty = true;
        }
        Ok(matched)
    }

    /// Delete the feature `fid`. Returns whether a row matched.
    ///
    /// The bounding box is not shrunk (that would need a rescan; an
    /// over-estimate is spec-legal).
    pub fn delete(&mut self, fid: i64) -> Result<bool> {
        let sql = format!(
            "DELETE FROM {} WHERE {} = ?1",
            self.quoted_table, self.pk_expr
        );
        let matched = {
            let mut stmt = self.tx.prepare_cached(&sql)?;
            stmt.execute([fid])? > 0
        };
        if matched {
            self.dirty = true;
        }
        Ok(matched)
    }

    /// Flush `gpkg_contents` (`last_change`, and the bounding box when a
    /// geometry was written) and commit the transaction.
    pub fn commit(self) -> Result<()> {
        self.flush()?.commit()?;
        Ok(())
    }

    /// The connection underlying this writer's transaction, so a caller holding
    /// the writer can run additional statements inside the same transaction.
    pub(crate) fn connection(&self) -> &Connection {
        &self.tx
    }

    /// Flush the `gpkg_contents` metadata and hand back the still-open
    /// transaction, leaving it to the caller to commit.
    ///
    /// The bulk `write_all` path uses this to keep the row inserts and the
    /// index rebuild in one transaction. Dropping the returned transaction
    /// without committing rolls the whole write back, exactly as dropping the
    /// writer would have.
    pub(crate) fn flush(self) -> Result<Transaction<'conn>> {
        let Self {
            tx,
            table_name,
            bbox,
            dirty,
            bbox_dirty,
            ..
        } = self;
        if dirty {
            tx.execute(
                "UPDATE gpkg_contents \
                 SET last_change = strftime('%Y-%m-%dT%H:%M:%fZ','now') \
                 WHERE table_name = ?1",
                [&table_name],
            )?;
        }
        if bbox_dirty && let Some([min_x, max_x, min_y, max_y]) = bbox.bounds() {
            tx.execute(
                "UPDATE gpkg_contents \
                 SET min_x = ?1, min_y = ?2, max_x = ?3, max_y = ?4 \
                 WHERE table_name = ?5",
                rusqlite::params![min_x, min_y, max_x, max_y, table_name],
            )?;
        }
        Ok(tx)
    }

    /// Validate the geometry's `z`/`m` against the column and encode it to a GPB
    /// blob, returning the blob and its XY envelope (for the bbox fold).
    fn encode_geometry<G: GeometryTrait<T = f64>>(
        &self,
        geometry: &G,
    ) -> Result<(Vec<u8>, Option<[f64; 4]>)> {
        let geom = self
            .geometry
            .as_ref()
            .ok_or_else(|| Error::NoGeometryColumn {
                table_name: self.table_name.clone(),
            })?;
        let dim = geometry.dim();
        let has_z = matches!(dim, Dimensions::Xyz | Dimensions::Xyzm);
        let has_m = matches!(dim, Dimensions::Xym | Dimensions::Xyzm);
        self.check_zm("z", geom.z, has_z, &geom.name)?;
        self.check_zm("m", geom.m, has_m, &geom.name)?;
        encode_gpb(geometry, geom.srs_id).map_err(|e| Error::Core(e.into()))
    }

    /// Enforce a `z`/`m` presence constraint for a written geometry.
    fn check_zm(
        &self,
        dimension: &'static str,
        constraint: ZmFlag,
        present: bool,
        column: &str,
    ) -> Result<()> {
        let ok = match constraint {
            ZmFlag::Prohibited => !present,
            ZmFlag::Mandatory => present,
            ZmFlag::Optional => true,
            // `ZmFlag` is `#[non_exhaustive]`; a future constraint we do not
            // understand should not block a write.
            _ => true,
        };
        if ok {
            return Ok(());
        }
        Err(Error::ZmViolation {
            table_name: self.table_name.clone(),
            column: column.to_owned(),
            dimension,
            constraint,
            verb: if present { "carries" } else { "lacks" },
        })
    }

    fn check_values(&self, values: &[Value]) -> Result<()> {
        self.check_value_count(values.len())
    }

    /// [`Self::check_values`] against a count, for a caller whose values are not
    /// a `Value` slice.
    fn check_value_count(&self, found: usize) -> Result<()> {
        if found == self.value_columns.len() {
            return Ok(());
        }
        let _ = found;
        Err(Error::ValueCountMismatch {
            table_name: self.table_name.clone(),
            expected: self.value_columns.len(),
            found,
        })
    }

    /// Build the `INSERT` statement for the given fid/geometry presence.
    /// The cached `INSERT` for this combination of explicit id and geometry.
    fn insert_sql(&self, with_fid: bool, with_geometry: bool) -> &str {
        let index = usize::from(with_fid) | (usize::from(with_geometry) << 1);
        self.insert_sql.get(index).map_or("", String::as_str)
    }

    /// Compose one of the four `INSERT` statements. Called once per writer.
    fn build_insert_sql(&self, with_fid: bool, with_geometry: bool) -> String {
        let mut columns: Vec<&str> = Vec::with_capacity(self.value_columns.len() + 2);
        if with_fid {
            columns.push(&self.pk_expr);
        }
        for column in &self.value_columns {
            columns.push(column);
        }
        if with_geometry && let Some(geom) = &self.geometry {
            columns.push(&geom.quoted_name);
        }
        if columns.is_empty() {
            return format!("INSERT INTO {} DEFAULT VALUES", self.quoted_table);
        }
        let placeholders = (1..=columns.len())
            .map(|i| format!("?{i}"))
            .collect::<Vec<_>>()
            .join(", ");
        format!(
            "INSERT INTO {} ({}) VALUES ({placeholders})",
            self.quoted_table,
            columns.join(", ")
        )
    }

    /// Build the `UPDATE ... WHERE <pk> = ?` statement.
    fn update_sql(&self, with_geometry: bool) -> String {
        let mut assignments: Vec<String> = Vec::with_capacity(self.value_columns.len() + 1);
        let mut index = 1;
        for column in &self.value_columns {
            assignments.push(format!("{column} = ?{index}"));
            index += 1;
        }
        if with_geometry && let Some(geom) = &self.geometry {
            assignments.push(format!("{} = ?{index}", geom.quoted_name));
            index += 1;
        }
        if assignments.is_empty() {
            // Nothing to change (an attribute table with only a primary key):
            // a self-assignment keeps the statement valid and rows-affected
            // meaningful.
            assignments.push(format!("{pk} = {pk}", pk = self.pk_expr));
        }
        format!(
            "UPDATE {} SET {} WHERE {} = ?{index}",
            self.quoted_table,
            assignments.join(", "),
            self.pk_expr
        )
    }

    /// [`Self::exec_insert`] for bindings that may borrow rather than own.
    #[cfg(feature = "arrow")]
    fn exec_insert_bound(
        &self,
        sql: &str,
        binds: &[rusqlite::types::ToSqlOutput<'_>],
        fid: Option<i64>,
    ) -> Result<i64> {
        let mut stmt = self.tx.prepare_cached(sql)?;
        stmt.execute(params_from_iter(binds.iter()))?;
        Ok(fid.unwrap_or_else(|| self.tx.last_insert_rowid()))
    }

    fn exec_insert(&self, sql: &str, binds: &[SqlValue], fid: Option<i64>) -> Result<i64> {
        let mut stmt = self.tx.prepare_cached(sql)?;
        stmt.execute(params_from_iter(binds.iter()))?;
        Ok(fid.unwrap_or_else(|| self.tx.last_insert_rowid()))
    }

    fn exec_update(&self, sql: &str, binds: &[SqlValue]) -> Result<bool> {
        let mut stmt = self.tx.prepare_cached(sql)?;
        Ok(stmt.execute(params_from_iter(binds.iter()))? > 0)
    }
}

/// Read the existing `gpkg_contents` bounding box for `table` as
/// `[min_x, max_x, min_y, max_y]`, or `None` when the row or any bound is
/// absent.
fn read_contents_bbox(conn: &Connection, table: &str) -> Result<Option<[f64; 4]>> {
    let row = conn
        .query_row(
            "SELECT min_x, min_y, max_x, max_y FROM gpkg_contents WHERE table_name = ?1",
            [table],
            |r| {
                Ok((
                    r.get::<_, Option<f64>>(0)?,
                    r.get::<_, Option<f64>>(1)?,
                    r.get::<_, Option<f64>>(2)?,
                    r.get::<_, Option<f64>>(3)?,
                ))
            },
        )
        .optional()?;
    Ok(match row {
        Some((Some(min_x), Some(min_y), Some(max_x), Some(max_y))) => {
            Some([min_x, max_x, min_y, max_y])
        }
        _ => None,
    })
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{GeoPackage, GeometrySpec, TableSchemaBuilder};
    use geo_types::Point;
    use geopackage_core::types::GeometryType;

    /// A [`bulk::TestFault`] that fails the index build outright, standing in
    /// for a crash or an I/O error between staging the rows and rebuilding the
    /// index.
    fn fail_the_build(_: &Connection, _: &str) -> Result<()> {
        Err(Error::NoSpatialIndex {
            table_name: "pts".to_owned(),
            column_name: "geom".to_owned(),
        })
    }

    /// An indexed layer over an empty table, which is the state that makes
    /// `write_all` take the bulk path.
    fn indexed_empty_layer() -> (tempfile::TempDir, GeoPackage) {
        let dir = tempfile::tempdir().unwrap();
        let gpkg = GeoPackage::create(dir.path().join("t.gpkg")).unwrap();
        let layer = gpkg
            .create_layer(
                &TableSchemaBuilder::new("pts")
                    .geometry(GeometrySpec::new(GeometryType::Point, 4326))
                    // Created here, so the test controls when it exists.
                    .spatial_index(false),
            )
            .unwrap();
        layer.create_spatial_index().unwrap();
        (dir, gpkg)
    }

    /// A bulk `write_all` that fails during the index build must leave nothing
    /// behind: not the rows, not a half-built index, not a dropped trigger set.
    ///
    /// This is the atomicity the bulk path used to lack. The rebuild ran in its
    /// own transaction, because building the index in an `ATTACH`ed scratch
    /// database required autocommit, so the rows were already committed by the
    /// time it ran and a failure here left them against a stale index. The
    /// assertions below all fail against that arrangement.
    #[test]
    fn failed_bulk_write_rolls_back_rows_and_index() {
        let (_dir, gpkg) = indexed_empty_layer();
        let layer = gpkg.layer("pts").unwrap();

        let features: Vec<NewFeature<Point<f64>>> = (1..=50)
            .map(|i| {
                let f = f64::from(i);
                NewFeature::new(Point::new(f, -f), Vec::new()).with_fid(i64::from(i))
            })
            .collect();

        let result =
            layer.write_all_impl(features, 0, BulkIndexOptions::always_bulk(), fail_the_build);
        assert!(result.is_err(), "the build should have failed here");

        // No rows: the inserts rolled back with the failed build.
        let rows: i64 = gpkg
            .connection()
            .query_row("SELECT count(*) FROM pts", [], |r| r.get(0))
            .unwrap();
        assert_eq!(rows, 0, "rows survived a failed bulk build");

        // The triggers are dropped inside the same transaction, so the rollback
        // restores them and the index is usable without a repair.
        assert_eq!(
            layer.spatial_index_status().unwrap(),
            crate::SpatialIndexStatus::Current,
            "index left desynchronised by a failed bulk build"
        );

        // And the layer still works: a later write is indexed as normal.
        let mut writer = layer.writer().unwrap();
        writer.insert(Some(1), &Point::new(5.0, 5.0), &[]).unwrap();
        writer.commit().unwrap();
        let indexed: i64 = gpkg
            .connection()
            .query_row("SELECT count(*) FROM rtree_pts_geom", [], |r| r.get(0))
            .unwrap();
        assert_eq!(indexed, 1, "triggers did not survive the rollback");
    }

    /// The same atomicity on the other branch: a bulk `write_all` that adds its
    /// entries to a populated index, and then fails, must also leave nothing
    /// behind.
    ///
    /// `failed_bulk_write_rolls_back_rows_and_index` cannot cover this. It writes
    /// into an empty index, which always rebuilds, so the branch that appends
    /// had no failure test of its own.
    #[test]
    fn failed_append_write_rolls_back_rows_and_index() {
        let (_dir, gpkg) = indexed_empty_layer();
        let layer = gpkg.layer("pts").unwrap();

        // Populate the index, so that a small write into it appends rather than
        // rebuilding.
        {
            let mut writer = layer.writer().unwrap();
            for i in 1..=100 {
                let f = f64::from(i);
                writer
                    .insert(Some(i64::from(i)), &Point::new(f, -f), &[])
                    .unwrap();
            }
            writer.commit().unwrap();
        }

        // 5 new entries against 100 indexed is under the rebuild ratio.
        let features: Vec<NewFeature<Point<f64>>> = (101..=105)
            .map(|i| {
                let f = f64::from(i);
                NewFeature::new(Point::new(f, -f), Vec::new()).with_fid(i64::from(i))
            })
            .collect();
        let result = layer.write_all_impl(
            features,
            0,
            BulkIndexOptions::with_threshold(1),
            fail_the_build,
        );
        assert!(result.is_err(), "the append should have failed here");

        let conn = gpkg.connection();
        let rows: i64 = conn
            .query_row("SELECT count(*) FROM pts", [], |r| r.get(0))
            .unwrap();
        assert_eq!(rows, 100, "rows survived a failed append");
        let indexed: i64 = conn
            .query_row("SELECT count(*) FROM rtree_pts_geom", [], |r| r.get(0))
            .unwrap();
        assert_eq!(indexed, 100, "index entries survived a failed append");
        assert_eq!(
            layer.spatial_index_status().unwrap(),
            crate::SpatialIndexStatus::Current,
            "index left desynchronised by a failed append"
        );
    }
}