use geo_traits::{Dimensions, GeometryTrait};
use geopackage_core::extensions::{Extension, GEOM_TYPE_EXTENSION_DEFINITION};
use geopackage_core::geometry::encode_gpb;
use geopackage_core::geometry::encode_gpb_from_wkb;
use geopackage_core::ident::quote;
use geopackage_core::schema::{ColumnConstraint, ConstraintKind};
use geopackage_core::triggers;
use geopackage_core::types::{GeometryTypeSet, ZmFlag};
use rusqlite::types::{ToSqlOutput, Value as SqlValue, ValueRef};
use rusqlite::{CachedStatement, Connection, Params, params_from_iter};
use crate::bulk::{self, BulkIndexOptions};
use crate::extensions;
use crate::transaction::WriteTransaction;
const MERGE_REBUILD_RATIO: usize = 10;
use crate::index::drop_all_rtree_triggers;
use crate::value::{value_ref_to_bind, value_to_bind};
use crate::{Error, Layer, Result, Value, ValueRef as CellRef};
#[derive(Debug, Clone)]
pub struct NewFeature<G> {
pub fid: Option<i64>,
pub geometry: Option<G>,
pub values: Vec<Value>,
}
impl<G> NewFeature<G> {
pub fn new(geometry: G, values: Vec<Value>) -> Self {
Self {
fid: None,
geometry: Some(geometry),
values,
}
}
pub fn attributes(values: Vec<Value>) -> Self {
Self {
fid: None,
geometry: None,
values,
}
}
#[must_use]
pub fn with_fid(mut self, fid: i64) -> Self {
self.fid = Some(fid);
self
}
}
pub(crate) trait WritableRow {
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_owned(self.fid, &self.values)
.map(|fid| (fid, None)),
}
}
}
#[derive(Debug)]
struct GeomTarget {
name: String,
quoted_name: String,
srs_id: i32,
z: ZmFlag,
m: ZmFlag,
}
#[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])
}
}
pub struct FeatureWriter<'conn> {
tx: WriteTransaction<'conn>,
conn: &'conn Connection,
table_name: String,
quoted_table: String,
pk_expr: String,
value_columns: Vec<ValueColumn>,
geometry: Option<GeomTarget>,
bbox: BboxFold,
insert_stmts: [CachedStatement<'conn>; 4],
update_stmts: [CachedStatement<'conn>; 2],
delete_stmt: CachedStatement<'conn>,
constraints: ColumnConstraints<'conn>,
partial_columns: Vec<String>,
partial_stmt: CachedStatement<'conn>,
dirty: bool,
bbox_dirty: bool,
bbox_covers_layer: bool,
geometry_types: GeometryTypeSet,
}
impl<'a> Layer<'a> {
pub fn writer(&self) -> Result<FeatureWriter<'a>> {
self.gpkg().check_writable(self.table_name())?;
let conn: &Connection = self.gpkg().connection();
let tx = WriteTransaction::begin(conn)?;
let existing = self.stored_extent()?;
let bbox_covers_layer = existing.is_some() || !self.has_rows()?;
let pk_name = self.primary_key_column();
let pk_expr = match pk_name {
Some(pk) => quote(pk)?,
None => "rowid".to_owned(),
};
let value_columns = self
.value_columns()
.iter()
.map(|c| {
Ok(ValueColumn {
quoted: quote(&c.name)?,
name: c.name.clone(),
})
})
.collect::<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.map(|b| [b.min_x, b.max_x, b.min_y, b.max_y]));
let quoted_table = quote(self.table_name())?;
let shape = Shape {
table_name: self.table_name(),
quoted_table: "ed_table,
pk_expr: &pk_expr,
value_columns: &value_columns,
geometry: geometry.as_ref(),
};
let insert_stmts = [
conn.prepare_cached(&build_insert_sql(&shape, false, false))?,
conn.prepare_cached(&build_insert_sql(&shape, true, false))?,
conn.prepare_cached(&build_insert_sql(&shape, false, true))?,
conn.prepare_cached(&build_insert_sql(&shape, true, true))?,
];
let update_stmts = [
conn.prepare_cached(&build_update_sql(&shape, false))?,
conn.prepare_cached(&build_update_sql(&shape, true))?,
];
let delete_stmt =
conn.prepare_cached(&format!("DELETE FROM {quoted_table} WHERE {pk_expr} = ?1"))?;
let partial_stmt = conn.prepare_cached(&build_partial_update_sql(&shape, &[])?)?;
let constraints = ColumnConstraints::read(self, conn, &value_columns)?;
Ok(FeatureWriter {
tx,
conn,
table_name: self.table_name().to_owned(),
quoted_table,
pk_expr,
value_columns,
geometry,
bbox,
insert_stmts,
update_stmts,
delete_stmt,
partial_columns: Vec::new(),
partial_stmt,
dirty: false,
bbox_dirty: false,
bbox_covers_layer,
constraints,
geometry_types: GeometryTypeSet::new(),
})
}
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())
}
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)
}
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>,
{
self.gpkg().check_writable(self.table_name())?;
let mut iter = features.into_iter();
let (bulk, buffered) = self.bulk_write_engages(&mut iter, options)?;
let features = buffered.into_iter().chain(iter);
if bulk {
self.write_all_bulk(features, options, fault)
} else {
self.write_all_batched(features, batch_size)
}
}
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;
if threshold == usize::MAX {
return Ok((false, Vec::new()));
}
let (lower, upper) = features.size_hint();
if upper.is_some_and(|upper| upper < threshold) {
return Ok((false, Vec::new()));
}
if lower >= threshold {
return Ok((self.has_spatial_index()?, Vec::new()));
}
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))
}
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 {
drop(batch);
}
Ok(fids)
}
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()?;
drop_all_rtree_triggers(writer.connection(), table, column)?;
let conn = writer.connection();
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);
}
let tx = writer.flush()?;
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(
conn,
table,
column,
pk,
&rtree,
options,
precomputed,
fault,
reinstall,
)?;
} else {
append_entries(conn, &rtree, &entries)?;
fault(conn, &rtree)?;
reinstall(conn)?;
}
tx.commit()?;
Ok(fids)
})()
}
}
fn rebuild_beats_append(new_entries: usize, indexed: usize) -> bool {
if indexed == 0 {
return true;
}
new_entries >= indexed / MERGE_REBUILD_RATIO
}
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(())
}
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> {
pub fn insert<G: GeometryTrait<T = f64>>(
&mut self,
fid: Option<i64>,
geometry: &G,
values: &[CellRef<'_>],
) -> Result<i64> {
self.check_constraints(values)?;
self.insert_geometry_binds(
fid,
geometry,
values.len(),
values.iter().copied().map(value_ref_to_bind),
)
.map(|(assigned, _)| assigned)
}
pub fn insert_wkb(
&mut self,
fid: Option<i64>,
wkb: &[u8],
values: &[CellRef<'_>],
) -> Result<i64> {
self.check_value_count(values.len())?;
self.check_constraints(values)?;
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 assigned = self.exec_insert(
fid.is_some(),
true,
params_from_iter(
fid.map(|id| ToSqlOutput::Borrowed(ValueRef::Integer(id)))
.into_iter()
.chain(values.iter().copied().map(value_ref_to_bind))
.chain(std::iter::once(ToSqlOutput::Owned(SqlValue::Blob(
encoded.blob,
)))),
),
fid,
)?;
if let Some(envelope) = encoded.xy_envelope {
self.bbox.add(envelope);
self.bbox_dirty = true;
}
self.geometry_types.extend(encoded.extension_types);
self.dirty = true;
Ok(assigned)
}
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_constraints(values)?;
self.insert_geometry_binds(
fid,
geometry,
values.len(),
values.iter().map(value_to_bind),
)
}
fn insert_geometry_binds<'v, G, I>(
&mut self,
fid: Option<i64>,
geometry: &G,
count: usize,
binds: I,
) -> Result<(i64, Option<[f64; 4]>)>
where
G: GeometryTrait<T = f64>,
I: Iterator<Item = ToSqlOutput<'v>>,
{
self.check_value_count(count)?;
let (blob, xy) = self.encode_geometry(geometry)?;
let assigned = self.exec_insert(
fid.is_some(),
true,
params_from_iter(
fid.map(|id| ToSqlOutput::Borrowed(ValueRef::Integer(id)))
.into_iter()
.chain(binds)
.chain(std::iter::once(ToSqlOutput::Owned(SqlValue::Blob(blob)))),
),
fid,
)?;
if let Some(envelope) = xy {
self.bbox.add(envelope);
self.bbox_dirty = true;
}
self.dirty = true;
Ok((assigned, xy))
}
#[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]>)> {
self.check_value_count(values.len())?;
self.check_constraints(values)?;
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 assigned = self.exec_insert(
fid.is_some(),
true,
params_from_iter(
fid.map(|id| ToSqlOutput::Borrowed(ValueRef::Integer(id)))
.into_iter()
.chain(values.iter().map(borrow_bind))
.chain(std::iter::once(ToSqlOutput::Owned(SqlValue::Blob(
encoded.blob,
)))),
),
fid,
)?;
if let Some(envelope) = encoded.xy_envelope {
self.bbox.add(envelope);
self.bbox_dirty = true;
}
self.geometry_types.extend(encoded.extension_types);
self.dirty = true;
Ok((assigned, encoded.xy_envelope))
}
#[cfg(feature = "arrow")]
pub(crate) fn insert_row_bound(
&mut self,
fid: Option<i64>,
values: &[rusqlite::types::ToSqlOutput<'_>],
) -> Result<i64> {
self.check_value_count(values.len())?;
self.check_constraints(values)?;
let assigned = self.exec_insert(
fid.is_some(),
false,
params_from_iter(
fid.map(|id| ToSqlOutput::Borrowed(ValueRef::Integer(id)))
.into_iter()
.chain(values.iter().map(borrow_bind)),
),
fid,
)?;
self.dirty = true;
Ok(assigned)
}
pub fn insert_row(&mut self, fid: Option<i64>, values: &[CellRef<'_>]) -> Result<i64> {
self.check_constraints(values)?;
self.insert_row_binds(
fid,
values.len(),
values.iter().copied().map(value_ref_to_bind),
)
}
pub(crate) fn insert_row_owned(&mut self, fid: Option<i64>, values: &[Value]) -> Result<i64> {
self.check_constraints(values)?;
self.insert_row_binds(fid, values.len(), values.iter().map(value_to_bind))
}
fn insert_row_binds<'v, I>(&mut self, fid: Option<i64>, count: usize, binds: I) -> Result<i64>
where
I: Iterator<Item = ToSqlOutput<'v>>,
{
self.check_value_count(count)?;
let assigned = self.exec_insert(
fid.is_some(),
false,
params_from_iter(
fid.map(|id| ToSqlOutput::Borrowed(ValueRef::Integer(id)))
.into_iter()
.chain(binds),
),
fid,
)?;
self.dirty = true;
Ok(assigned)
}
pub fn update<G: GeometryTrait<T = f64>>(
&mut self,
fid: i64,
geometry: &G,
values: &[CellRef<'_>],
) -> Result<bool> {
self.check_value_count(values.len())?;
self.check_constraints(values)?;
let (blob, xy) = self.encode_geometry(geometry)?;
let matched = self.exec_update(
true,
params_from_iter(values.iter().copied().map(value_ref_to_bind).chain([
ToSqlOutput::Owned(SqlValue::Blob(blob)),
ToSqlOutput::Borrowed(ValueRef::Integer(fid)),
])),
)?;
if matched {
if let Some(envelope) = xy {
self.bbox.add(envelope);
self.bbox_dirty = true;
}
self.dirty = true;
}
Ok(matched)
}
pub fn update_wkb(&mut self, fid: i64, wkb: &[u8], values: &[CellRef<'_>]) -> Result<bool> {
self.check_value_count(values.len())?;
self.check_constraints(values)?;
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 matched = self.exec_update(
true,
params_from_iter(values.iter().copied().map(value_ref_to_bind).chain([
ToSqlOutput::Owned(SqlValue::Blob(encoded.blob)),
ToSqlOutput::Borrowed(ValueRef::Integer(fid)),
])),
)?;
if matched {
if let Some(envelope) = encoded.xy_envelope {
self.bbox.add(envelope);
self.bbox_dirty = true;
}
self.geometry_types.extend(encoded.extension_types);
self.dirty = true;
}
Ok(matched)
}
pub fn update_row(&mut self, fid: i64, values: &[CellRef<'_>]) -> Result<bool> {
self.check_value_count(values.len())?;
self.check_constraints(values)?;
let matched =
self.exec_update(
false,
params_from_iter(values.iter().copied().map(value_ref_to_bind).chain(
std::iter::once(ToSqlOutput::Borrowed(ValueRef::Integer(fid))),
)),
)?;
if matched {
self.dirty = true;
}
Ok(matched)
}
pub fn update_columns(&mut self, fid: i64, columns: &[(&str, CellRef<'_>)]) -> Result<bool> {
self.check_named_constraints(columns)?;
if !self.partial_matches(columns) {
let sql = build_partial_update_sql(&self.shape(), columns)?;
self.partial_stmt = self.conn.prepare_cached(&sql)?;
self.partial_columns.clear();
self.partial_columns
.extend(columns.iter().map(|(name, _)| (*name).to_owned()));
}
let matched = self.partial_stmt.execute(params_from_iter(
columns
.iter()
.map(|(_, value)| value_ref_to_bind(*value))
.chain(std::iter::once(ToSqlOutput::Borrowed(ValueRef::Integer(
fid,
)))),
))? > 0;
if matched {
self.dirty = true;
}
Ok(matched)
}
pub fn update_column(&mut self, fid: i64, column: &str, value: CellRef<'_>) -> Result<bool> {
self.update_columns(fid, &[(column, value)])
}
fn partial_matches(&self, columns: &[(&str, CellRef<'_>)]) -> bool {
self.partial_columns.len() == columns.len()
&& self
.partial_columns
.iter()
.zip(columns)
.all(|(held, (name, _))| held.as_str() == *name)
}
fn shape(&self) -> Shape<'_> {
Shape {
table_name: &self.table_name,
quoted_table: &self.quoted_table,
pk_expr: &self.pk_expr,
value_columns: &self.value_columns,
geometry: self.geometry.as_ref(),
}
}
pub fn delete(&mut self, fid: i64) -> Result<bool> {
let matched = self.delete_stmt.execute([fid])? > 0;
if matched {
self.dirty = true;
}
Ok(matched)
}
pub fn commit(self) -> Result<()> {
self.flush()?.commit()?;
Ok(())
}
pub(crate) fn connection(&self) -> &'conn Connection {
self.conn
}
pub(crate) fn flush(self) -> Result<WriteTransaction<'conn>> {
let Self {
tx,
conn,
table_name,
geometry,
bbox,
dirty,
bbox_dirty,
bbox_covers_layer,
geometry_types,
..
} = self;
if let Some(geom) = geometry.as_ref() {
for ty in geometry_types.iter() {
extensions::register_if_absent(
conn,
Some(&table_name),
Some(&geom.name),
&Extension::GeometryType(ty).name(),
GEOM_TYPE_EXTENSION_DEFINITION,
"read-write",
)?;
}
}
if dirty {
conn.execute(
"UPDATE gpkg_contents \
SET last_change = strftime('%Y-%m-%dT%H:%M:%fZ','now') \
WHERE table_name = ?1",
[&table_name],
)?;
}
if bbox_dirty
&& bbox_covers_layer
&& let Some([min_x, max_x, min_y, max_y]) = bbox.bounds()
{
conn.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)
}
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()))
}
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,
_ => 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_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,
})
}
fn check_constraints<V: AsCheckable>(&mut self, values: &[V]) -> Result<()> {
if self.constraints.is_empty() {
return Ok(());
}
for (index, value) in values.iter().enumerate() {
if !self.constraints.satisfied(index, value.as_checkable())? {
return Err(self.violation(index, value.as_checkable()));
}
}
Ok(())
}
fn check_named_constraints(&mut self, columns: &[(&str, CellRef<'_>)]) -> Result<()> {
if self.constraints.is_empty() {
return Ok(());
}
for (name, value) in columns {
let Some(index) = self
.value_columns
.iter()
.position(|column| column.name == *name)
else {
continue;
};
if !self.constraints.satisfied(index, value.as_checkable())? {
return Err(self.violation(index, value.as_checkable()));
}
}
Ok(())
}
fn violation(&self, index: usize, value: Checkable<'_>) -> Error {
let (constraint_name, constraint) = match self.constraints.at(index) {
Some(constraint) => (constraint.name.clone(), constraint.kind.to_string()),
None => (String::new(), String::new()),
};
Error::ColumnConstraintViolation {
table_name: self.table_name.clone(),
column_name: self
.value_columns
.get(index)
.map_or_else(String::new, |column| column.name.clone()),
constraint_name,
constraint,
value: match value {
Checkable::Null => "NULL".to_owned(),
Checkable::Integer(number) => number.to_string(),
Checkable::Real(number) => number.to_string(),
Checkable::Text(text) => format!("{text:?}"),
Checkable::Unchecked => "(unchecked)".to_owned(),
},
}
}
fn insert_stmt(&mut self, with_fid: bool, with_geometry: bool) -> &mut CachedStatement<'conn> {
let [plain, fid_only, geom_only, both] = &mut self.insert_stmts;
match (with_fid, with_geometry) {
(false, false) => plain,
(true, false) => fid_only,
(false, true) => geom_only,
(true, true) => both,
}
}
fn update_stmt(&mut self, with_geometry: bool) -> &mut CachedStatement<'conn> {
let [plain, with_geom] = &mut self.update_stmts;
if with_geometry { with_geom } else { plain }
}
fn exec_insert<P: Params>(
&mut self,
with_fid: bool,
with_geometry: bool,
binds: P,
fid: Option<i64>,
) -> Result<i64> {
self.insert_stmt(with_fid, with_geometry).execute(binds)?;
Ok(fid.unwrap_or_else(|| self.conn.last_insert_rowid()))
}
fn exec_update<P: Params>(&mut self, with_geometry: bool, binds: P) -> Result<bool> {
Ok(self.update_stmt(with_geometry).execute(binds)? > 0)
}
}
struct Shape<'s> {
table_name: &'s str,
quoted_table: &'s str,
pk_expr: &'s str,
value_columns: &'s [ValueColumn],
geometry: Option<&'s GeomTarget>,
}
struct ValueColumn {
name: String,
quoted: String,
}
struct ColumnConstraints<'conn> {
per_column: Vec<Option<ColumnConstraint>>,
glob: Option<CachedStatement<'conn>>,
}
impl<'conn> ColumnConstraints<'conn> {
fn read(
layer: &Layer<'_>,
conn: &'conn Connection,
value_columns: &[ValueColumn],
) -> Result<Self> {
if !layer.gpkg().enforces_column_constraints() {
return Ok(Self::none());
}
let described = layer.gpkg().data_columns(layer.table_name())?;
let mut per_column = Vec::with_capacity(value_columns.len());
for column in value_columns {
let constraint_name = described
.iter()
.find(|described| described.column_name == column.name)
.and_then(|described| described.constraint_name.as_deref());
per_column.push(match constraint_name {
Some(name) => layer.gpkg().column_constraint(name)?,
None => None,
});
}
let needs_glob = per_column
.iter()
.flatten()
.any(|constraint| matches!(constraint.kind, ConstraintKind::Glob(_)));
let glob = match needs_glob {
true => Some(conn.prepare_cached("SELECT ?1 GLOB ?2")?),
false => None,
};
Ok(Self { per_column, glob })
}
fn none() -> Self {
Self {
per_column: Vec::new(),
glob: None,
}
}
fn is_empty(&self) -> bool {
self.per_column.iter().all(Option::is_none)
}
fn at(&self, index: usize) -> Option<&ColumnConstraint> {
self.per_column.get(index).and_then(Option::as_ref)
}
fn satisfied(&mut self, index: usize, value: Checkable<'_>) -> Result<bool> {
let Self { per_column, glob } = self;
let Some(Some(constraint)) = per_column.get(index) else {
return Ok(true);
};
match (&constraint.kind, value) {
(_, Checkable::Null | Checkable::Unchecked) => Ok(true),
(ConstraintKind::Range { .. }, Checkable::Text(_)) => Ok(false),
(ConstraintKind::Range { .. }, Checkable::Integer(number)) => {
Ok(in_range(&constraint.kind, number as f64))
}
(ConstraintKind::Range { .. }, Checkable::Real(number)) => {
Ok(in_range(&constraint.kind, number))
}
(ConstraintKind::Enum(members), value) => Ok(match value {
Checkable::Text(text) => members.iter().any(|member| member == text),
Checkable::Integer(number) => members.contains(&number.to_string()),
Checkable::Real(number) => members.contains(&number.to_string()),
Checkable::Null | Checkable::Unchecked => true,
}),
(ConstraintKind::Glob(pattern), value) => {
let statement = glob
.as_mut()
.expect("a glob constraint means the statement was prepared");
let matched: i64 = match value {
Checkable::Text(text) => {
statement.query_one(rusqlite::params![text, pattern], |r| r.get(0))?
}
Checkable::Integer(number) => {
statement.query_one(rusqlite::params![number, pattern], |r| r.get(0))?
}
Checkable::Real(number) => {
statement.query_one(rusqlite::params![number, pattern], |r| r.get(0))?
}
Checkable::Null | Checkable::Unchecked => 1,
};
Ok(matched != 0)
}
}
}
}
fn in_range(kind: &ConstraintKind, value: f64) -> bool {
let ConstraintKind::Range {
min,
min_is_inclusive,
max,
max_is_inclusive,
} = kind
else {
return false;
};
let above = if *min_is_inclusive {
value >= *min
} else {
value > *min
};
let below = if *max_is_inclusive {
value <= *max
} else {
value < *max
};
above && below
}
#[derive(Debug, Clone, Copy)]
pub(crate) enum Checkable<'a> {
Null,
Integer(i64),
Real(f64),
Text(&'a str),
Unchecked,
}
pub(crate) trait AsCheckable {
fn as_checkable(&self) -> Checkable<'_>;
}
impl AsCheckable for Value {
fn as_checkable(&self) -> Checkable<'_> {
match self {
Self::Null => Checkable::Null,
Self::Integer(value) => Checkable::Integer(*value),
Self::Boolean(value) => Checkable::Integer(i64::from(*value)),
Self::Float(value) => Checkable::Real(*value),
Self::Text(value) => Checkable::Text(value),
Self::Blob(_) | Self::Date(_) | Self::DateTime(_) => Checkable::Unchecked,
}
}
}
impl AsCheckable for CellRef<'_> {
fn as_checkable(&self) -> Checkable<'_> {
match self {
Self::Null => Checkable::Null,
Self::Integer(value) => Checkable::Integer(*value),
Self::Boolean(value) => Checkable::Integer(i64::from(*value)),
Self::Float(value) => Checkable::Real(*value),
Self::Text(value) => Checkable::Text(value),
Self::Blob(_) | Self::Date(_) | Self::DateTime(_) => Checkable::Unchecked,
}
}
}
impl AsCheckable for ToSqlOutput<'_> {
fn as_checkable(&self) -> Checkable<'_> {
let value = match self {
Self::Borrowed(value) => *value,
Self::Owned(value) => value.into(),
_ => return Checkable::Unchecked,
};
match value {
rusqlite::types::ValueRef::Null => Checkable::Null,
rusqlite::types::ValueRef::Integer(value) => Checkable::Integer(value),
rusqlite::types::ValueRef::Real(value) => Checkable::Real(value),
rusqlite::types::ValueRef::Text(bytes) => match std::str::from_utf8(bytes) {
Ok(text) => Checkable::Text(text),
Err(_) => Checkable::Unchecked,
},
rusqlite::types::ValueRef::Blob(_) => Checkable::Unchecked,
}
}
}
fn build_insert_sql(shape: &Shape<'_>, with_fid: bool, with_geometry: bool) -> String {
let mut columns: Vec<&str> = Vec::with_capacity(shape.value_columns.len() + 2);
if with_fid {
columns.push(shape.pk_expr);
}
for column in shape.value_columns {
columns.push(&column.quoted);
}
if with_geometry && let Some(geom) = shape.geometry {
columns.push(&geom.quoted_name);
}
if columns.is_empty() {
return format!("INSERT INTO {} DEFAULT VALUES", shape.quoted_table);
}
let placeholders = (1..=columns.len())
.map(|i| format!("?{i}"))
.collect::<Vec<_>>()
.join(", ");
format!(
"INSERT INTO {} ({}) VALUES ({placeholders})",
shape.quoted_table,
columns.join(", ")
)
}
fn build_update_sql(shape: &Shape<'_>, with_geometry: bool) -> String {
let mut assignments: Vec<String> = Vec::with_capacity(shape.value_columns.len() + 1);
let mut index = 1;
for column in shape.value_columns {
assignments.push(format!("{} = ?{index}", column.quoted));
index += 1;
}
if with_geometry && let Some(geom) = shape.geometry {
assignments.push(format!("{} = ?{index}", geom.quoted_name));
index += 1;
}
if assignments.is_empty() {
assignments.push(format!("{pk} = {pk}", pk = shape.pk_expr));
}
format!(
"UPDATE {} SET {} WHERE {} = ?{index}",
shape.quoted_table,
assignments.join(", "),
shape.pk_expr
)
}
fn build_partial_update_sql(shape: &Shape<'_>, columns: &[(&str, CellRef<'_>)]) -> Result<String> {
let mut assignments: Vec<String> = Vec::with_capacity(columns.len());
for (position, (name, _)) in columns.iter().enumerate() {
if columns
.iter()
.take(position)
.any(|(earlier, _)| earlier == name)
{
return Err(Error::DuplicateUpdateColumn {
table_name: shape.table_name.to_owned(),
column_name: (*name).to_owned(),
});
}
let column = shape
.value_columns
.iter()
.find(|candidate| candidate.name == *name)
.ok_or_else(|| Error::NoSuchColumn {
table_name: shape.table_name.to_owned(),
column_name: (*name).to_owned(),
})?;
assignments.push(format!("{} = ?{}", column.quoted, position + 1));
}
let fid_placeholder = assignments.len() + 1;
if assignments.is_empty() {
assignments.push(format!("{pk} = {pk}", pk = shape.pk_expr));
}
Ok(format!(
"UPDATE {} SET {} WHERE {} = ?{fid_placeholder}",
shape.quoted_table,
assignments.join(", "),
shape.pk_expr
))
}
#[cfg(feature = "arrow")]
fn borrow_bind<'a>(bind: &'a ToSqlOutput<'_>) -> ToSqlOutput<'a> {
match bind {
ToSqlOutput::Borrowed(value) => ToSqlOutput::Borrowed(*value),
ToSqlOutput::Owned(value) => ToSqlOutput::Borrowed(ValueRef::from(value)),
other => other.clone(),
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::{GeoPackage, GeometrySpec, TableSchemaBuilder};
use geo_types::Point;
use geopackage_core::types::GeometryType;
fn fail_the_build(_: &Connection, _: &str) -> Result<()> {
Err(Error::NoSpatialIndex {
table_name: "pts".to_owned(),
column_name: "geom".to_owned(),
})
}
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))
.spatial_index(false),
)
.unwrap();
layer.create_spatial_index().unwrap();
(dir, gpkg)
}
#[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");
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");
assert_eq!(
layer.spatial_index_status().unwrap(),
crate::SpatialIndexStatus::Current,
"index left desynchronised by a failed bulk build"
);
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");
}
#[test]
fn failed_append_write_rolls_back_rows_and_index() {
let (_dir, gpkg) = indexed_empty_layer();
let layer = gpkg.layer("pts").unwrap();
{
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();
}
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"
);
}
}