use std::collections::HashMap;
use std::fmt::Write as _;
use diesel::sql_types::{BigInt, Nullable, Text};
use diesel_async::RunQueryDsl as _;
use scoped_futures::ScopedFutureExt as _;
use crate::db::{RuntimeConnection, scoped_immediate_transaction};
use crate::{AutumnError, AutumnResult};
#[cfg(not(feature = "sqlite"))]
const PH1: &str = "$1";
#[cfg(not(feature = "sqlite"))]
const PH2: &str = "$2";
#[cfg(not(feature = "sqlite"))]
const PH3: &str = "$3";
#[cfg(feature = "sqlite")]
const PH1: &str = "?";
#[cfg(feature = "sqlite")]
const PH2: &str = "?";
#[cfg(feature = "sqlite")]
const PH3: &str = "?";
#[cfg(not(feature = "sqlite"))]
const FOR_UPDATE: &str = " FOR UPDATE";
#[cfg(feature = "sqlite")]
const FOR_UPDATE: &str = "";
#[cfg(not(feature = "sqlite"))]
const IS_DISTINCT_FROM: &str = "IS DISTINCT FROM";
#[cfg(feature = "sqlite")]
const IS_DISTINCT_FROM: &str = "IS NOT";
#[cfg(not(feature = "sqlite"))]
const IS_NOT_DISTINCT_FROM: &str = "IS NOT DISTINCT FROM";
#[cfg(feature = "sqlite")]
const IS_NOT_DISTINCT_FROM: &str = "IS";
const CHILD_ALIAS: &str = "__autumn_cc_child";
const DELETED_AT: &str = "deleted_at";
#[derive(Debug)]
pub struct CounterCacheSpec<M: 'static> {
pub child_table: &'static str,
pub child_pk: &'static str,
pub child_soft_delete: bool,
pub fk_column: &'static str,
pub parent_table: &'static str,
pub parent_pk: &'static str,
pub counter_column: &'static str,
pub fk_of: fn(&M) -> Option<i64>,
pub pk_of: fn(&M) -> i64,
pub live_of: fn(&M) -> bool,
pub tenant_column: Option<&'static str>,
pub tenant_of: Option<fn(&M) -> Option<String>>,
pub contrib_of: fn(&M) -> i64,
pub contrib_sql: &'static str,
pub filter_sql: &'static str,
pub derivation: Option<&'static crate::derivation::DerivationDef>,
}
impl<M: 'static> Clone for CounterCacheSpec<M> {
fn clone(&self) -> Self {
*self
}
}
impl<M: 'static> Copy for CounterCacheSpec<M> {}
pub trait AutumnCounterCaches: Sized + 'static {
const HAS_COUNTER_CACHES: bool = false;
#[must_use]
fn counter_caches() -> &'static [CounterCacheSpec<Self>] {
&[]
}
}
impl<T: Sized + 'static> AutumnCounterCaches for T {}
#[must_use]
pub fn is_plain_identifier(s: &str) -> bool {
!s.is_empty()
&& !s.starts_with(|c: char| c.is_ascii_digit())
&& s.chars().all(|c| c.is_ascii_alphanumeric() || c == '_')
}
fn assert_spec_idents(view: &SqlView) {
assert!(
is_plain_identifier(view.child_table)
&& is_plain_identifier(view.child_pk)
&& is_plain_identifier(view.fk_column)
&& is_plain_identifier(view.parent_table)
&& is_plain_identifier(view.parent_pk)
&& is_plain_identifier(view.counter_column)
&& view.tenant_column.is_none_or(is_plain_identifier),
"counter-cache spec carries a non-identifier name; it would be spliced \
verbatim into SQL"
);
assert!(
is_lowered_contribution(view.contrib_sql),
"counter-cache spec carries a contribution that is neither `1` nor one \
child column; it would be spliced verbatim into SQL"
);
}
fn is_lowered_contribution(sql: &str) -> bool {
if sql == "1" {
return true;
}
sql.strip_prefix("{c}.\"")
.and_then(|rest| rest.strip_suffix('"'))
.is_some_and(is_plain_identifier)
}
pub fn quote_ident(ident: &str) -> String {
format!("\"{ident}\"")
}
#[derive(Debug, Clone, Copy)]
pub struct SqlView {
pub child_table: &'static str,
pub child_pk: &'static str,
pub child_soft_delete: bool,
pub fk_column: &'static str,
pub parent_table: &'static str,
pub parent_pk: &'static str,
pub counter_column: &'static str,
pub contrib_sql: &'static str,
pub filter_sql: &'static str,
pub tenant_column: Option<&'static str>,
}
impl SqlView {
const fn is_plain(&self) -> bool {
self.contrib_sql.len() == 1
&& self.contrib_sql.as_bytes()[0] == b'1'
&& self.filter_sql.is_empty()
}
const fn counts_rows(&self) -> bool {
self.contrib_sql.len() == 1 && self.contrib_sql.as_bytes()[0] == b'1'
}
}
const fn view<M: 'static>(spec: &CounterCacheSpec<M>) -> SqlView {
SqlView {
child_table: spec.child_table,
child_pk: spec.child_pk,
child_soft_delete: spec.child_soft_delete,
fk_column: spec.fk_column,
parent_table: spec.parent_table,
parent_pk: spec.parent_pk,
counter_column: spec.counter_column,
contrib_sql: spec.contrib_sql,
filter_sql: spec.filter_sql,
tenant_column: spec.tenant_column,
}
}
struct Quoted {
child_table: String,
child_pk: String,
fk_column: String,
parent_table: String,
parent_pk: String,
counter_column: String,
}
fn quoted(view: &SqlView) -> Quoted {
Quoted {
child_table: quote_ident(view.child_table),
child_pk: quote_ident(view.child_pk),
fk_column: quote_ident(view.fk_column),
parent_table: quote_ident(view.parent_table),
parent_pk: quote_ident(view.parent_pk),
counter_column: quote_ident(view.counter_column),
}
}
#[cfg(not(feature = "sqlite"))]
const BINARY_COLLATION: &str = "COLLATE \"C\"";
#[cfg(feature = "sqlite")]
const BINARY_COLLATION: &str = "COLLATE BINARY";
fn with_alias(sql: &str, alias: &str) -> String {
let out = sql.replace("{c}", alias).replace("{bin}", BINARY_COLLATION);
assert!(
!out.contains('{'),
"a lowered fragment carries an unresolved placeholder: {out}"
);
out
}
fn filter_predicate(view: &SqlView, alias: &str) -> String {
with_alias(view.filter_sql, alias)
}
fn contrib_expr(view: &SqlView, alias: &str) -> String {
with_alias(view.contrib_sql, alias)
}
fn aggregate_expr(view: &SqlView, alias: &str) -> String {
if view.counts_rows() {
"COUNT(*)".to_owned()
} else {
sum_expr(&contrib_expr(view, alias))
}
}
#[cfg(not(feature = "sqlite"))]
pub fn sum_expr(contrib: &str) -> String {
format!("COALESCE(SUM({contrib}), 0)")
}
#[cfg(feature = "sqlite")]
pub fn sum_expr(contrib: &str) -> String {
let (hi, lo) = split_sums(contrib);
split_join(&hi, &lo)
}
#[cfg(feature = "sqlite")]
const LOW_MASK: u32 = u32::MAX;
#[cfg(feature = "sqlite")]
fn split_sums(contrib: &str) -> (String, String) {
(
format!("COALESCE(SUM(({contrib}) >> 32), 0)"),
format!("COALESCE(SUM(({contrib}) & {LOW_MASK}), 0)"),
)
}
#[cfg(feature = "sqlite")]
fn split_join(hi: &str, lo: &str) -> String {
let (min, max, radix) = (i32::MIN, i32::MAX, 1_i64 << 32);
let hi = format!("({hi} + (({lo}) >> 32))");
format!(
"CASE WHEN {hi} BETWEEN {min} AND {max} \
THEN {hi} * {radix} + (({lo}) & {LOW_MASK}) \
ELSE abs(-9223372036854775807 - 1) END"
)
}
fn contrib_case_expr(view: &SqlView, alias: &str) -> String {
let contrib = contrib_expr(view, alias);
if view.filter_sql.is_empty() {
return format!("CAST({contrib} AS BIGINT)");
}
let filter = filter_predicate(view, alias);
format!("CAST(CASE WHEN 1 = 1{filter} THEN {contrib} ELSE 0 END AS BIGINT)")
}
fn tenant_predicate_joined(view: &SqlView) -> String {
let Some(tenant_column) = view.tenant_column else {
return String::new();
};
let tenant_column = quote_ident(tenant_column);
let parent_table = quote_ident(view.parent_table);
format!(
" AND {parent_table}.{tenant_column} {IS_NOT_DISTINCT_FROM} \
{CHILD_ALIAS}.{tenant_column}"
)
}
fn tenant_predicate_captured(view: &SqlView) -> String {
let Some(tenant_column) = view.tenant_column else {
return String::new();
};
let tenant_column = quote_ident(tenant_column);
let parent_table = quote_ident(view.parent_table);
format!(" AND CAST({parent_table}.{tenant_column} AS TEXT) {IS_NOT_DISTINCT_FROM} {PH3}")
}
fn tenant_predicate(view: &SqlView, child_id: i64) -> String {
let Some(tenant_column) = view.tenant_column else {
return String::new();
};
let tenant_column = quote_ident(tenant_column);
let Quoted {
child_table,
child_pk,
parent_table,
..
} = quoted(view);
format!(
" AND EXISTS \
(SELECT 1 FROM {child_table} AS {CHILD_ALIAS}_t \
WHERE {CHILD_ALIAS}_t.{child_pk} = {child_id} \
AND {parent_table}.{tenant_column} {IS_NOT_DISTINCT_FROM} \
{CHILD_ALIAS}_t.{tenant_column})"
)
}
fn accumulator(view: &SqlView, counter_column: &str) -> String {
if view.is_plain() {
counter_column.to_owned()
} else {
format!("COALESCE({counter_column}, 0)")
}
}
fn live_predicate(view: &SqlView, want_live: bool) -> String {
if !view.child_soft_delete {
return String::new();
}
let op = if want_live { "IS NULL" } else { "IS NOT NULL" };
let deleted_at = quote_ident(DELETED_AT);
format!(" AND {CHILD_ALIAS}.{deleted_at} {op}")
}
pub async fn counter_cache_serialize_self_referential<M: 'static>(
conn: &mut RuntimeConnection,
specs: &[CounterCacheSpec<M>],
) -> AutumnResult<()> {
for table in self_referential_tables(specs) {
serialize_table(conn, table).await?;
}
Ok(())
}
fn self_referential_tables<M: 'static>(specs: &[CounterCacheSpec<M>]) -> Vec<&'static str> {
let mut tables: Vec<&'static str> = specs
.iter()
.filter(|spec| spec.parent_table == spec.child_table)
.map(|spec| spec.child_table)
.collect();
tables.sort_unstable();
tables.dedup();
tables
}
#[cfg(not(feature = "sqlite"))]
async fn serialize_table(conn: &mut RuntimeConnection, table: &str) -> AutumnResult<()> {
diesel::sql_query(
"SELECT pg_advisory_xact_lock(hashtext('autumn_counter_cache'), hashtext($1))",
)
.bind::<Text, _>(table)
.execute(conn)
.await
.map_err(AutumnError::from)?;
Ok(())
}
#[cfg(feature = "sqlite")]
#[allow(clippy::unused_async, reason = "one call shape for both backends")]
async fn serialize_table(_conn: &mut RuntimeConnection, _table: &str) -> AutumnResult<()> {
Ok(())
}
fn refuse_ledgered_parent(view: &SqlView) -> AutumnResult<()> {
crate::ledger::refuse_out_of_band_write(view.parent_table, "counter cache")
}
pub async fn counter_cache_apply_delta<M: 'static>(
conn: &mut RuntimeConnection,
spec: &CounterCacheSpec<M>,
parent_id: i64,
delta: i64,
scope: TenantScope,
) -> AutumnResult<()> {
let view = view(spec);
assert_spec_idents(&view);
refuse_ledgered_parent(&view)?;
let Quoted {
parent_table,
parent_pk,
counter_column,
..
} = quoted(&view);
let tenant = match &scope {
TenantScope::SameTenantAsChild(child_id) => tenant_predicate(&view, *child_id),
TenantScope::Captured(_) => tenant_predicate_captured(&view),
TenantScope::Unscoped => String::new(),
};
let accumulator = accumulator(&view, &counter_column);
let sql = format!(
"UPDATE {parent_table} SET {counter_column} = {accumulator} + {PH1} \
WHERE {parent_table}.{parent_pk} = {PH2}{tenant}"
);
let query = diesel::sql_query(sql)
.bind::<BigInt, _>(delta)
.bind::<BigInt, _>(parent_id);
if let (TenantScope::Captured(tenant), true) = (scope, view.tenant_column.is_some()) {
query
.bind::<Nullable<Text>, _>(tenant)
.execute(conn)
.await
.map_err(AutumnError::from)?;
return Ok(());
}
query.execute(conn).await.map_err(AutumnError::from)?;
Ok(())
}
#[doc(hidden)]
pub async fn counter_cache_apply_delta_by_child_id<M: 'static>(
conn: &mut RuntimeConnection,
spec: &CounterCacheSpec<M>,
child_id: i64,
delta: i64,
child_state: ChildState,
) -> AutumnResult<()> {
let view = view(spec);
assert_spec_idents(&view);
refuse_ledgered_parent(&view)?;
let state_predicate = match child_state {
ChildState::Any => String::new(),
ChildState::Live => live_predicate(&view, true),
ChildState::SoftDeleted => live_predicate(&view, false),
};
let tenant = tenant_predicate(&view, child_id);
if !view.is_plain() {
debug_assert!(
delta == 1 || delta == -1,
"a derivation delta carries only a sign; the weight comes from the row"
);
let sql =
weighted_delta_by_child_id_sql(&view, child_id, delta < 0, &state_predicate, &tenant);
diesel::sql_query(sql)
.execute(conn)
.await
.map_err(AutumnError::from)?;
return Ok(());
}
let Quoted {
child_table,
child_pk,
fk_column,
parent_table,
parent_pk,
counter_column,
..
} = quoted(&view);
let sql = format!(
"UPDATE {parent_table} SET {counter_column} = {counter_column} + {PH1} \
WHERE {parent_table}.{parent_pk} IN \
(SELECT {CHILD_ALIAS}.{fk_column} FROM {child_table} AS {CHILD_ALIAS} \
WHERE {CHILD_ALIAS}.{child_pk} = {PH2} \
AND {CHILD_ALIAS}.{fk_column} IS NOT NULL{state_predicate}{FOR_UPDATE}){tenant}"
);
diesel::sql_query(sql)
.bind::<BigInt, _>(delta)
.bind::<BigInt, _>(child_id)
.execute(conn)
.await
.map_err(AutumnError::from)?;
Ok(())
}
fn weighted_delta_by_child_id_sql(
view: &SqlView,
child_id: i64,
subtract: bool,
state_predicate: &str,
tenant: &str,
) -> String {
let Quoted {
child_table,
child_pk,
fk_column,
parent_table,
parent_pk,
counter_column,
..
} = quoted(view);
let filter = filter_predicate(view, CHILD_ALIAS);
let contrib = contrib_expr(view, CHILD_ALIAS);
let sign = if subtract { "-" } else { "+" };
let accumulator = accumulator(view, &counter_column);
format!(
"UPDATE {parent_table} SET {counter_column} = {accumulator} {sign} \
COALESCE((SELECT {contrib} FROM {child_table} AS {CHILD_ALIAS} \
WHERE {CHILD_ALIAS}.{child_pk} = {child_id} \
AND {CHILD_ALIAS}.{fk_column} IS NOT NULL{state_predicate}{filter}), 0) \
WHERE {parent_table}.{parent_pk} IN \
(SELECT {CHILD_ALIAS}.{fk_column} FROM {child_table} AS {CHILD_ALIAS} \
WHERE {CHILD_ALIAS}.{child_pk} = {child_id} \
AND {CHILD_ALIAS}.{fk_column} \
IS NOT NULL{state_predicate}{filter}{FOR_UPDATE}){tenant}"
)
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum TenantScope {
SameTenantAsChild(i64),
Captured(Option<String>),
Unscoped,
}
#[doc(hidden)]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ChildState {
Any,
Live,
SoftDeleted,
}
#[doc(hidden)]
pub async fn counter_cache_after_insert<M: Send + Sync + 'static>(
conn: &mut RuntimeConnection,
specs: &[CounterCacheSpec<M>],
record: &M,
) -> AutumnResult<()> {
counter_cache_serialize_self_referential(conn, specs).await?;
let mut contributions: Vec<Contribution> = Vec::with_capacity(specs.len());
for (index, spec) in specs.iter().enumerate() {
if !(spec.live_of)(record) {
continue;
}
let contrib = (spec.contrib_of)(record);
if contrib == 0 {
continue;
}
if let Some(parent_id) = (spec.fk_of)(record) {
let scope = TenantScope::SameTenantAsChild((spec.pk_of)(record));
contributions.push((index, parent_id, contrib, scope));
}
}
apply_ordered(conn, specs, contributions).await
}
#[doc(hidden)]
pub async fn counter_cache_after_insert_many<M: Send + Sync + 'static>(
conn: &mut RuntimeConnection,
specs: &[CounterCacheSpec<M>],
records: &[M],
) -> AutumnResult<()> {
counter_cache_serialize_self_referential(conn, specs).await?;
let mut contributions: Vec<Contribution> = Vec::new();
for (index, spec) in specs.iter().enumerate() {
for record in records {
if !(spec.live_of)(record) {
continue;
}
let contrib = (spec.contrib_of)(record);
if contrib == 0 {
continue;
}
if let Some(parent_id) = (spec.fk_of)(record) {
let scope = TenantScope::SameTenantAsChild((spec.pk_of)(record));
contributions.push((index, parent_id, contrib, scope));
}
}
}
apply_ordered(conn, specs, contributions).await
}
type Contribution = (usize, i64, i64, TenantScope);
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Captured {
pub parent: i64,
pub contrib: i64,
pub tenant: Option<String>,
}
pub type CapturedContribution = Option<Captured>;
pub type CapturedContributions = Vec<(i64, Vec<CapturedContribution>)>;
async fn apply_ordered<M: 'static>(
conn: &mut RuntimeConnection,
specs: &[CounterCacheSpec<M>],
contributions: Vec<Contribution>,
) -> AutumnResult<()> {
if contributions.is_empty() {
return Ok(());
}
for (spec_index, parent_id, delta, scope) in fold_and_order(specs, contributions) {
counter_cache_apply_delta(conn, &specs[spec_index], parent_id, delta, scope).await?;
}
Ok(())
}
fn fold_and_order<M: 'static>(
specs: &[CounterCacheSpec<M>],
contributions: Vec<Contribution>,
) -> Vec<Contribution> {
let mut totals: Vec<(usize, i64, i128, TenantScope)> = Vec::with_capacity(contributions.len());
let mut seen: HashMap<(usize, i64), usize> = HashMap::new();
for (spec_index, parent_id, delta, scope) in contributions {
if specs[spec_index].tenant_column.is_some() {
totals.push((spec_index, parent_id, i128::from(delta), scope));
continue;
}
if let Some(&at) = seen.get(&(spec_index, parent_id)) {
totals[at].2 += i128::from(delta);
} else {
seen.insert((spec_index, parent_id), totals.len());
totals.push((spec_index, parent_id, i128::from(delta), scope));
}
}
let mut folded: Vec<Contribution> = Vec::with_capacity(totals.len());
for (spec_index, parent_id, mut total, scope) in totals {
while total != 0 {
let chunk = total.clamp(i128::from(i64::MIN), i128::from(i64::MAX));
#[allow(clippy::cast_possible_truncation)]
folded.push((spec_index, parent_id, chunk as i64, scope.clone()));
total -= chunk;
}
}
folded
.sort_by_key(|&(spec_index, parent_id, _, _)| (specs[spec_index].parent_table, parent_id));
order_tenant_runs(specs, &mut folded);
folded
}
fn order_tenant_runs<M: 'static>(specs: &[CounterCacheSpec<M>], folded: &mut [Contribution]) {
let mut start = 0;
while start < folded.len() {
let (spec_index, parent_id) = (folded[start].0, folded[start].1);
let mut end = start + 1;
while end < folded.len() && folded[end].0 == spec_index && folded[end].1 == parent_id {
end += 1;
}
if specs[spec_index].tenant_column.is_some() && end - start > 1 {
let run = &mut folded[start..end];
let total: i128 = run.iter().map(|(_, _, delta, _)| i128::from(*delta)).sum();
let mut remaining: Vec<Contribution> = run.to_vec();
let mut prefix: i128 = 0;
for slot in run.iter_mut() {
let toward_total = prefix < total || (prefix == total && total < 0);
let pick = if toward_total {
(0..remaining.len()).rev().max_by_key(|&i| remaining[i].2)
} else {
(0..remaining.len()).min_by_key(|&i| remaining[i].2)
}
.expect("a run has at least one delta left per slot");
let next = remaining.remove(pick);
prefix += i128::from(next.2);
*slot = next;
}
}
start = end;
}
}
fn specs_in_lock_order<M: 'static>(specs: &[CounterCacheSpec<M>]) -> Vec<usize> {
let mut order: Vec<usize> = (0..specs.len()).collect();
order.sort_by_key(|&i| (specs[i].parent_table, specs[i].counter_column));
order
}
pub async fn counter_cache_after_insert_by_id<M: 'static>(
conn: &mut RuntimeConnection,
specs: &[CounterCacheSpec<M>],
child_id: i64,
) -> AutumnResult<()> {
counter_cache_serialize_self_referential(conn, specs).await?;
for index in specs_in_lock_order(specs) {
let spec = &specs[index];
let state = if spec.child_soft_delete {
ChildState::Live
} else {
ChildState::Any
};
counter_cache_apply_delta_by_child_id(conn, spec, child_id, 1, state).await?;
}
Ok(())
}
pub async fn counter_cache_before_delete_by_id<M: 'static>(
conn: &mut RuntimeConnection,
specs: &[CounterCacheSpec<M>],
child_id: i64,
) -> AutumnResult<()> {
counter_cache_serialize_self_referential(conn, specs).await?;
for index in specs_in_lock_order(specs) {
let spec = &specs[index];
let state = if spec.child_soft_delete {
ChildState::Live
} else {
ChildState::Any
};
counter_cache_apply_delta_by_child_id(conn, spec, child_id, -1, state).await?;
}
Ok(())
}
fn child_lock_sql(view: &SqlView, id_list: &str) -> String {
let Quoted {
child_table,
child_pk,
..
} = quoted(view);
format!(
"SELECT {child_pk} AS id FROM {child_table} \
WHERE {child_pk} IN ({id_list}) ORDER BY {child_pk}{FOR_UPDATE}"
)
}
#[doc(hidden)]
pub async fn counter_cache_before_delete_many<M: 'static>(
conn: &mut RuntimeConnection,
specs: &[CounterCacheSpec<M>],
child_ids: &[i64],
) -> AutumnResult<()> {
counter_cache_serialize_self_referential(conn, specs).await?;
if specs.is_empty() || child_ids.is_empty() {
return Ok(());
}
let id_list = id_list(child_ids);
let first = view(&specs[0]);
assert_spec_idents(&first);
diesel::sql_query(child_lock_sql(&first, &id_list))
.load::<IdRow>(conn)
.await
.map_err(AutumnError::from)?;
for index in specs_in_lock_order(specs) {
let view = view(&specs[index]);
assert_spec_idents(&view);
refuse_ledgered_parent(&view)?;
let sql = bulk_decrement_sql(&view, &id_list);
diesel::sql_query(sql)
.execute(conn)
.await
.map_err(AutumnError::from)?;
}
Ok(())
}
#[doc(hidden)]
pub async fn counter_cache_before_detach_many<M: 'static>(
conn: &mut RuntimeConnection,
specs: &[CounterCacheSpec<M>],
fk_column: &str,
child_ids: &[i64],
) -> AutumnResult<()> {
counter_cache_serialize_self_referential(conn, specs).await?;
let detached: Vec<CounterCacheSpec<M>> = specs
.iter()
.filter(|spec| spec.fk_column == fk_column)
.copied()
.collect();
counter_cache_before_delete_many(conn, &detached, child_ids).await
}
fn bulk_decrement_sql(view: &SqlView, id_list: &str) -> String {
let Quoted {
child_table,
child_pk,
fk_column,
parent_table,
parent_pk,
counter_column,
..
} = quoted(view);
let live = if view.child_soft_delete {
live_predicate(view, true)
} else {
String::new()
};
let filter = filter_predicate(view, CHILD_ALIAS);
let aggregate = aggregate_expr(view, CHILD_ALIAS);
let tenant = tenant_predicate_joined(view);
let accumulator = accumulator(view, &counter_column);
let (minuend, aggregate) = bulk_decrement_parts(view, &accumulator, aggregate);
format!(
"UPDATE {parent_table} SET {counter_column} = {minuend}\
(SELECT {aggregate} FROM {child_table} AS {CHILD_ALIAS} \
WHERE {CHILD_ALIAS}.{fk_column} = {parent_table}.{parent_pk} \
AND {CHILD_ALIAS}.{child_pk} IN ({id_list}){live}{filter}{tenant}) \
WHERE {parent_table}.{parent_pk} IN \
(SELECT {CHILD_ALIAS}.{fk_column} FROM {child_table} AS {CHILD_ALIAS} \
WHERE {CHILD_ALIAS}.{child_pk} IN ({id_list}) \
AND {CHILD_ALIAS}.{fk_column} IS NOT NULL{live}{filter} \
AND {CHILD_ALIAS}.{fk_column} = {parent_table}.{parent_pk}{tenant})"
)
}
#[cfg(not(feature = "sqlite"))]
fn bulk_decrement_parts(_view: &SqlView, current: &str, aggregate: String) -> (String, String) {
(format!("{current} - "), aggregate)
}
#[cfg(feature = "sqlite")]
fn bulk_decrement_parts(view: &SqlView, current: &str, aggregate: String) -> (String, String) {
if view.counts_rows() {
return (format!("{current} - "), aggregate);
}
let Quoted {
parent_table,
counter_column,
..
} = quoted(view);
let parent_value = accumulator(view, &format!("{parent_table}.{counter_column}"));
let (hi, lo) = split_sums(&contrib_expr(view, CHILD_ALIAS));
let hi = format!("(({parent_value}) >> 32) - {hi}");
let lo = format!("(({parent_value}) & {LOW_MASK}) - {lo}");
(String::new(), split_join(&hi, &lo))
}
fn id_list(ids: &[i64]) -> String {
let mut out = String::new();
for (i, id) in ids.iter().enumerate() {
if i > 0 {
out.push(',');
}
out.push_str(&id.to_string());
}
out
}
#[doc(hidden)]
pub async fn counter_cache_before_restore_by_id<M: 'static>(
conn: &mut RuntimeConnection,
specs: &[CounterCacheSpec<M>],
child_id: i64,
) -> AutumnResult<()> {
counter_cache_serialize_self_referential(conn, specs).await?;
for index in specs_in_lock_order(specs) {
let spec = &specs[index];
if spec.child_soft_delete {
counter_cache_apply_delta_by_child_id(conn, spec, child_id, 1, ChildState::SoftDeleted)
.await?;
}
}
Ok(())
}
#[doc(hidden)]
pub async fn counter_cache_capture_fks<M: 'static>(
conn: &mut RuntimeConnection,
specs: &[CounterCacheSpec<M>],
child_id: i64,
) -> AutumnResult<Vec<CapturedContribution>> {
counter_cache_serialize_self_referential(conn, specs).await?;
if specs.is_empty() {
return Ok(Vec::new());
}
let mut out = Vec::with_capacity(specs.len());
for spec in specs {
let view = view(spec);
assert_spec_idents(&view);
let Quoted {
child_table,
child_pk,
fk_column,
..
} = quoted(&view);
let live = live_predicate(&view, true);
if let Some(tenant_column) = captured_tenant_column(spec, &view) {
let contrib = contrib_case_expr(&view, CHILD_ALIAS);
let sql = format!(
"SELECT {CHILD_ALIAS}.{fk_column} AS fk_value, \
{contrib} AS contrib_value, \
CAST({CHILD_ALIAS}.{tenant_column} AS TEXT) AS tenant_text \
FROM {child_table} AS {CHILD_ALIAS} \
WHERE {CHILD_ALIAS}.{child_pk} = {PH1}{live}{FOR_UPDATE}"
);
let row: Option<FkContribTenantRow> = diesel::sql_query(sql)
.bind::<BigInt, _>(child_id)
.get_result::<FkContribTenantRow>(conn)
.await
.optional_row()?;
out.push(row.and_then(FkContribTenantRow::captured));
continue;
}
if view.is_plain() {
let sql = format!(
"SELECT {CHILD_ALIAS}.{fk_column} AS fk_value \
FROM {child_table} AS {CHILD_ALIAS} \
WHERE {CHILD_ALIAS}.{child_pk} = {PH1}{live}{FOR_UPDATE}"
);
let row: Option<FkRow> = diesel::sql_query(sql)
.bind::<BigInt, _>(child_id)
.get_result::<FkRow>(conn)
.await
.optional_row()?;
out.push(row.and_then(|r| r.fk_value).map(|fk| captured(fk, 1)));
continue;
}
let contrib = contrib_case_expr(&view, CHILD_ALIAS);
let sql = format!(
"SELECT {CHILD_ALIAS}.{fk_column} AS fk_value, \
{contrib} AS contrib_value \
FROM {child_table} AS {CHILD_ALIAS} \
WHERE {CHILD_ALIAS}.{child_pk} = {PH1}{live}{FOR_UPDATE}"
);
let row: Option<FkContribRow> = diesel::sql_query(sql)
.bind::<BigInt, _>(child_id)
.get_result::<FkContribRow>(conn)
.await
.optional_row()?;
out.push(row.and_then(|r| r.fk_value.map(|fk| captured(fk, r.contrib_value))));
}
Ok(out)
}
const fn captured(parent: i64, contrib: i64) -> Captured {
Captured {
parent,
contrib,
tenant: None,
}
}
fn captured_tenant_column<M: 'static>(
spec: &CounterCacheSpec<M>,
view: &SqlView,
) -> Option<String> {
let tenant_column = view.tenant_column?;
spec.tenant_of?;
Some(quote_ident(tenant_column))
}
#[doc(hidden)]
pub async fn counter_cache_after_update<M: Send + Sync + 'static>(
conn: &mut RuntimeConnection,
specs: &[CounterCacheSpec<M>],
before: &[CapturedContribution],
record: &M,
) -> AutumnResult<()> {
let mut moves: Vec<Contribution> = Vec::with_capacity(specs.len() * 2);
for (index, spec) in specs.iter().enumerate() {
let old = before.get(index).cloned().flatten();
let new = contribution_of(spec, record);
push_diff(index, old, new, (spec.pk_of)(record), &mut moves);
}
apply_ordered(conn, specs, moves).await
}
fn contribution_of<M: 'static>(spec: &CounterCacheSpec<M>, record: &M) -> CapturedContribution {
if !(spec.live_of)(record) {
return None;
}
(spec.fk_of)(record).map(|parent| Captured {
parent,
contrib: (spec.contrib_of)(record),
tenant: spec.tenant_of.and_then(|tenant_of| tenant_of(record)),
})
}
fn push_diff(
index: usize,
old: CapturedContribution,
new: CapturedContribution,
witness: i64,
out: &mut Vec<Contribution>,
) {
if old == new {
return;
}
let live = TenantScope::SameTenantAsChild(witness);
let new_tenant = new.as_ref().and_then(|new| new.tenant.clone());
if let (Some(old), Some(new)) = (&old, &new)
&& old.parent == new.parent
&& old.tenant == new.tenant
{
if let Some(delta) = new.contrib.checked_sub(old.contrib) {
if delta != 0 {
out.push((index, old.parent, delta, live));
}
return;
}
push_removal(index, old.parent, old.contrib, live.clone(), out);
out.push((index, new.parent, new.contrib, live));
return;
}
if let Some(old) = old
&& old.contrib != 0
{
let scope = if old.tenant == new_tenant {
live.clone()
} else {
TenantScope::Captured(old.tenant)
};
push_removal(index, old.parent, old.contrib, scope, out);
}
if let Some(new) = new
&& new.contrib != 0
{
out.push((index, new.parent, new.contrib, live));
}
}
fn push_removal(
index: usize,
parent: i64,
contrib: i64,
scope: TenantScope,
out: &mut Vec<Contribution>,
) {
if let Some(negated) = contrib.checked_neg() {
out.push((index, parent, negated, scope));
} else {
let half = contrib / 2;
out.push((index, parent, -half, scope.clone()));
out.push((index, parent, -(contrib - half), scope));
}
}
#[doc(hidden)]
pub async fn counter_cache_capture_fks_many<M: 'static>(
conn: &mut RuntimeConnection,
specs: &[CounterCacheSpec<M>],
child_ids: &[i64],
) -> AutumnResult<CapturedContributions> {
counter_cache_serialize_self_referential(conn, specs).await?;
if specs.is_empty() || child_ids.is_empty() {
return Ok(Vec::new());
}
let id_list = id_list(child_ids);
let mut by_child: HashMap<i64, Vec<CapturedContribution>> = HashMap::new();
for (index, spec) in specs.iter().enumerate() {
let view = view(spec);
assert_spec_idents(&view);
let Quoted {
child_table,
child_pk,
fk_column,
..
} = quoted(&view);
let live = live_predicate(&view, true);
if let Some(tenant_column) = captured_tenant_column(spec, &view) {
let contrib = contrib_case_expr(&view, CHILD_ALIAS);
let sql = format!(
"SELECT {CHILD_ALIAS}.{child_pk} AS child_id, \
{CHILD_ALIAS}.{fk_column} AS fk_value, \
{contrib} AS contrib_value, \
CAST({CHILD_ALIAS}.{tenant_column} AS TEXT) AS tenant_text \
FROM {child_table} AS {CHILD_ALIAS} \
WHERE {CHILD_ALIAS}.{child_pk} IN ({id_list}){live} \
ORDER BY {CHILD_ALIAS}.{child_pk}{FOR_UPDATE}"
);
let rows: Vec<ChildFkContribTenantRow> = diesel::sql_query(sql)
.load::<ChildFkContribTenantRow>(conn)
.await
.map_err(AutumnError::from)?;
for row in rows {
let entry = by_child
.entry(row.child_id)
.or_insert_with(|| vec![None; specs.len()]);
entry[index] = row.captured();
}
continue;
}
if view.is_plain() {
let sql = format!(
"SELECT {CHILD_ALIAS}.{child_pk} AS child_id, \
{CHILD_ALIAS}.{fk_column} AS fk_value \
FROM {child_table} AS {CHILD_ALIAS} \
WHERE {CHILD_ALIAS}.{child_pk} IN ({id_list}){live} \
ORDER BY {CHILD_ALIAS}.{child_pk}{FOR_UPDATE}"
);
let rows: Vec<ChildFkRow> = diesel::sql_query(sql)
.load::<ChildFkRow>(conn)
.await
.map_err(AutumnError::from)?;
for row in rows {
let entry = by_child
.entry(row.child_id)
.or_insert_with(|| vec![None; specs.len()]);
entry[index] = row.fk_value.map(|fk| captured(fk, 1));
}
continue;
}
let contrib = contrib_case_expr(&view, CHILD_ALIAS);
let sql = format!(
"SELECT {CHILD_ALIAS}.{child_pk} AS child_id, \
{CHILD_ALIAS}.{fk_column} AS fk_value, \
{contrib} AS contrib_value \
FROM {child_table} AS {CHILD_ALIAS} \
WHERE {CHILD_ALIAS}.{child_pk} IN ({id_list}){live} \
ORDER BY {CHILD_ALIAS}.{child_pk}{FOR_UPDATE}"
);
let rows: Vec<ChildFkContribRow> = diesel::sql_query(sql)
.load::<ChildFkContribRow>(conn)
.await
.map_err(AutumnError::from)?;
for row in rows {
let entry = by_child
.entry(row.child_id)
.or_insert_with(|| vec![None; specs.len()]);
entry[index] = row.fk_value.map(|fk| captured(fk, row.contrib_value));
}
}
let mut out: CapturedContributions = by_child.into_iter().collect();
out.sort_unstable_by_key(|(id, _)| *id);
Ok(out)
}
#[doc(hidden)]
pub async fn counter_cache_after_update_many<M: Send + Sync + 'static>(
conn: &mut RuntimeConnection,
specs: &[CounterCacheSpec<M>],
before: &[(i64, Vec<CapturedContribution>)],
records: &[M],
) -> AutumnResult<()> {
if specs.is_empty() {
return Ok(());
}
let pk_of = specs[0].pk_of;
let mut contributions: Vec<Contribution> = Vec::new();
for (index, spec) in specs.iter().enumerate() {
for record in records {
let child_id = pk_of(record);
let Ok(found) = before.binary_search_by_key(&child_id, |(id, _)| *id) else {
continue;
};
let old = before[found].1.get(index).cloned().flatten();
let new = contribution_of(spec, record);
push_diff(index, old, new, child_id, &mut contributions);
}
}
apply_ordered(conn, specs, contributions).await
}
#[doc(hidden)]
pub async fn counter_cache_after_upsert_many<M: Send + Sync + 'static>(
conn: &mut RuntimeConnection,
specs: &[CounterCacheSpec<M>],
existing: &[M],
upserted: &[M],
) -> AutumnResult<()> {
counter_cache_serialize_self_referential(conn, specs).await?;
if specs.is_empty() {
return Ok(());
}
let pk_of = specs[0].pk_of;
let before: HashMap<i64, Vec<CapturedContribution>> = existing
.iter()
.map(|row| {
(
pk_of(row),
specs
.iter()
.map(|spec| contribution_of(spec, row))
.collect(),
)
})
.collect();
let mut contributions: Vec<Contribution> = Vec::new();
for (index, spec) in specs.iter().enumerate() {
for record in upserted {
let child_id = pk_of(record);
let new = contribution_of(spec, record);
match before.get(&child_id) {
None => {
if let Some(new) = new
&& new.contrib != 0
{
let scope = TenantScope::SameTenantAsChild(child_id);
contributions.push((index, new.parent, new.contrib, scope));
}
}
Some(old_fks) => {
let old = old_fks.get(index).cloned().flatten();
push_diff(index, old, new, child_id, &mut contributions);
}
}
}
}
apply_ordered(conn, specs, contributions).await
}
const RECOMPUTE_BATCH: i64 = 1_000;
pub fn ground_truth_sql(view: &SqlView) -> String {
let Quoted {
child_table,
fk_column,
parent_table,
parent_pk,
..
} = quoted(view);
let live = live_predicate(view, true);
let filter = filter_predicate(view, CHILD_ALIAS);
let tenant = tenant_predicate_joined(view);
let aggregate = aggregate_expr(view, CHILD_ALIAS);
format!(
"(SELECT {aggregate} FROM {child_table} AS {CHILD_ALIAS} \
WHERE {CHILD_ALIAS}.{fk_column} = {parent_table}.{parent_pk}{live}{filter}{tenant})"
)
}
pub fn recompute_update_sql(view: &SqlView, ids: &str) -> String {
let Quoted {
parent_table,
parent_pk,
counter_column,
..
} = quoted(view);
let ground_truth = ground_truth_sql(view);
format!(
"UPDATE {parent_table} SET {counter_column} = {ground_truth} \
WHERE {parent_table}.{parent_pk} IN ({ids}) \
AND {parent_table}.{counter_column} {IS_DISTINCT_FROM} {ground_truth}"
)
}
pub fn drift_sql(view: &SqlView, limit: i64) -> String {
let Quoted {
parent_table,
parent_pk,
counter_column,
..
} = quoted(view);
let ground_truth = ground_truth_sql(view);
format!(
"SELECT COUNT(*) AS count FROM \
(SELECT 1 AS drifted FROM \
(SELECT * FROM {parent_table} ORDER BY {parent_table}.{parent_pk} DESC LIMIT {limit}) \
AS {parent_table} \
WHERE {parent_table}.{counter_column} {IS_DISTINCT_FROM} {ground_truth}) \
AS __autumn_cc_drift"
)
}
async fn recompute_batch(
conn: &mut RuntimeConnection,
view: &SqlView,
ids: &[i64],
) -> AutumnResult<usize> {
if ids.is_empty() {
return Ok(0);
}
let view = *view;
let ids = ids.to_vec();
scoped_immediate_transaction::<usize, AutumnError, _>(conn, move |conn| {
async move { recompute_batch_statements(conn, &view, &ids).await }.scope_boxed()
})
.await
}
pub async fn recompute_batch_statements(
conn: &mut RuntimeConnection,
view: &SqlView,
ids: &[i64],
) -> AutumnResult<usize> {
if ids.is_empty() {
return Ok(0);
}
assert_spec_idents(view);
refuse_ledgered_parent(view)?;
let id_list = id_list(ids);
let parent_table = quote_ident(view.parent_table);
let parent_pk = quote_ident(view.parent_pk);
let lock_sql = format!(
"SELECT {parent_pk} AS id FROM {parent_table} \
WHERE {parent_pk} IN ({id_list}) ORDER BY {parent_pk}{FOR_UPDATE}"
);
diesel::sql_query(lock_sql)
.load::<IdRow>(&mut *conn)
.await
.map_err(AutumnError::from)?;
diesel::sql_query(recompute_update_sql(view, &id_list))
.execute(&mut *conn)
.await
.map_err(AutumnError::from)
}
pub async fn parent_id_page(
conn: &mut RuntimeConnection,
view: &SqlView,
cursor: Option<i64>,
limit: i64,
) -> AutumnResult<Vec<i64>> {
assert_spec_idents(view);
let parent_table = quote_ident(view.parent_table);
let parent_pk = quote_ident(view.parent_pk);
let mut page_sql = format!("SELECT {parent_pk} AS id FROM {parent_table}");
if cursor.is_some() {
let _ = write!(page_sql, " WHERE {parent_pk} > {PH1}");
}
let _ = write!(page_sql, " ORDER BY {parent_pk} LIMIT {limit}");
let query = diesel::sql_query(page_sql);
let page = if let Some(after) = cursor {
query.bind::<BigInt, _>(after).load::<IdRow>(conn).await
} else {
query.load::<IdRow>(conn).await
}
.map_err(AutumnError::from)?;
Ok(page.iter().map(|row| row.id).collect())
}
pub async fn recompute_view(
conn: &mut RuntimeConnection,
view: &SqlView,
parent_id: Option<i64>,
) -> AutumnResult<usize> {
assert_spec_idents(view);
if let Some(id) = parent_id {
return recompute_batch(conn, view, &[id]).await;
}
let batch = if view.child_table == view.parent_table {
1
} else {
RECOMPUTE_BATCH
};
let mut touched = 0usize;
let mut cursor: Option<i64> = None;
loop {
let ids = parent_id_page(conn, view, cursor, batch).await?;
let Some(&last) = ids.last() else { break };
cursor = Some(last);
touched += recompute_batch_retrying(conn, view, &ids).await?;
}
Ok(touched)
}
const LOCK_CONTENTION_RETRIES: u32 = 5;
async fn recompute_batch_retrying(
conn: &mut RuntimeConnection,
view: &SqlView,
ids: &[i64],
) -> AutumnResult<usize> {
let mut attempt = 0;
loop {
match recompute_batch(conn, view, ids).await {
Err(error) if attempt < LOCK_CONTENTION_RETRIES && is_lock_contention(&error) => {
attempt += 1;
tracing::warn!(
parent_table = view.parent_table,
column = view.counter_column,
attempt,
error = %error,
"repair batch lost a lock race; retrying"
);
tokio::time::sleep(std::time::Duration::from_millis(25 * u64::from(attempt))).await;
}
outcome => return outcome,
}
}
}
pub fn is_lock_contention(error: &AutumnError) -> bool {
let message = error.to_string();
message.contains("deadlock detected")
|| message.contains("could not serialize access")
|| message.contains("database is locked")
}
#[doc(hidden)]
pub async fn counter_cache_recompute<M: 'static>(
conn: &mut RuntimeConnection,
specs: &[CounterCacheSpec<M>],
parent_id: Option<i64>,
) -> AutumnResult<usize> {
let mut touched = 0usize;
for index in specs_in_lock_order(specs) {
touched += recompute_view(conn, &view(&specs[index]), parent_id).await?;
}
Ok(touched)
}
#[derive(diesel::QueryableByName)]
struct ChildFkRow {
#[diesel(sql_type = BigInt)]
child_id: i64,
#[diesel(sql_type = Nullable<BigInt>)]
fk_value: Option<i64>,
}
#[derive(diesel::QueryableByName)]
struct ChildFkContribRow {
#[diesel(sql_type = BigInt)]
child_id: i64,
#[diesel(sql_type = Nullable<BigInt>)]
fk_value: Option<i64>,
#[diesel(sql_type = BigInt)]
contrib_value: i64,
}
#[derive(diesel::QueryableByName)]
struct IdRow {
#[diesel(sql_type = BigInt)]
id: i64,
}
#[derive(diesel::QueryableByName)]
struct FkRow {
#[diesel(sql_type = Nullable<BigInt>)]
fk_value: Option<i64>,
}
#[derive(diesel::QueryableByName)]
struct FkContribRow {
#[diesel(sql_type = Nullable<BigInt>)]
fk_value: Option<i64>,
#[diesel(sql_type = BigInt)]
contrib_value: i64,
}
#[derive(diesel::QueryableByName)]
struct FkContribTenantRow {
#[diesel(sql_type = Nullable<BigInt>)]
fk_value: Option<i64>,
#[diesel(sql_type = BigInt)]
contrib_value: i64,
#[diesel(sql_type = Nullable<Text>)]
tenant_text: Option<String>,
}
impl FkContribTenantRow {
fn captured(self) -> CapturedContribution {
self.fk_value.map(|parent| Captured {
parent,
contrib: self.contrib_value,
tenant: self.tenant_text,
})
}
}
#[derive(diesel::QueryableByName)]
struct ChildFkContribTenantRow {
#[diesel(sql_type = BigInt)]
child_id: i64,
#[diesel(sql_type = Nullable<BigInt>)]
fk_value: Option<i64>,
#[diesel(sql_type = BigInt)]
contrib_value: i64,
#[diesel(sql_type = Nullable<Text>)]
tenant_text: Option<String>,
}
impl ChildFkContribTenantRow {
fn captured(self) -> CapturedContribution {
self.fk_value.map(|parent| Captured {
parent,
contrib: self.contrib_value,
tenant: self.tenant_text,
})
}
}
trait OptionalRow<T> {
fn optional_row(self) -> AutumnResult<Option<T>>;
}
impl<T> OptionalRow<T> for Result<T, diesel::result::Error> {
fn optional_row(self) -> AutumnResult<Option<T>> {
match self {
Ok(value) => Ok(Some(value)),
Err(diesel::result::Error::NotFound) => Ok(None),
Err(e) => Err(AutumnError::from(e)),
}
}
}
#[cfg(test)]
mod tests {
use super::*;
struct Dummy;
fn spec(soft: bool) -> CounterCacheSpec<Dummy> {
CounterCacheSpec {
child_table: "comments",
child_pk: "id",
child_soft_delete: soft,
fk_column: "post_id",
parent_table: "posts",
parent_pk: "id",
counter_column: "comment_count",
fk_of: |_| Some(1),
pk_of: |_| 1,
live_of: |_| true,
tenant_column: None,
tenant_of: None,
contrib_of: |_| 1,
contrib_sql: "1",
filter_sql: "",
derivation: None,
}
}
fn sum_spec() -> CounterCacheSpec<Dummy> {
CounterCacheSpec {
counter_column: "visible_score",
contrib_sql: "{c}.\"score\"",
filter_sql: " AND ({c}.\"published\" = TRUE)",
..spec(false)
}
}
#[test]
fn plain_identifiers_are_accepted_and_sql_fragments_are_not() {
assert!(is_plain_identifier("comment_count"));
assert!(is_plain_identifier("_x9"));
assert!(!is_plain_identifier(""));
assert!(!is_plain_identifier("9lives"));
assert!(!is_plain_identifier("comment_count; DROP TABLE posts"));
assert!(!is_plain_identifier("comment count"));
assert!(!is_plain_identifier("\"comment_count\""));
}
#[test]
fn the_live_predicate_is_emitted_only_for_a_soft_deleting_child() {
assert_eq!(live_predicate(&view(&spec(false)), true), "");
assert_eq!(
live_predicate(&view(&spec(true)), true),
format!(" AND {CHILD_ALIAS}.\"deleted_at\" IS NULL")
);
assert_eq!(
live_predicate(&view(&spec(true)), false),
format!(" AND {CHILD_ALIAS}.\"deleted_at\" IS NOT NULL")
);
}
#[test]
fn a_bulk_decrement_locks_its_children_in_ascending_id_order() {
let sql = child_lock_sql(&view(&spec(false)), "3,1,2");
assert!(
sql.starts_with("SELECT \"id\" AS id FROM \"comments\""),
"{sql}"
);
assert!(sql.contains("WHERE \"id\" IN (3,1,2)"), "{sql}");
assert!(
sql.contains(&format!("ORDER BY \"id\"{FOR_UPDATE}")),
"the lock must be taken in a deterministic order: {sql}"
);
}
#[test]
fn a_recompute_batch_is_scoped_to_the_ids_it_locked() {
let sql = recompute_update_sql(&view(&spec(false)), "1,2,3");
assert!(sql.contains("\"posts\".\"id\" IN (1,2,3)"), "{sql}");
assert!(
sql.contains(&format!("\"posts\".\"comment_count\" {IS_DISTINCT_FROM}")),
"a healthy parent must still be left unwritten: {sql}"
);
}
fn two_legs() -> Vec<CounterCacheSpec<Dummy>> {
let mut users = spec(false);
users.parent_table = "users";
users.counter_column = "sent_count";
let mut posts = spec(false);
posts.parent_table = "posts";
posts.counter_column = "comment_count";
vec![users, posts]
}
#[test]
fn deltas_go_out_in_a_globally_stable_lock_order() {
let specs = two_legs();
let ordered = fold_and_order(
&specs,
vec![
(0, 5, 1, TenantScope::SameTenantAsChild(100)),
(1, 9, 1, TenantScope::SameTenantAsChild(100)),
(1, 2, 1, TenantScope::SameTenantAsChild(100)),
],
);
let keys: Vec<(&str, i64)> = ordered
.iter()
.map(|(i, parent_id, _, _)| (specs[*i].parent_table, *parent_id))
.collect();
assert_eq!(keys, vec![("posts", 2), ("posts", 9), ("users", 5)]);
assert_eq!(specs_in_lock_order(&specs), vec![1, 0]);
}
#[test]
fn contributions_to_one_parent_fold_into_a_single_statement() {
let specs = two_legs();
let ordered = fold_and_order(
&specs,
vec![
(1, 2, -1, TenantScope::SameTenantAsChild(10)),
(1, 2, 1, TenantScope::SameTenantAsChild(11)),
(1, 7, 1, TenantScope::SameTenantAsChild(12)),
(1, 7, 1, TenantScope::SameTenantAsChild(13)),
(0, 4, -1, TenantScope::SameTenantAsChild(14)),
],
);
assert_eq!(
ordered,
vec![
(1, 7, 2, TenantScope::SameTenantAsChild(12)),
(0, 4, -1, TenantScope::SameTenantAsChild(14))
]
);
}
#[test]
fn a_net_total_no_single_delta_can_carry_goes_out_in_same_sign_pieces() {
let specs = two_legs();
let ordered = fold_and_order(
&specs,
vec![
(1, 2, i64::MAX, TenantScope::SameTenantAsChild(10)),
(1, 2, i64::MAX, TenantScope::SameTenantAsChild(11)),
],
);
assert_eq!(
ordered,
vec![
(1, 2, i64::MAX, TenantScope::SameTenantAsChild(10)),
(1, 2, i64::MAX, TenantScope::SameTenantAsChild(10))
]
);
let ordered = fold_and_order(
&specs,
vec![
(1, 2, i64::MAX, TenantScope::SameTenantAsChild(10)),
(1, 2, -5, TenantScope::SameTenantAsChild(11)),
(1, 2, 5, TenantScope::SameTenantAsChild(12)),
(1, 2, 1, TenantScope::SameTenantAsChild(13)),
],
);
assert_eq!(
ordered,
vec![
(1, 2, i64::MAX, TenantScope::SameTenantAsChild(10)),
(1, 2, 1, TenantScope::SameTenantAsChild(10))
]
);
}
#[test]
fn weights_that_cancel_across_an_overflowing_intermediate_issue_nothing() {
let specs = two_legs();
let ordered = fold_and_order(
&specs,
vec![
(1, 2, i64::MAX, TenantScope::SameTenantAsChild(10)),
(1, 2, i64::MAX, TenantScope::SameTenantAsChild(11)),
(1, 2, -i64::MAX, TenantScope::SameTenantAsChild(12)),
(1, 2, -i64::MAX, TenantScope::SameTenantAsChild(13)),
],
);
assert!(ordered.is_empty(), "{ordered:?}");
let ordered = fold_and_order(
&specs,
vec![
(1, 2, i64::MAX, TenantScope::SameTenantAsChild(10)),
(1, 2, i64::MAX, TenantScope::SameTenantAsChild(11)),
(1, 2, -i64::MAX, TenantScope::SameTenantAsChild(12)),
(1, 2, -7, TenantScope::SameTenantAsChild(13)),
],
);
assert_eq!(
ordered,
vec![(1, 2, i64::MAX - 7, TenantScope::SameTenantAsChild(10))]
);
}
#[test]
fn a_weight_edit_whose_difference_does_not_fit_goes_out_as_two_deltas() {
let mut out = Vec::new();
push_diff(
0,
Some(captured(7, i64::MIN)),
Some(captured(7, i64::MAX)),
1,
&mut out,
);
assert_eq!(
out,
vec![
(0, 7, i64::MAX / 2 + 1, TenantScope::SameTenantAsChild(1)),
(0, 7, i64::MAX / 2 + 1, TenantScope::SameTenantAsChild(1)),
(0, 7, i64::MAX, TenantScope::SameTenantAsChild(1))
]
);
let specs = two_legs();
assert_eq!(
fold_and_order(&specs, out.clone()),
vec![
(0, 7, i64::MAX, TenantScope::SameTenantAsChild(1)),
(0, 7, i64::MAX, TenantScope::SameTenantAsChild(1)),
(0, 7, 1, TenantScope::SameTenantAsChild(1))
]
);
let mut out = Vec::new();
push_diff(0, Some(captured(7, 3)), Some(captured(7, 10)), 1, &mut out);
assert_eq!(out, vec![(0, 7, 7, TenantScope::SameTenantAsChild(1))]);
}
#[test]
fn removing_the_one_unnegatable_weight_splits_it_in_two() {
let mut out = Vec::new();
push_diff(0, Some(captured(7, i64::MIN)), None, 1, &mut out);
assert_eq!(
out,
vec![
(0, 7, i64::MAX / 2 + 1, TenantScope::SameTenantAsChild(1)),
(0, 7, i64::MAX / 2 + 1, TenantScope::SameTenantAsChild(1))
]
);
assert_eq!(
out.iter()
.map(|(_, _, delta, _)| i128::from(*delta))
.sum::<i128>(),
-i128::from(i64::MIN)
);
let mut out = Vec::new();
push_diff(
0,
Some(captured(7, i64::MIN)),
Some(captured(8, 4)),
1,
&mut out,
);
assert_eq!(
out,
vec![
(0, 7, i64::MAX / 2 + 1, TenantScope::SameTenantAsChild(1)),
(0, 7, i64::MAX / 2 + 1, TenantScope::SameTenantAsChild(1)),
(0, 8, 4, TenantScope::SameTenantAsChild(1))
]
);
}
#[test]
fn a_tenant_scoped_run_is_ordered_so_no_prefix_leaves_the_span_of_zero_and_total() {
let mut specs = two_legs();
specs[1].tenant_column = Some("tenant_id");
let ordered = fold_and_order(
&specs,
vec![
(1, 2, i64::MAX, TenantScope::SameTenantAsChild(10)),
(1, 2, i64::MAX, TenantScope::SameTenantAsChild(11)),
(1, 2, -i64::MAX, TenantScope::SameTenantAsChild(12)),
],
);
let deltas: Vec<i64> = ordered.iter().map(|(_, _, d, _)| *d).collect();
assert_eq!(deltas, vec![i64::MAX, -i64::MAX, i64::MAX]);
let ordered = fold_and_order(
&specs,
vec![
(1, 2, -i64::MAX, TenantScope::SameTenantAsChild(10)),
(1, 2, -i64::MAX, TenantScope::SameTenantAsChild(11)),
(1, 2, i64::MAX, TenantScope::SameTenantAsChild(12)),
],
);
let deltas: Vec<i64> = ordered.iter().map(|(_, _, d, _)| *d).collect();
assert_eq!(deltas, vec![-i64::MAX, i64::MAX, -i64::MAX]);
let witnesses: Vec<TenantScope> = ordered.iter().map(|(_, _, _, w)| w.clone()).collect();
assert_eq!(
witnesses,
vec![
TenantScope::SameTenantAsChild(10),
TenantScope::SameTenantAsChild(12),
TenantScope::SameTenantAsChild(11)
]
);
}
fn captured_in(parent: i64, contrib: i64, tenant: Option<&str>) -> Captured {
Captured {
parent,
contrib,
tenant: tenant.map(str::to_owned),
}
}
#[test]
fn a_tenant_change_takes_the_old_weight_off_under_the_old_tenant() {
let live = TenantScope::SameTenantAsChild(1);
let was_a = TenantScope::Captured(Some("a".to_owned()));
let mut out = Vec::new();
push_diff(
0,
Some(captured_in(7, 3, Some("a"))),
Some(captured_in(7, 3, Some("b"))),
1,
&mut out,
);
assert_eq!(
out,
vec![(0, 7, -3, was_a.clone()), (0, 7, 3, live.clone())]
);
let mut out = Vec::new();
push_diff(
0,
Some(captured_in(7, 3, Some("a"))),
Some(captured_in(8, 3, Some("b"))),
1,
&mut out,
);
assert_eq!(
out,
vec![(0, 7, -3, was_a.clone()), (0, 8, 3, live.clone())]
);
let mut out = Vec::new();
push_diff(
0,
Some(captured_in(7, 1, None)),
Some(captured_in(7, 1, Some("b"))),
1,
&mut out,
);
assert_eq!(
out,
vec![
(0, 7, -1, TenantScope::Captured(None)),
(0, 7, 1, live.clone())
]
);
let mut out = Vec::new();
push_diff(
0,
Some(captured_in(7, 3, Some("a"))),
Some(captured_in(7, 10, Some("a"))),
1,
&mut out,
);
assert_eq!(out, vec![(0, 7, 7, live)]);
let mut out = Vec::new();
push_diff(0, Some(captured_in(7, 3, Some("a"))), None, 1, &mut out);
assert_eq!(out, vec![(0, 7, -3, was_a)]);
}
#[test]
fn only_a_table_with_a_leg_onto_itself_is_serialized_and_only_once() {
assert!(self_referential_tables(&two_legs()).is_empty());
let mut replies = spec(false);
replies.parent_table = "comments";
replies.counter_column = "reply_count";
let mut likes = spec(false);
likes.parent_table = "comments";
likes.counter_column = "like_count";
let mut users = spec(false);
users.child_table = "avatars";
users.parent_table = "avatars";
assert_eq!(
self_referential_tables(&[users, replies, spec(false), likes]),
vec!["avatars", "comments"]
);
}
#[test]
fn a_captured_tenant_scopes_the_parent_by_its_text() {
let mut tenanted = spec(false);
tenanted.tenant_column = Some("tenant_id");
assert_eq!(
tenant_predicate_captured(&view(&tenanted)),
format!(" AND CAST(\"posts\".\"tenant_id\" AS TEXT) {IS_NOT_DISTINCT_FROM} {PH3}")
);
assert_eq!(tenant_predicate_captured(&view(&spec(false))), "");
}
#[test]
fn a_tenant_scoped_leg_keeps_one_statement_per_child() {
let mut specs = two_legs();
specs[1].tenant_column = Some("tenant_id");
let ordered = fold_and_order(
&specs,
vec![
(1, 3, 1, TenantScope::SameTenantAsChild(20)),
(1, 3, 1, TenantScope::SameTenantAsChild(21)),
],
);
assert_eq!(
ordered,
vec![
(1, 3, 1, TenantScope::SameTenantAsChild(20)),
(1, 3, 1, TenantScope::SameTenantAsChild(21))
]
);
}
#[test]
fn a_tenant_discriminator_is_matched_null_safely() {
let mut tenanted = spec(false);
tenanted.tenant_column = Some("tenant_id");
let joined = tenant_predicate_joined(&view(&tenanted));
assert!(joined.contains(IS_NOT_DISTINCT_FROM), "{joined}");
assert!(
!joined.contains("\"tenant_id\" = "),
"plain equality is not NULL-safe: {joined}"
);
let keyed = tenant_predicate(&view(&tenanted), 7);
assert!(keyed.contains("EXISTS"), "{keyed}");
assert!(keyed.contains(IS_NOT_DISTINCT_FROM), "{keyed}");
assert!(keyed.contains("\"id\" = 7"), "{keyed}");
assert_eq!(tenant_predicate_joined(&view(&spec(false))), "");
assert_eq!(tenant_predicate(&view(&spec(false)), 7), "");
}
#[test]
fn a_counter_column_named_after_a_sql_keyword_still_produces_valid_sql() {
let mut keyword = spec(false);
keyword.counter_column = "order";
keyword.parent_table = "group";
let sql = recompute_update_sql(&view(&keyword), "1");
assert!(
sql.starts_with("UPDATE \"group\" SET \"order\" = "),
"{sql}"
);
assert!(sql.contains("\"group\".\"order\" "), "{sql}");
assert!(
!sql.contains(" order "),
"no bare keyword may survive: {sql}"
);
}
#[test]
fn a_plain_counter_cache_keeps_its_pre_derivation_sql() {
let plain = view(&spec(false));
assert!(plain.is_plain());
assert!(plain.counts_rows());
assert_eq!(filter_predicate(&plain, CHILD_ALIAS), "");
assert_eq!(aggregate_expr(&plain, CHILD_ALIAS), "COUNT(*)");
assert!(
!recompute_update_sql(&plain, "1").contains("COALESCE"),
"a plain counter cache must not grow a COALESCE"
);
assert_eq!(
accumulator(&plain, "\"comment_count\""),
"\"comment_count\"",
"a plain counter cache reads its column bare"
);
assert!(
!bulk_decrement_sql(&plain, "7").contains("COALESCE"),
"a plain bulk decrement must not grow a COALESCE"
);
}
#[test]
fn a_derivation_delta_never_writes_null_into_the_maintained_column() {
let mut filtered = spec(false);
filtered.filter_sql = " AND ({c}.\"published\" = TRUE)";
let filtered = view(&filtered);
assert_eq!(
accumulator(&filtered, "\"comment_count\""),
"COALESCE(\"comment_count\", 0)"
);
assert!(
bulk_decrement_sql(&filtered, "7")
.contains("SET \"comment_count\" = COALESCE(\"comment_count\", 0) -"),
"{}",
bulk_decrement_sql(&filtered, "7")
);
let sum = view(&sum_spec());
assert!(
weighted_delta_by_child_id_sql(&sum, 1, false, "", "")
.contains("SET \"visible_score\" = COALESCE(\"visible_score\", 0) +"),
"the by-id weighted delta coalesces too"
);
}
#[test]
fn a_lowered_contribution_is_either_one_or_a_single_child_column() {
assert!(is_lowered_contribution("1"));
assert!(is_lowered_contribution("{c}.\"score\""));
assert!(!is_lowered_contribution("{c}.\"score\" + 1"));
assert!(!is_lowered_contribution("(SELECT 1)"));
assert!(!is_lowered_contribution(""));
assert!(!is_lowered_contribution("{c}.\"a\"; DROP TABLE posts --\""));
}
#[test]
fn a_filter_may_carry_a_comment_marker_inside_a_quoted_literal() {
let mut filtered = spec(false);
filtered.filter_sql = " AND ({c}.\"status\" = 'a -- b; c')";
let filtered = view(&filtered);
assert_spec_idents(&filtered);
assert!(
bulk_decrement_sql(&filtered, "7").contains("'a -- b; c'"),
"the literal reaches the statement unaltered"
);
}
#[test]
fn a_filtered_count_carries_its_filter_into_every_statement() {
let mut filtered = spec(false);
filtered.filter_sql = " AND ({c}.\"published\" = TRUE)";
let filtered = view(&filtered);
assert!(
!filtered.is_plain(),
"a filter is not a plain counter cache"
);
assert!(filtered.counts_rows(), "a filtered count still counts rows");
let delete = bulk_decrement_sql(&filtered, "7,8");
assert_eq!(
delete,
"UPDATE \"posts\" SET \"comment_count\" = \
COALESCE(\"comment_count\", 0) - \
(SELECT COUNT(*) FROM \"comments\" AS __autumn_cc_child \
WHERE __autumn_cc_child.\"post_id\" = \"posts\".\"id\" \
AND __autumn_cc_child.\"id\" IN (7,8) \
AND (__autumn_cc_child.\"published\" = TRUE)) \
WHERE \"posts\".\"id\" IN \
(SELECT __autumn_cc_child.\"post_id\" FROM \"comments\" AS __autumn_cc_child \
WHERE __autumn_cc_child.\"id\" IN (7,8) \
AND __autumn_cc_child.\"post_id\" IS NOT NULL \
AND (__autumn_cc_child.\"published\" = TRUE) \
AND __autumn_cc_child.\"post_id\" = \"posts\".\"id\")"
);
assert_eq!(
contrib_case_expr(&filtered, CHILD_ALIAS),
"CAST(CASE WHEN 1 = 1 AND (__autumn_cc_child.\"published\" = TRUE) \
THEN 1 ELSE 0 END AS BIGINT)"
);
}
#[cfg(not(feature = "sqlite"))]
#[test]
fn the_postgres_sum_is_a_plain_coalesced_sum() {
assert_eq!(sum_expr("x"), "COALESCE(SUM(x), 0)");
}
#[test]
fn a_sum_recompute_assigns_the_summed_contribution() {
let sum = view(&sum_spec());
assert!(!sum.counts_rows());
let score_sum = sum_expr("__autumn_cc_child.\"score\"");
assert_eq!(
recompute_update_sql(&sum, "1,2"),
format!(
"UPDATE \"posts\" SET \"visible_score\" = \
(SELECT {score_sum} \
FROM \"comments\" AS __autumn_cc_child \
WHERE __autumn_cc_child.\"post_id\" = \"posts\".\"id\" \
AND (__autumn_cc_child.\"published\" = TRUE)) \
WHERE \"posts\".\"id\" IN (1,2) \
AND \"posts\".\"visible_score\" {IS_DISTINCT_FROM} \
(SELECT {score_sum} \
FROM \"comments\" AS __autumn_cc_child \
WHERE __autumn_cc_child.\"post_id\" = \"posts\".\"id\" \
AND (__autumn_cc_child.\"published\" = TRUE))"
)
);
}
#[test]
fn a_weighted_by_id_delta_reads_the_weight_from_the_row() {
let sql = weighted_delta_by_child_id_sql(&view(&sum_spec()), 9, false, "", "");
assert!(
sql.starts_with(
"UPDATE \"posts\" SET \"visible_score\" = \
COALESCE(\"visible_score\", 0) + \
COALESCE((SELECT __autumn_cc_child.\"score\""
),
"{sql}"
);
assert_eq!(
sql.matches("__autumn_cc_child.\"id\" = 9").count(),
2,
"{sql}"
);
assert!(
sql.contains(&format!(
"IS NOT NULL AND (__autumn_cc_child.\"published\" = TRUE){FOR_UPDATE})"
)),
"{sql}"
);
let down = weighted_delta_by_child_id_sql(&view(&sum_spec()), 9, true, "", "");
assert!(
down.contains("\"visible_score\" = COALESCE(\"visible_score\", 0) - COALESCE("),
"{down}"
);
}
#[test]
fn a_filter_flip_on_the_same_parent_is_one_delta() {
let mut out = Vec::new();
push_diff(0, Some(captured(4, 0)), Some(captured(4, 1)), 11, &mut out);
assert_eq!(out, vec![(0, 4, 1, TenantScope::SameTenantAsChild(11))]);
out.clear();
push_diff(0, Some(captured(4, 0)), Some(captured(5, 0)), 11, &mut out);
assert!(out.is_empty(), "{out:?}");
out.clear();
push_diff(0, Some(captured(4, 3)), Some(captured(5, 3)), 11, &mut out);
assert_eq!(
out,
vec![
(0, 4, -3, TenantScope::SameTenantAsChild(11)),
(0, 5, 3, TenantScope::SameTenantAsChild(11))
]
);
out.clear();
push_diff(0, Some(captured(4, 3)), Some(captured(4, 10)), 11, &mut out);
assert_eq!(out, vec![(0, 4, 7, TenantScope::SameTenantAsChild(11))]);
}
#[test]
fn drift_is_one_aggregate_over_the_parent_table() {
let sql = drift_sql(&view(&sum_spec()), 10_000);
assert!(
sql.starts_with("SELECT COUNT(*) AS count FROM (SELECT 1 AS drifted FROM "),
"{sql}"
);
assert!(sql.contains(IS_DISTINCT_FROM), "{sql}");
assert!(
sql.contains(
"FROM (SELECT * FROM \"posts\" ORDER BY \"posts\".\"id\" DESC LIMIT 10000) AS \"posts\" WHERE"
),
"the window is materialised under the table's own name: {sql}"
);
assert!(sql.ends_with(") AS __autumn_cc_drift"), "{sql}");
}
#[test]
fn a_model_without_an_inherent_shadow_resolves_to_the_empty_blanket() {
const { assert!(!Dummy::HAS_COUNTER_CACHES) };
assert!(Dummy::counter_caches().is_empty());
}
#[cfg(feature = "sqlite")]
mod sqlite_sum {
use super::{CHILD_ALIAS, aggregate_expr, filter_predicate, sum_spec, view};
use diesel::Connection as _;
use proptest::prelude::*;
#[derive(diesel::QueryableByName)]
struct SumRow {
#[diesel(sql_type = diesel::sql_types::BigInt)]
total: i64,
}
fn sqlite_sum(rows: &[(i64, bool)]) -> Result<i64, diesel::result::Error> {
let mut conn =
diesel::SqliteConnection::establish(":memory:").expect("in-memory database");
diesel::RunQueryDsl::execute(
diesel::sql_query(
"CREATE TABLE comments (id INTEGER PRIMARY KEY, \
score BIGINT NOT NULL, published BOOLEAN NOT NULL)",
),
&mut conn,
)
.expect("DDL");
for (score, published) in rows {
diesel::RunQueryDsl::execute(
diesel::sql_query("INSERT INTO comments (score, published) VALUES (?, ?)")
.bind::<diesel::sql_types::BigInt, _>(score)
.bind::<diesel::sql_types::Bool, _>(published),
&mut conn,
)
.expect("seed row");
}
let view = view(&sum_spec());
let sql = format!(
"SELECT {} AS total FROM \"comments\" AS {CHILD_ALIAS} WHERE 1 = 1{}",
aggregate_expr(&view, CHILD_ALIAS),
filter_predicate(&view, CHILD_ALIAS)
);
diesel::RunQueryDsl::get_result::<SumRow>(diesel::sql_query(sql), &mut conn)
.map(|row| row.total)
}
fn exact_sum(rows: &[(i64, bool)]) -> Option<i64> {
let total: i128 = rows
.iter()
.filter(|(_, published)| *published)
.map(|(score, _)| i128::from(*score))
.sum();
i64::try_from(total).ok()
}
fn published(scores: &[i64]) -> Vec<(i64, bool)> {
scores.iter().map(|score| (*score, true)).collect()
}
#[test]
fn an_empty_set_sums_to_zero() {
assert_eq!(sqlite_sum(&[]).expect("sum"), 0);
assert_eq!(sqlite_sum(&[(5, false)]).expect("sum"), 0);
}
#[test]
fn a_partial_sum_past_max_does_not_fail_the_scan() {
assert_eq!(
sqlite_sum(&published(&[i64::MAX, 1, -1])).expect("sum"),
i64::MAX
);
}
#[test]
fn a_partial_sum_past_min_does_not_fail_the_scan() {
assert_eq!(
sqlite_sum(&published(&[i64::MIN, -1, 1])).expect("sum"),
i64::MIN
);
}
#[test]
fn a_total_that_does_not_fit_is_an_error() {
for scores in [[i64::MAX, 1], [i64::MIN, -1]] {
let err = sqlite_sum(&published(&scores)).expect_err("no i64 holds the total");
assert!(err.to_string().contains("integer overflow"), "{err}");
}
}
fn weight() -> impl Strategy<Value = i64> {
prop_oneof![
Just(i64::MAX),
Just(i64::MIN),
Just(0),
Just(1),
Just(-1),
Just(1 << 32),
Just(-(1 << 32)),
Just((1 << 32) - 1),
any::<i64>(),
]
}
fn sqlite_bulk_decrement(
rows: &[(i64, bool)],
start: i64,
deleted: &[i64],
) -> Result<i64, diesel::result::Error> {
let mut conn =
diesel::SqliteConnection::establish(":memory:").expect("in-memory database");
for ddl in [
"CREATE TABLE posts (id INTEGER PRIMARY KEY, visible_score BIGINT NOT NULL)",
"CREATE TABLE comments (id INTEGER PRIMARY KEY, post_id BIGINT NOT NULL, \
score BIGINT NOT NULL, published BOOLEAN NOT NULL)",
] {
diesel::RunQueryDsl::execute(diesel::sql_query(ddl), &mut conn).expect("DDL");
}
diesel::RunQueryDsl::execute(
diesel::sql_query("INSERT INTO posts (id, visible_score) VALUES (1, ?)")
.bind::<diesel::sql_types::BigInt, _>(start),
&mut conn,
)
.expect("seed post");
for (score, published) in rows {
diesel::RunQueryDsl::execute(
diesel::sql_query(
"INSERT INTO comments (post_id, score, published) VALUES (1, ?, ?)",
)
.bind::<diesel::sql_types::BigInt, _>(score)
.bind::<diesel::sql_types::Bool, _>(published),
&mut conn,
)
.expect("seed row");
}
let sql = super::bulk_decrement_sql(&view(&sum_spec()), &super::id_list(deleted));
diesel::RunQueryDsl::execute(diesel::sql_query(sql), &mut conn)?;
diesel::RunQueryDsl::get_result::<SumRow>(
diesel::sql_query("SELECT visible_score AS total FROM posts WHERE id = 1"),
&mut conn,
)
.map(|row| row.total)
}
#[test]
fn a_bulk_decrement_whose_batch_sum_leaves_i64_still_lands() {
let rows = published(&[i64::MAX, i64::MAX, i64::MIN]);
assert_eq!(
sqlite_bulk_decrement(&rows, i64::MAX - 1, &[1, 2]).expect("decrement"),
i64::MIN
);
}
#[test]
fn a_bulk_decrement_whose_result_leaves_i64_is_an_error() {
let err = sqlite_bulk_decrement(&published(&[1]), i64::MIN, &[1])
.expect_err("MIN - 1 has no i64 value");
assert!(err.to_string().contains("integer overflow"), "{err}");
}
proptest! {
#[test]
fn the_sum_is_exact_or_an_overflow_error(
rows in prop::collection::vec((weight(), any::<bool>()), 0..8)
) {
match (exact_sum(&rows), sqlite_sum(&rows)) {
(Some(want), got) => prop_assert_eq!(got.ok(), Some(want)),
(None, Ok(got)) => prop_assert!(false, "{rows:?} gave {got}"),
(None, Err(err)) => {
prop_assert!(err.to_string().contains("integer overflow"), "{err}");
}
}
}
#[test]
fn the_bulk_decrement_is_exact_or_an_overflow_error(
rows in prop::collection::vec((weight(), any::<bool>(), any::<bool>()), 0..8)
) {
let seeded: Vec<(i64, bool)> = rows.iter().map(|(w, p, _)| (*w, *p)).collect();
let Some(start) = exact_sum(&seeded) else {
return Ok(());
};
let deleted: Vec<i64> = (1_i64..)
.zip(&rows)
.filter(|(_, (_, _, delete))| *delete)
.map(|(id, _)| id)
.collect();
let kept: Vec<(i64, bool)> = rows
.iter()
.filter(|(_, _, delete)| !*delete)
.map(|(w, p, _)| (*w, *p))
.collect();
let want = exact_sum(&kept);
let got = sqlite_bulk_decrement(&seeded, start, &deleted);
match (want, got) {
(Some(want), got) => prop_assert_eq!(got.ok(), Some(want)),
(None, Ok(got)) => prop_assert!(false, "{rows:?} gave {got}"),
(None, Err(err)) => {
prop_assert!(err.to_string().contains("integer overflow"), "{err}");
}
}
}
}
}
}