use indexmap::IndexMap as IndexMapRaw;
use alloc::vec;
use alloc::vec::Vec;
use core::fmt::Debug;
use core::hash::Hash;
use core::ops::{BitOr, BitOrAssign};
use crate::{
SchemaWithPK,
builders::{
ChangeDelete, ChangesetFormat, ChangesetOp, Insert, Operation, PatchDelete, PatchsetFormat,
PatchsetOp, Update, format::Format,
},
encoding::{MaybeValue, Value, encode_defined_value, encode_value, markers, op_codes},
};
type IndexMap<K, V> = IndexMapRaw<K, V, hashbrown::DefaultHashBuilder>;
type RowMap<F, S, B> = IndexMap<Vec<Value<S, B>>, Operation<F, S, B>>;
type TableMap<F, T, S, B> = IndexMap<T, RowMap<F, S, B>>;
type RowVec<F, S, B> = Vec<(Vec<Value<S, B>>, Operation<F, S, B>)>;
type TableVec<F, T, S, B> = Vec<(T, RowVec<F, S, B>)>;
const fn hash_append(h: u32, add: u32) -> u32 {
(h << 3) ^ h ^ add
}
#[allow(clippy::cast_sign_loss)]
fn session_hash_append_i64(h: u32, i: i64) -> u32 {
let lo = (i as u64 & 0xFFFF_FFFF) as u32;
let hi = ((i as u64 >> 32) & 0xFFFF_FFFF) as u32;
let h = hash_append(h, lo);
hash_append(h, hi)
}
fn session_hash_append_blob(mut h: u32, data: &[u8]) -> u32 {
for &byte in data {
h = hash_append(h, u32::from(byte));
}
h
}
fn session_hash_pk<S: AsRef<str>, B: AsRef<[u8]>>(pk: &[Value<S, B>]) -> u32 {
let mut h: u32 = 0;
for value in pk {
match value {
Value::Integer(i) => {
h = hash_append(h, 1); h = session_hash_append_i64(h, *i);
}
Value::Real(f) => {
h = hash_append(h, 2); let i = i64::from_ne_bytes(f.to_ne_bytes());
h = session_hash_append_i64(h, i);
}
Value::Text(s) => {
h = hash_append(h, 3); h = session_hash_append_blob(h, s.as_ref().as_bytes());
}
Value::Blob(b) => {
h = hash_append(h, 4); h = session_hash_append_blob(h, b.as_ref());
}
Value::Null => {
}
}
}
h
}
fn session_row_order<S: AsRef<str>, B: AsRef<[u8]>, V>(
rows: &IndexMap<Vec<Value<S, B>>, V>,
) -> Vec<usize> {
let n = rows.len();
if n == 0 {
return Vec::new();
}
let pks: Vec<&Vec<Value<S, B>>> = rows.keys().collect();
let mut n_change: usize = 0;
let mut buckets: Vec<Vec<usize>> = Vec::new();
for idx in 0..n {
if n_change == 0 || idx >= n_change / 2 {
let new_size = if n_change == 0 { 256 } else { n_change * 2 };
let mut new_buckets: Vec<Vec<usize>> = vec![Vec::new(); new_size];
for old_bucket in &buckets {
for &entry_idx in old_bucket.iter().rev() {
let h = session_hash_pk(pks[entry_idx]) as usize % new_size;
new_buckets[h].push(entry_idx);
}
}
buckets = new_buckets;
n_change = new_size;
}
let h = session_hash_pk(pks[idx]) as usize % n_change;
buckets[h].push(idx);
}
let mut order = Vec::with_capacity(n);
for bucket in &buckets {
for &idx in bucket.iter().rev() {
order.push(idx);
}
}
order
}
fn write_table_header<T: SchemaWithPK>(out: &mut Vec<u8>, marker: u8, table: &T) {
out.push(marker);
let num_cols = table.number_of_columns();
out.push(u8::try_from(num_cols).unwrap());
let pk_start = out.len();
out.resize(pk_start + num_cols, 0);
table.write_pk_flags(&mut out[pk_start..]);
out.extend(table.name().as_bytes());
out.push(0);
}
fn patchset_pk_mapping<T: SchemaWithPK>(table: &T) -> (Vec<u8>, Vec<Option<usize>>) {
let num_cols = table.number_of_columns();
let mut pk_flags = alloc::vec![0u8; num_cols];
table.write_pk_flags(&mut pk_flags);
let mut pk_col_to_pk_pos: Vec<Option<usize>> = alloc::vec![None; num_cols];
let mut pk_cols: Vec<(usize, u8)> = pk_flags
.iter()
.enumerate()
.filter_map(|(i, &ord)| if ord > 0 { Some((i, ord)) } else { None })
.collect();
pk_cols.sort_by_key(|(_, ord)| *ord);
for (pos, (col_idx, _)) in pk_cols.into_iter().enumerate() {
pk_col_to_pk_pos[col_idx] = Some(pos);
}
(pk_flags, pk_col_to_pk_pos)
}
fn encode_patchset_delete_values<S: AsRef<str>, B: AsRef<[u8]>>(
out: &mut Vec<u8>,
pk_flags: &[u8],
pk_col_to_pk_pos: &[Option<usize>],
pk: &[Value<S, B>],
) {
for (col_idx, &pk_flag) in pk_flags.iter().enumerate() {
if pk_flag > 0 {
if let Some(pk_pos) = pk_col_to_pk_pos[col_idx] {
encode_value(out, Some(&pk[pk_pos]));
} else {
encode_value::<S, B>(out, None);
}
}
}
}
fn encode_patchset_update_old_values<S: AsRef<str>, B: AsRef<[u8]>>(
out: &mut Vec<u8>,
pk_flags: &[u8],
pk_col_to_pk_pos: &[Option<usize>],
pk: &[Value<S, B>],
) {
for (col_idx, &pk_flag) in pk_flags.iter().enumerate() {
if pk_flag > 0
&& let Some(pk_pos) = pk_col_to_pk_pos[col_idx]
{
encode_defined_value(out, &pk[pk_pos]);
}
}
}
fn encode_changeset_op<S: AsRef<str> + Clone + Debug, B: AsRef<[u8]> + Clone + Debug>(
out: &mut Vec<u8>,
op: &Operation<ChangesetFormat, S, B>,
) {
match op {
Operation::Insert { values, indirect } => {
out.push(op_codes::INSERT);
out.push(u8::from(*indirect));
for value in values {
encode_value(out, Some(value));
}
}
Operation::Delete {
data: values,
indirect,
} => {
out.push(op_codes::DELETE);
out.push(u8::from(*indirect));
for value in values {
encode_value(out, Some(value));
}
}
Operation::Update { values, indirect } => {
out.push(op_codes::UPDATE);
out.push(u8::from(*indirect));
for (old, _new) in values {
encode_value(out, old.as_ref());
}
for (_old, new) in values {
encode_value(out, new.as_ref());
}
}
}
}
fn encode_patchset_op<S: AsRef<str>, B: AsRef<[u8]>>(
out: &mut Vec<u8>,
op: &Operation<PatchsetFormat, S, B>,
pk: &[Value<S, B>],
pk_flags: &[u8],
pk_col_to_pk_pos: &[Option<usize>],
) {
match op {
Operation::Insert { values, indirect } => {
out.push(op_codes::INSERT);
out.push(u8::from(*indirect));
for value in values {
encode_value(out, Some(value));
}
}
Operation::Delete { data: (), indirect } => {
out.push(op_codes::DELETE);
out.push(u8::from(*indirect));
encode_patchset_delete_values(out, pk_flags, pk_col_to_pk_pos, pk);
}
Operation::Update { values, indirect } => {
out.push(op_codes::UPDATE);
out.push(u8::from(*indirect));
encode_patchset_update_old_values(out, pk_flags, pk_col_to_pk_pos, pk);
for (col_idx, &pk_flag) in pk_flags.iter().enumerate() {
if pk_flag == 0 {
encode_value(out, values[col_idx].1.as_ref());
}
}
}
}
}
#[derive(Debug, Clone)]
pub struct DiffSetBuilder<F: Format<S, B>, T: SchemaWithPK, S, B> {
pub(crate) tables: TableMap<F, T, S, B>,
}
impl<F: Format<S, B>, T: SchemaWithPK, S, B> PartialEq for DiffSetBuilder<F, T, S, B>
where
S: PartialEq + Eq + Hash + AsRef<str>,
B: PartialEq + Eq + Hash + AsRef<[u8]>,
F::Old: PartialEq,
F::DeleteData: PartialEq,
{
fn eq(&self, other: &Self) -> bool {
self.tables
.iter()
.filter(|(_, ops)| !ops.is_empty())
.eq(other.tables.iter().filter(|(_, ops)| !ops.is_empty()))
}
}
impl<F: Format<S, B>, T: SchemaWithPK, S, B> Eq for DiffSetBuilder<F, T, S, B>
where
S: Eq + Hash + AsRef<str>,
B: Eq + Hash + AsRef<[u8]>,
F::Old: Eq,
F::DeleteData: Eq,
{
}
pub type ChangeSet<T, S, B> = DiffSetBuilder<ChangesetFormat, T, S, B>;
pub type PatchSet<T, S, B> = DiffSetBuilder<PatchsetFormat, T, S, B>;
impl<F: Format<S, B>, T: SchemaWithPK, S: AsRef<str> + Hash + Eq, B: AsRef<[u8]> + Hash + Eq>
Default for DiffSetBuilder<F, T, S, B>
{
fn default() -> Self {
Self::new()
}
}
impl<
T: SchemaWithPK,
S: Clone + Debug + Hash + Eq + AsRef<str>,
B: Clone + Debug + Hash + Eq + AsRef<[u8]>,
> From<&DiffSetBuilder<ChangesetFormat, T, S, B>> for Vec<u8>
{
#[inline]
fn from(builder: &DiffSetBuilder<ChangesetFormat, T, S, B>) -> Self {
builder.build()
}
}
impl<
T: SchemaWithPK,
S: Clone + Debug + Hash + Eq + AsRef<str>,
B: Clone + Debug + Hash + Eq + AsRef<[u8]>,
> From<DiffSetBuilder<ChangesetFormat, T, S, B>> for Vec<u8>
{
#[inline]
fn from(builder: DiffSetBuilder<ChangesetFormat, T, S, B>) -> Self {
builder.build()
}
}
impl<T: SchemaWithPK, S: AsRef<str> + Clone + Hash + Eq, B: AsRef<[u8]> + Clone + Hash + Eq>
From<&DiffSetBuilder<PatchsetFormat, T, S, B>> for Vec<u8>
{
#[inline]
fn from(builder: &DiffSetBuilder<PatchsetFormat, T, S, B>) -> Self {
builder.build()
}
}
impl<T: SchemaWithPK, S: AsRef<str> + Clone + Hash + Eq, B: AsRef<[u8]> + Clone + Hash + Eq>
From<DiffSetBuilder<PatchsetFormat, T, S, B>> for Vec<u8>
{
#[inline]
fn from(builder: DiffSetBuilder<PatchsetFormat, T, S, B>) -> Self {
builder.build()
}
}
impl<F: Format<S, B>, T: SchemaWithPK, S, B> DiffSetBuilder<F, T, S, B> {
pub(super) fn table<'builder>(&'builder self, name: &str) -> Option<&'builder T> {
self.tables.keys().find(|t| t.name() == name)
}
}
impl<F: Format<S, B>, T: SchemaWithPK, S: AsRef<str> + Hash + Eq, B: AsRef<[u8]> + Hash + Eq>
DiffSetBuilder<F, T, S, B>
{
#[inline]
#[must_use]
pub fn new() -> Self {
Self {
tables: IndexMap::default(),
}
}
#[inline]
fn ensure_table(&mut self, table: &T) -> &mut RowMap<F, S, B> {
self.tables.entry(table.clone()).or_default()
}
pub fn add_table(&mut self, table: &T) -> &mut Self {
self.ensure_table(table);
self
}
#[inline]
#[must_use]
pub fn is_empty(&self) -> bool {
self.tables.values().all(IndexMap::is_empty)
}
#[inline]
#[must_use]
pub fn len(&self) -> usize {
self.tables.values().map(IndexMap::len).sum()
}
pub(crate) fn add_operation(
&mut self,
table: &T,
pk: Vec<Value<S, B>>,
new_op: Operation<F, S, B>,
) -> &mut Self
where
S: Clone,
B: Clone,
Operation<F, S, B>: core::ops::Add<Output = Option<Operation<F, S, B>>>,
{
let rows = self.ensure_table(table);
match rows.shift_remove_full(&pk) {
None => {
rows.insert(pk, new_op);
}
Some((original_index, _removed_key, existing)) => {
match (&existing, &new_op) {
(Operation::Insert { .. }, Operation::Update { .. }) => {
if let Some(combined) = existing + new_op
&& let Operation::Insert { values, .. } = &combined
{
let new_pk = table.extract_pk(values);
rows.shift_remove(&new_pk);
let index = original_index.min(rows.len());
rows.shift_insert(index, new_pk, combined);
}
}
_ => {
if let Some(combined) = existing + new_op {
rows.shift_insert(original_index, pk, combined);
}
}
}
}
}
self
}
}
pub trait DiffOps<T: SchemaWithPK, S, B>: Sized {
type Format: Format<S, B>;
type DeleteArg;
fn insert(self, insert: Insert<T, S, B>) -> DiffSetBuilder<Self::Format, T, S, B>;
fn delete(self, delete: Self::DeleteArg) -> DiffSetBuilder<Self::Format, T, S, B>;
fn update(self, update: Update<T, Self::Format, S, B>)
-> DiffSetBuilder<Self::Format, T, S, B>;
}
impl<
T: SchemaWithPK,
S: Clone + Debug + Hash + Eq + AsRef<str>,
B: Clone + Debug + Hash + Eq + AsRef<[u8]>,
> DiffOps<T, S, B> for DiffSetBuilder<ChangesetFormat, T, S, B>
{
type Format = ChangesetFormat;
type DeleteArg = ChangeDelete<T, S, B>;
fn insert(mut self, insert: Insert<T, S, B>) -> Self {
let pk = insert.extract_pk();
let table = insert.as_ref().clone();
let indirect = insert.indirect;
self.add_operation(
&table,
pk,
Operation::Insert {
values: insert.into_values(),
indirect,
},
);
self
}
fn delete(mut self, delete: ChangeDelete<T, S, B>) -> Self {
let pk = delete.as_ref().extract_pk(&delete.values);
let table = delete.as_ref().clone();
let indirect = delete.indirect;
self.add_operation(
&table,
pk,
Operation::Delete {
data: delete.into_values(),
indirect,
},
);
self
}
fn update(mut self, update: Update<T, ChangesetFormat, S, B>) -> Self {
let old_values: Vec<_> = update
.values()
.iter()
.map(|(old, _): &(_, _)| old.clone().unwrap_or(Value::Null))
.collect();
let pk = update.as_ref().extract_pk(&old_values);
let table = update.as_ref().clone();
let indirect = update.indirect;
let values: Vec<(MaybeValue<S, B>, MaybeValue<S, B>)> = update.into();
self.add_operation(&table, pk, Operation::Update { values, indirect });
self
}
}
impl<T: SchemaWithPK, S: Clone + Hash + Eq + AsRef<str>, B: Clone + Hash + Eq + AsRef<[u8]>>
DiffOps<T, S, B> for DiffSetBuilder<PatchsetFormat, T, S, B>
{
type Format = PatchsetFormat;
type DeleteArg = PatchDelete<T, S, B>;
fn insert(mut self, insert: Insert<T, S, B>) -> Self {
let pk = insert.extract_pk();
let table = insert.as_ref().clone();
let indirect = insert.indirect;
self.add_operation(
&table,
pk,
Operation::Insert {
values: insert.into_values(),
indirect,
},
);
self
}
fn delete(mut self, delete: PatchDelete<T, S, B>) -> Self {
let indirect = delete.indirect;
self.add_operation(
&delete.table,
delete.pk,
Operation::Delete { data: (), indirect },
);
self
}
fn update(mut self, update: Update<T, PatchsetFormat, S, B>) -> Self {
let pk = update.extract_pk();
let table = update.as_ref().clone();
let indirect = update.indirect;
let values: Vec<((), MaybeValue<S, B>)> = update.into();
self.add_operation(&table, pk, Operation::Update { values, indirect });
self
}
}
impl<
T: SchemaWithPK,
S: Clone + Debug + Hash + Eq + AsRef<str>,
B: Clone + Debug + Hash + Eq + AsRef<[u8]>,
> DiffOps<T, S, B> for DiffSet<ChangesetFormat, T, S, B>
{
type Format = ChangesetFormat;
type DeleteArg = ChangeDelete<T, S, B>;
fn insert(self, insert: Insert<T, S, B>) -> DiffSetBuilder<ChangesetFormat, T, S, B> {
let builder: DiffSetBuilder<ChangesetFormat, T, S, B> = self.into();
builder.insert(insert)
}
fn delete(self, delete: ChangeDelete<T, S, B>) -> DiffSetBuilder<ChangesetFormat, T, S, B> {
let builder: DiffSetBuilder<ChangesetFormat, T, S, B> = self.into();
builder.delete(delete)
}
fn update(
self,
update: Update<T, ChangesetFormat, S, B>,
) -> DiffSetBuilder<ChangesetFormat, T, S, B> {
let builder: DiffSetBuilder<ChangesetFormat, T, S, B> = self.into();
builder.update(update)
}
}
impl<T: SchemaWithPK, S: Clone + Hash + Eq + AsRef<str>, B: Clone + Hash + Eq + AsRef<[u8]>>
DiffOps<T, S, B> for DiffSet<PatchsetFormat, T, S, B>
{
type Format = PatchsetFormat;
type DeleteArg = PatchDelete<T, S, B>;
fn insert(self, insert: Insert<T, S, B>) -> DiffSetBuilder<PatchsetFormat, T, S, B> {
let builder: DiffSetBuilder<PatchsetFormat, T, S, B> = self.into();
builder.insert(insert)
}
fn delete(self, delete: PatchDelete<T, S, B>) -> DiffSetBuilder<PatchsetFormat, T, S, B> {
let builder: DiffSetBuilder<PatchsetFormat, T, S, B> = self.into();
builder.delete(delete)
}
fn update(
self,
update: Update<T, PatchsetFormat, S, B>,
) -> DiffSetBuilder<PatchsetFormat, T, S, B> {
let builder: DiffSetBuilder<PatchsetFormat, T, S, B> = self.into();
builder.update(update)
}
}
impl<T: crate::schema::NamedColumns, S: Clone + Hash + Eq + AsRef<str> + for<'a> From<&'a str>>
DiffSetBuilder<PatchsetFormat, T, S, Vec<u8>>
{
pub fn digest_sql<'input>(
&mut self,
input: &'input str,
) -> Result<&mut Self, crate::builders::sql::ParseError<'input>> {
let mut parser = crate::builders::sql::Parser::new(input, self);
parser.digest_all()?;
Ok(self)
}
}
impl<
T: SchemaWithPK,
S: Clone + Debug + Hash + Eq + AsRef<str>,
B: Clone + Debug + Hash + Eq + AsRef<[u8]>,
> DiffSetBuilder<ChangesetFormat, T, S, B>
{
#[must_use]
pub fn build(&self) -> Vec<u8> {
let mut out = Vec::new();
for (table, rows) in &self.tables {
if rows.is_empty() {
continue;
}
write_table_header(&mut out, markers::CHANGESET, table);
for idx in session_row_order(rows) {
let (_pk, op) = rows.get_index(idx).unwrap();
encode_changeset_op(&mut out, op);
}
}
out
}
}
impl<T: SchemaWithPK, S: Clone + Hash + Eq + AsRef<str>, B: Clone + Hash + Eq + AsRef<[u8]>>
DiffSetBuilder<PatchsetFormat, T, S, B>
{
#[must_use]
pub fn build(&self) -> Vec<u8> {
let mut out = Vec::new();
for (table, rows) in &self.tables {
if rows.is_empty() {
continue;
}
write_table_header(&mut out, markers::PATCHSET, table);
let (pk_flags, pk_col_to_pk_pos) = patchset_pk_mapping(table);
for idx in session_row_order(rows) {
let (pk, op) = rows.get_index(idx).unwrap();
encode_patchset_op(&mut out, op, pk, &pk_flags, &pk_col_to_pk_pos);
}
}
out
}
}
use crate::builders::operation::Reverse;
impl<
T: SchemaWithPK,
S: Clone + Debug + Hash + Eq + AsRef<str>,
B: Clone + Debug + Hash + Eq + AsRef<[u8]>,
> Reverse for DiffSetBuilder<ChangesetFormat, T, S, B>
{
type Output = DiffSetBuilder<ChangesetFormat, T, S, B>;
fn reverse(self) -> Self::Output {
let mut reversed: DiffSetBuilder<ChangesetFormat, T, S, B> = DiffSetBuilder::new();
for (table, rows) in self.tables {
for (pk, op) in rows {
let rev_op = op.reverse();
reversed.add_operation(&table, pk, rev_op);
}
}
reversed
}
}
impl<
F: Format<S, B>,
T: SchemaWithPK,
S: Clone + Hash + Eq + AsRef<str>,
B: Clone + Hash + Eq + AsRef<[u8]>,
> BitOrAssign for DiffSetBuilder<F, T, S, B>
where
Operation<F, S, B>: core::ops::Add<Output = Option<Operation<F, S, B>>>,
{
fn bitor_assign(&mut self, rhs: Self) {
for (table, rows) in rhs.tables {
for (pk, op) in rows {
self.add_operation(&table, pk, op);
}
}
}
}
impl<
F: Format<S, B>,
T: SchemaWithPK,
S: Clone + Hash + Eq + AsRef<str>,
B: Clone + Hash + Eq + AsRef<[u8]>,
> BitOr for DiffSetBuilder<F, T, S, B>
where
Operation<F, S, B>: core::ops::Add<Output = Option<Operation<F, S, B>>>,
{
type Output = Self;
#[inline]
fn bitor(mut self, rhs: Self) -> Self::Output {
self |= rhs;
self
}
}
#[derive(Debug, Clone)]
pub struct DiffSet<F: Format<S, B>, T: SchemaWithPK, S, B> {
pub(crate) tables: TableVec<F, T, S, B>,
}
impl<F: Format<S, B>, T: SchemaWithPK, S, B> PartialEq for DiffSet<F, T, S, B>
where
S: PartialEq + Eq + Hash + AsRef<str>,
B: PartialEq + Eq + Hash + AsRef<[u8]>,
F::Old: PartialEq,
F::DeleteData: PartialEq,
{
fn eq(&self, other: &Self) -> bool {
self.tables
.iter()
.filter(|(_, ops)| !ops.is_empty())
.eq(other.tables.iter().filter(|(_, ops)| !ops.is_empty()))
}
}
impl<F: Format<S, B>, T: SchemaWithPK, S, B> Eq for DiffSet<F, T, S, B>
where
S: Eq + Hash + AsRef<str>,
B: Eq + Hash + AsRef<[u8]>,
F::Old: Eq,
F::DeleteData: Eq,
{
}
impl<F: Format<S, B>, T: SchemaWithPK, S: AsRef<str> + Hash + Eq, B: AsRef<[u8]> + Hash + Eq>
Default for DiffSet<F, T, S, B>
{
fn default() -> Self {
Self { tables: Vec::new() }
}
}
impl<F: Format<S, B>, T: SchemaWithPK, S: AsRef<str> + Hash + Eq, B: AsRef<[u8]> + Hash + Eq>
DiffSet<F, T, S, B>
{
#[inline]
#[must_use]
pub fn is_empty(&self) -> bool {
self.tables.iter().all(|(_, rows)| rows.is_empty())
}
#[inline]
#[must_use]
pub fn len(&self) -> usize {
self.tables.iter().map(|(_, rows)| rows.len()).sum()
}
}
impl<F: Format<S, B>, T: SchemaWithPK, S: AsRef<str> + Hash + Eq, B: AsRef<[u8]> + Hash + Eq>
DiffSet<F, T, S, B>
{
pub fn tables(&self) -> impl Iterator<Item = &T> {
self.tables
.iter()
.filter(|(_, rows)| !rows.is_empty())
.map(|(t, _)| t)
}
}
impl<T: SchemaWithPK, S: Clone + Debug + AsRef<str>, B: Clone + Debug + AsRef<[u8]>>
DiffSet<ChangesetFormat, T, S, B>
{
pub fn iter(&self) -> impl Iterator<Item = ChangesetOp<'_, T, S, B>> {
self.tables.iter().flat_map(|(table, rows)| {
rows.iter().map(move |(_pk, op)| match op {
Operation::Insert { values, indirect } => ChangesetOp::Insert {
table,
values: values.as_slice(),
indirect: *indirect,
},
Operation::Update { values, indirect } => ChangesetOp::Update {
table,
values: values.as_slice(),
indirect: *indirect,
},
Operation::Delete { data, indirect } => ChangesetOp::Delete {
table,
old_values: data.as_slice(),
indirect: *indirect,
},
})
})
}
}
impl<T: SchemaWithPK, S: Clone + AsRef<str>, B: Clone + AsRef<[u8]>>
DiffSet<PatchsetFormat, T, S, B>
{
pub fn iter(&self) -> impl Iterator<Item = PatchsetOp<'_, T, S, B>> {
self.tables.iter().flat_map(|(table, rows)| {
rows.iter().map(move |(pk, op)| match op {
Operation::Insert { values, indirect } => PatchsetOp::Insert {
table,
values: values.as_slice(),
indirect: *indirect,
},
Operation::Update { values, indirect } => PatchsetOp::Update {
table,
pk: pk.as_slice(),
entries: values.as_slice(),
indirect: *indirect,
},
Operation::Delete { indirect, .. } => PatchsetOp::Delete {
table,
pk: pk.as_slice(),
indirect: *indirect,
},
})
})
}
}
impl<
T: SchemaWithPK,
S: Clone + Debug + Hash + Eq + AsRef<str>,
B: Clone + Debug + Hash + Eq + AsRef<[u8]>,
> DiffSet<ChangesetFormat, T, S, B>
{
#[must_use]
pub fn build(&self) -> Vec<u8> {
let mut out = Vec::new();
for (table, rows) in &self.tables {
if rows.is_empty() {
continue;
}
write_table_header(&mut out, markers::CHANGESET, table);
for (_pk, op) in rows {
encode_changeset_op(&mut out, op);
}
}
out
}
}
impl<T: SchemaWithPK, S: Clone + Hash + Eq + AsRef<str>, B: Clone + Hash + Eq + AsRef<[u8]>>
DiffSet<PatchsetFormat, T, S, B>
{
#[must_use]
pub fn build(&self) -> Vec<u8> {
let mut out = Vec::new();
for (table, rows) in &self.tables {
if rows.is_empty() {
continue;
}
write_table_header(&mut out, markers::PATCHSET, table);
let (pk_flags, pk_col_to_pk_pos) = patchset_pk_mapping(table);
for (pk, op) in rows {
encode_patchset_op(&mut out, op, pk, &pk_flags, &pk_col_to_pk_pos);
}
}
out
}
}
impl<
T: SchemaWithPK,
S: Clone + Debug + Hash + Eq + AsRef<str>,
B: Clone + Debug + Hash + Eq + AsRef<[u8]>,
> From<&DiffSet<ChangesetFormat, T, S, B>> for Vec<u8>
{
#[inline]
fn from(diffset: &DiffSet<ChangesetFormat, T, S, B>) -> Self {
diffset.build()
}
}
impl<
T: SchemaWithPK,
S: Clone + Debug + Hash + Eq + AsRef<str>,
B: Clone + Debug + Hash + Eq + AsRef<[u8]>,
> From<DiffSet<ChangesetFormat, T, S, B>> for Vec<u8>
{
#[inline]
fn from(diffset: DiffSet<ChangesetFormat, T, S, B>) -> Self {
diffset.build()
}
}
impl<T: SchemaWithPK, S: AsRef<str> + Clone + Hash + Eq, B: AsRef<[u8]> + Clone + Hash + Eq>
From<&DiffSet<PatchsetFormat, T, S, B>> for Vec<u8>
{
#[inline]
fn from(diffset: &DiffSet<PatchsetFormat, T, S, B>) -> Self {
diffset.build()
}
}
impl<T: SchemaWithPK, S: AsRef<str> + Clone + Hash + Eq, B: AsRef<[u8]> + Clone + Hash + Eq>
From<DiffSet<PatchsetFormat, T, S, B>> for Vec<u8>
{
#[inline]
fn from(diffset: DiffSet<PatchsetFormat, T, S, B>) -> Self {
diffset.build()
}
}
impl<
T: SchemaWithPK,
S: Clone + Debug + Hash + Eq + AsRef<str>,
B: Clone + Debug + Hash + Eq + AsRef<[u8]>,
> Reverse for DiffSet<ChangesetFormat, T, S, B>
{
type Output = DiffSet<ChangesetFormat, T, S, B>;
fn reverse(self) -> Self::Output {
DiffSet {
tables: self
.tables
.into_iter()
.map(|(table, rows)| {
let rev_rows = rows
.into_iter()
.map(|(pk, op)| (pk, op.reverse()))
.collect();
(table, rev_rows)
})
.collect(),
}
}
}
impl<F: Format<S, B>, T: SchemaWithPK, S: Hash + Eq + AsRef<str>, B: Hash + Eq + AsRef<[u8]>>
From<DiffSetBuilder<F, T, S, B>> for DiffSet<F, T, S, B>
{
fn from(builder: DiffSetBuilder<F, T, S, B>) -> Self {
Self {
tables: builder
.tables
.into_iter()
.map(|(table, rows)| {
let ordered_rows: RowVec<F, S, B> = rows.into_iter().collect();
(table, ordered_rows)
})
.collect(),
}
}
}
impl<F: Format<S, B>, T: SchemaWithPK, S: Hash + Eq + AsRef<str>, B: Hash + Eq + AsRef<[u8]>>
From<DiffSet<F, T, S, B>> for DiffSetBuilder<F, T, S, B>
{
fn from(diffset: DiffSet<F, T, S, B>) -> Self {
let mut builder = Self::new();
for (table, rows) in diffset.tables {
let map: IndexMap<Vec<Value<S, B>>, Operation<F, S, B>> = rows.into_iter().collect();
builder.tables.insert(table, map);
}
builder
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::builders::operation::Indirect;
use crate::encoding::Value;
use alloc::{string::String, vec};
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
struct TestTable {
name: String,
num_columns: usize,
pk_column: usize,
}
impl TestTable {
fn new(name: &str, num_columns: usize, pk_column: usize) -> Self {
Self {
name: name.into(),
num_columns,
pk_column,
}
}
}
impl crate::DynTable for TestTable {
fn name(&self) -> &str {
&self.name
}
fn number_of_columns(&self) -> usize {
self.num_columns
}
fn write_pk_flags(&self, buf: &mut [u8]) {
assert_eq!(buf.len(), self.num_columns);
buf.fill(0);
buf[self.pk_column] = 1;
}
}
impl crate::SchemaWithPK for TestTable {
fn number_of_primary_keys(&self) -> usize {
1
}
fn primary_key_index(&self, col_idx: usize) -> Option<usize> {
if col_idx == self.pk_column {
Some(0)
} else {
None
}
}
fn extract_pk<S: Clone, B: Clone>(
&self,
values: &impl crate::IndexableValues<Text = S, Binary = B>,
) -> alloc::vec::Vec<Value<S, B>> {
alloc::vec![
values
.get(self.pk_column)
.expect("primary key column index out of bounds, values shorter than schema")
]
}
}
type ChangesetBuilder = DiffSetBuilder<ChangesetFormat, TestTable, String, Vec<u8>>;
#[test]
fn test_insert_single_row() {
let table = TestTable::new("users", 2, 0);
let insert = Insert::from(table.clone())
.set(0, 1i64)
.unwrap()
.set(1, "alice")
.unwrap();
let builder = ChangesetBuilder::new().insert(insert);
assert_eq!(builder.len(), 1);
assert!(!builder.is_empty());
}
#[test]
fn test_insert_then_delete_cancels_out() {
let table = TestTable::new("users", 2, 0);
let insert = Insert::from(table.clone())
.set(0, 1i64)
.unwrap()
.set(1, "alice")
.unwrap();
let delete = ChangeDelete::from(table.clone())
.set(0, 1i64)
.unwrap()
.set(1, "alice")
.unwrap();
let builder = ChangesetBuilder::new().insert(insert).delete(delete);
assert_eq!(builder.len(), 0);
assert!(builder.is_empty());
}
#[test]
fn test_insert_then_update_becomes_insert() {
let table = TestTable::new("users", 2, 0);
let insert = Insert::from(table.clone())
.set(0, 1i64)
.unwrap()
.set(1, "alice")
.unwrap();
let update = Update::<TestTable, ChangesetFormat, String, Vec<u8>>::from(table.clone())
.set(0, 1i64, 1i64) .unwrap()
.set(1, "alice", "bob")
.unwrap();
let builder = ChangesetBuilder::new().insert(insert).update(update);
assert_eq!(builder.len(), 1);
}
#[test]
fn test_delete_then_insert_same_values_cancels_out() {
let table = TestTable::new("users", 2, 0);
let delete = ChangeDelete::from(table.clone())
.set(0, 1i64)
.unwrap()
.set(1, "alice")
.unwrap();
let insert = Insert::from(table.clone())
.set(0, 1i64)
.unwrap()
.set(1, "alice")
.unwrap();
let builder = ChangesetBuilder::new().delete(delete).insert(insert);
assert_eq!(builder.len(), 0);
assert!(builder.is_empty());
}
#[test]
fn test_delete_then_insert_different_values_becomes_update() {
let table = TestTable::new("users", 2, 0);
let delete = ChangeDelete::from(table.clone())
.set(0, 1i64)
.unwrap()
.set(1, "alice")
.unwrap();
let insert = Insert::from(table.clone())
.set(0, 1i64)
.unwrap()
.set(1, "bob")
.unwrap();
let builder = ChangesetBuilder::new().delete(delete).insert(insert);
assert_eq!(builder.len(), 1);
}
#[test]
fn test_multiple_rows() {
let table = TestTable::new("users", 2, 0);
let insert1 = Insert::from(table.clone())
.set(0, 1i64)
.unwrap()
.set(1, "alice")
.unwrap();
let insert2 = Insert::from(table.clone())
.set(0, 2i64)
.unwrap()
.set(1, "bob")
.unwrap();
let builder = ChangesetBuilder::new().insert(insert1).insert(insert2);
assert_eq!(builder.len(), 2);
}
#[test]
fn test_update_then_update_consolidates() {
let table = TestTable::new("users", 2, 0);
let update1 = Update::<TestTable, ChangesetFormat, String, Vec<u8>>::from(table.clone())
.set(0, 1i64, 1i64)
.unwrap()
.set(1, "alice", "bob")
.unwrap();
let update2 = Update::<TestTable, ChangesetFormat, String, Vec<u8>>::from(table.clone())
.set(0, 1i64, 1i64)
.unwrap()
.set(1, "bob", "charlie")
.unwrap();
let builder = ChangesetBuilder::new().update(update1).update(update2);
assert_eq!(builder.len(), 1);
}
#[test]
fn test_reverse_operation_insert_becomes_delete() {
let op: Operation<ChangesetFormat, String, Vec<u8>> = Operation::Insert {
values: vec![Value::Integer(1), Value::Text("alice".into())],
indirect: false,
};
let reversed = op.reverse();
let Operation::Delete { data, .. } = reversed else {
panic!("Expected Delete operation");
};
assert_eq!(
data,
vec![
Value::<String, Vec<u8>>::Integer(1),
Value::Text("alice".into())
]
);
}
#[test]
fn test_reverse_operation_delete_becomes_insert() {
let op: Operation<ChangesetFormat, String, Vec<u8>> = Operation::Delete {
data: vec![Value::Integer(1), Value::Text("alice".into())],
indirect: false,
};
let reversed = op.reverse();
let Operation::Insert { values, .. } = reversed else {
panic!("Expected Insert operation");
};
assert_eq!(
values,
vec![
Value::<String, Vec<u8>>::Integer(1),
Value::Text("alice".into())
]
);
}
#[test]
fn test_reverse_operation_update_swaps_old_new() {
let op: Operation<ChangesetFormat, String, Vec<u8>> = Operation::Update {
values: vec![
(Some(Value::Integer(1)), Some(Value::Integer(1))),
(
Some(Value::Text("alice".into())),
Some(Value::Text("bob".into())),
),
],
indirect: false,
};
let reversed = op.reverse();
let Operation::Update { values, .. } = reversed else {
panic!("Expected Update operation");
};
assert_eq!(
values[0],
(Some(Value::Integer(1)), Some(Value::Integer(1)))
);
assert_eq!(
values[1],
(
Some(Value::Text("bob".into())),
Some(Value::Text("alice".into()))
)
);
}
#[test]
fn test_reverse_builder_insert_becomes_delete() {
let table = TestTable::new("users", 2, 0);
let insert = Insert::from(table.clone())
.set(0, 1i64)
.unwrap()
.set(1, "alice")
.unwrap();
let builder = ChangesetBuilder::new().insert(insert);
let reversed = builder.reverse();
assert_eq!(reversed.len(), 1);
let rows = reversed.tables.get(&table).unwrap();
assert!(matches!(
rows.values().next().unwrap(),
Operation::Delete { .. }
));
}
#[test]
fn test_reverse_builder_delete_becomes_insert() {
let table = TestTable::new("users", 2, 0);
let delete = ChangeDelete::from(table.clone())
.set(0, 1i64)
.unwrap()
.set(1, "alice")
.unwrap();
let builder = ChangesetBuilder::new().delete(delete);
let reversed = builder.reverse();
assert_eq!(reversed.len(), 1);
let rows = reversed.tables.get(&table).unwrap();
assert!(matches!(
rows.values().next().unwrap(),
Operation::Insert { .. }
));
}
#[test]
fn test_reverse_builder_update_swaps() {
let table = TestTable::new("users", 2, 0);
let update = Update::<TestTable, ChangesetFormat, String, Vec<u8>>::from(table.clone())
.set(0, 1i64, 1i64)
.unwrap()
.set(1, "alice", "bob")
.unwrap();
let builder = ChangesetBuilder::new().update(update);
let reversed = builder.reverse();
assert_eq!(reversed.len(), 1);
let rows = reversed.tables.get(&table).unwrap();
let Operation::Update { values, .. } = rows.values().next().unwrap() else {
panic!("Expected Update operation");
};
assert_eq!(
values[1],
(
Some(Value::Text("bob".into())),
Some(Value::Text("alice".into()))
)
);
}
#[test]
fn test_reverse_is_involutory() {
let table = TestTable::new("users", 2, 0);
let insert = Insert::from(table.clone())
.set(0, 1i64)
.unwrap()
.set(1, "alice")
.unwrap();
let original_values = insert.into_values();
let insert2 = Insert::from(table.clone())
.set(0, 1i64)
.unwrap()
.set(1, "alice")
.unwrap();
let builder = ChangesetBuilder::new().insert(insert2);
let double_reversed = builder.reverse().reverse();
assert_eq!(double_reversed.len(), 1);
let rows = double_reversed.tables.get(&table).unwrap();
let Operation::Insert { values, .. } = rows.values().next().unwrap() else {
panic!("Expected Insert operation");
};
assert_eq!(values, &original_values);
}
#[test]
fn test_build_empty_builder() {
let builder = ChangesetBuilder::new();
let bytes = builder.build();
assert!(bytes.is_empty());
}
#[test]
fn test_build_insert_format() {
let table = TestTable::new("t", 2, 0);
let insert = Insert::from(table.clone())
.set(0, 1i64)
.unwrap()
.set(1, "a")
.unwrap();
let builder = ChangesetBuilder::new().insert(insert);
let bytes = builder.build();
assert!(!bytes.is_empty());
assert_eq!(bytes[0], b'T');
assert_eq!(bytes[1], 2);
assert_eq!(bytes[2], 1);
assert_eq!(bytes[3], 0);
assert_eq!(bytes[4], b't');
assert_eq!(bytes[5], 0);
assert_eq!(bytes[6], 0x12);
assert_eq!(bytes[7], 0);
}
#[test]
fn test_build_delete_format() {
let table = TestTable::new("t", 2, 0);
let delete = ChangeDelete::from(table.clone())
.set(0, 1i64)
.unwrap()
.set(1, "a")
.unwrap();
let builder = ChangesetBuilder::new().delete(delete);
let bytes = builder.build();
assert!(!bytes.is_empty());
assert_eq!(bytes[0], b'T');
assert_eq!(bytes[6], 0x09);
}
#[test]
fn test_build_update_format() {
let table = TestTable::new("t", 2, 0);
let update = Update::<TestTable, ChangesetFormat, String, Vec<u8>>::from(table.clone())
.set(0, 1i64, 1i64)
.unwrap()
.set(1, "a", "b")
.unwrap();
let builder = ChangesetBuilder::new().update(update);
let bytes = builder.build();
assert!(!bytes.is_empty());
assert_eq!(bytes[0], b'T');
assert_eq!(bytes[6], 0x17);
}
#[test]
fn test_build_multiple_operations() {
let table = TestTable::new("t", 2, 0);
let insert = Insert::from(table.clone())
.set(0, 1i64)
.unwrap()
.set(1, "a")
.unwrap();
let insert2 = Insert::from(table.clone())
.set(0, 2i64)
.unwrap()
.set(1, "b")
.unwrap();
let builder = ChangesetBuilder::new().insert(insert).insert(insert2);
let bytes = builder.build();
assert!(!bytes.is_empty());
assert_eq!(bytes[0], b'T');
}
#[test]
fn test_build_cancelled_operations_produce_empty() {
let table = TestTable::new("t", 2, 0);
let insert = Insert::from(table.clone())
.set(0, 1i64)
.unwrap()
.set(1, "a")
.unwrap();
let delete = ChangeDelete::from(table.clone())
.set(0, 1i64)
.unwrap()
.set(1, "a")
.unwrap();
let builder = ChangesetBuilder::new().insert(insert).delete(delete);
let bytes = builder.build();
assert!(bytes.is_empty());
}
#[test]
fn test_bitor_changeset_disjoint_rows() {
let table = TestTable::new("users", 2, 0);
let insert1 = Insert::from(table.clone())
.set(0, 1i64)
.unwrap()
.set(1, "alice")
.unwrap();
let insert2 = Insert::from(table.clone())
.set(0, 2i64)
.unwrap()
.set(1, "bob")
.unwrap();
let cs1 = ChangesetBuilder::new().insert(insert1);
let cs2 = ChangesetBuilder::new().insert(insert2);
let merged = cs1 | cs2;
assert_eq!(merged.len(), 2);
}
#[test]
fn test_bitor_changeset_consolidates_same_row() {
let table = TestTable::new("users", 2, 0);
let insert = Insert::from(table.clone())
.set(0, 1i64)
.unwrap()
.set(1, "alice")
.unwrap();
let delete = ChangeDelete::from(table.clone())
.set(0, 1i64)
.unwrap()
.set(1, "alice")
.unwrap();
let cs1 = ChangesetBuilder::new().insert(insert);
let cs2 = ChangesetBuilder::new().delete(delete);
let merged = cs1 | cs2;
assert_eq!(merged.len(), 0);
assert!(merged.is_empty());
}
#[test]
fn test_bitor_assign_changeset() {
let table = TestTable::new("users", 2, 0);
let insert1 = Insert::from(table.clone())
.set(0, 1i64)
.unwrap()
.set(1, "alice")
.unwrap();
let insert2 = Insert::from(table.clone())
.set(0, 2i64)
.unwrap()
.set(1, "bob")
.unwrap();
let mut cs = ChangesetBuilder::new().insert(insert1);
cs |= ChangesetBuilder::new().insert(insert2);
assert_eq!(cs.len(), 2);
}
#[test]
fn test_bitor_patchset_disjoint_rows() {
type PatchsetBuilder = DiffSetBuilder<PatchsetFormat, TestTable, String, Vec<u8>>;
let table = TestTable::new("users", 2, 0);
let insert1 = Insert::from(table.clone())
.set(0, 1i64)
.unwrap()
.set(1, "alice")
.unwrap();
let insert2 = Insert::from(table.clone())
.set(0, 2i64)
.unwrap()
.set(1, "bob")
.unwrap();
let ps1 = PatchsetBuilder::new().insert(insert1);
let ps2 = PatchsetBuilder::new().insert(insert2);
let merged = ps1 | ps2;
assert_eq!(merged.len(), 2);
}
#[test]
fn test_bitor_patchset_consolidates_same_row() {
type PatchsetBuilder = DiffSetBuilder<PatchsetFormat, TestTable, String, Vec<u8>>;
let table = TestTable::new("users", 2, 0);
let insert = Insert::from(table.clone())
.set(0, 1i64)
.unwrap()
.set(1, "alice")
.unwrap();
let delete = PatchDelete::new(table.clone(), vec![Value::Integer(1)]);
let ps1 = PatchsetBuilder::new().insert(insert);
let ps2 = PatchsetBuilder::new().delete(delete);
let merged = ps1 | ps2;
assert_eq!(merged.len(), 0);
assert!(merged.is_empty());
}
#[test]
fn test_bitor_multiple_tables() {
let table1 = TestTable::new("users", 2, 0);
let table2 = TestTable::new("posts", 2, 0);
let insert1 = Insert::from(table1.clone())
.set(0, 1i64)
.unwrap()
.set(1, "alice")
.unwrap();
let insert2 = Insert::from(table2.clone())
.set(0, 100i64)
.unwrap()
.set(1, "first post")
.unwrap();
let cs1 = ChangesetBuilder::new().insert(insert1);
let cs2 = ChangesetBuilder::new().insert(insert2);
let merged = cs1 | cs2;
assert_eq!(merged.len(), 2);
}
#[test]
fn test_bitor_insert_then_update_consolidates() {
let table = TestTable::new("users", 2, 0);
let insert = Insert::from(table.clone())
.set(0, 1i64)
.unwrap()
.set(1, "alice")
.unwrap();
let update = Update::<TestTable, ChangesetFormat, String, Vec<u8>>::from(table.clone())
.set(0, 1i64, 1i64)
.unwrap()
.set(1, "alice", "alicia")
.unwrap();
let cs1 = ChangesetBuilder::new().insert(insert);
let cs2 = ChangesetBuilder::new().update(update);
let merged = cs1 | cs2;
assert_eq!(merged.len(), 1);
}
#[test]
fn test_session_hash_growth_with_many_rows() {
let table = TestTable::new("many", 2, 0);
let mut cs = ChangesetBuilder::new();
for i in 0..200i64 {
let insert = Insert::from(table.clone())
.set(0, i)
.unwrap()
.set(1, alloc::format!("row-{i}"))
.unwrap();
cs = cs.insert(insert);
}
assert_eq!(cs.len(), 200);
let bytes = cs.build();
assert!(!bytes.is_empty());
let reparsed = crate::parser::ParsedDiffSet::try_from(bytes.as_slice()).unwrap();
let reparsed_bytes: Vec<u8> = reparsed.into();
assert_eq!(bytes, reparsed_bytes);
}
const INDIRECT_BYTE_OFFSET: usize = 7;
#[test]
fn test_build_insert_indirect_byte_set() {
let table = TestTable::new("t", 2, 0);
let insert = Insert::from(table)
.set(0, 1i64)
.unwrap()
.set(1, "a")
.unwrap()
.indirect(true);
let bytes = ChangesetBuilder::new().insert(insert).build();
assert_eq!(bytes[INDIRECT_BYTE_OFFSET], 1);
}
#[test]
fn test_build_delete_indirect_byte_set() {
let table = TestTable::new("t", 2, 0);
let delete = ChangeDelete::from(table)
.set(0, 1i64)
.unwrap()
.set(1, "a")
.unwrap()
.indirect(true);
let bytes = ChangesetBuilder::new().delete(delete).build();
assert_eq!(bytes[INDIRECT_BYTE_OFFSET], 1);
}
#[test]
fn test_build_update_indirect_byte_set() {
let table = TestTable::new("t", 2, 0);
let update = Update::<TestTable, ChangesetFormat, String, Vec<u8>>::from(table)
.set(0, 1i64, 1i64)
.unwrap()
.set(1, "a", "b")
.unwrap()
.indirect(true);
let bytes = ChangesetBuilder::new().update(update).build();
assert_eq!(bytes[INDIRECT_BYTE_OFFSET], 1);
}
#[test]
fn test_build_patchset_insert_indirect_byte() {
let table = TestTable::new("t", 2, 0);
let insert = Insert::from(table)
.set(0, 1i64)
.unwrap()
.set(1, "a")
.unwrap()
.indirect(true);
let patchset: PatchSet<TestTable, String, Vec<u8>> = PatchSet::new().insert(insert);
let bytes = patchset.build();
assert_eq!(bytes[INDIRECT_BYTE_OFFSET], 1);
}
#[test]
fn test_reverse_preserves_indirect() {
let insert: Operation<ChangesetFormat, String, Vec<u8>> = Operation::Insert {
values: vec![Value::Integer(1)],
indirect: true,
};
let reversed = insert.reverse();
assert!(reversed.indirect());
assert!(matches!(reversed, Operation::Delete { .. }));
let delete: Operation<ChangesetFormat, String, Vec<u8>> = Operation::Delete {
data: vec![Value::Integer(1)],
indirect: true,
};
let reversed = delete.reverse();
assert!(reversed.indirect());
assert!(matches!(reversed, Operation::Insert { .. }));
let update: Operation<ChangesetFormat, String, Vec<u8>> = Operation::Update {
values: vec![(Some(Value::Integer(1)), Some(Value::Integer(2)))],
indirect: true,
};
let reversed = update.reverse();
assert!(reversed.indirect());
assert!(matches!(reversed, Operation::Update { .. }));
}
#[test]
fn test_patchdelete_indirect_byte_set() {
let table = TestTable::new("t", 2, 0);
let delete: PatchDelete<TestTable, String, Vec<u8>> =
PatchDelete::new(table, vec![Value::Integer(1)]).indirect(true);
let bytes = PatchSet::new().delete(delete).build();
assert_eq!(bytes[INDIRECT_BYTE_OFFSET], 1);
}
#[test]
fn test_operation_eq_indirect_differs() {
let a: Operation<ChangesetFormat, String, Vec<u8>> = Operation::Insert {
values: vec![Value::Integer(1)],
indirect: false,
};
let b: Operation<ChangesetFormat, String, Vec<u8>> = Operation::Insert {
values: vec![Value::Integer(1)],
indirect: true,
};
assert_ne!(a, b);
}
#[test]
fn test_operation_eq_variant_mismatch() {
let insert: Operation<ChangesetFormat, String, Vec<u8>> = Operation::Insert {
values: vec![Value::Integer(1)],
indirect: false,
};
let delete: Operation<ChangesetFormat, String, Vec<u8>> = Operation::Delete {
data: vec![Value::Integer(1)],
indirect: false,
};
assert_ne!(insert, delete);
}
#[test]
fn test_bitor_indirect_rhs_wins() {
let table = TestTable::new("t", 2, 0);
let lhs_insert = Insert::from(table.clone())
.set(0, 1i64)
.unwrap()
.set(1, "a")
.unwrap()
.indirect(true);
let rhs_insert = Insert::from(table.clone())
.set(0, 1i64)
.unwrap()
.set(1, "a")
.unwrap()
.indirect(false);
let merged =
ChangesetBuilder::new().insert(lhs_insert) | ChangesetBuilder::new().insert(rhs_insert);
let rows = merged.tables.get(&table).unwrap();
let op = rows.values().next().unwrap();
assert!(!op.indirect(), "rhs (false) should win over lhs (true)");
let lhs_insert = Insert::from(table.clone())
.set(0, 2i64)
.unwrap()
.set(1, "b")
.unwrap()
.indirect(false);
let rhs_insert = Insert::from(table.clone())
.set(0, 2i64)
.unwrap()
.set(1, "b")
.unwrap()
.indirect(true);
let merged =
ChangesetBuilder::new().insert(lhs_insert) | ChangesetBuilder::new().insert(rhs_insert);
let rows = merged.tables.get(&table).unwrap();
let op = rows.values().next().unwrap();
assert!(op.indirect(), "rhs (true) should win over lhs (false)");
}
#[test]
fn test_roundtrip_indirect_changeset() {
let table = TestTable::new("t", 2, 0);
let insert = Insert::from(table)
.set(0, 1i64)
.unwrap()
.set(1, "a")
.unwrap()
.indirect(true);
let bytes = ChangesetBuilder::new().insert(insert).build();
let reparsed = crate::parser::ParsedDiffSet::try_from(bytes.as_slice()).unwrap();
let reparsed_bytes: Vec<u8> = reparsed.into();
assert_eq!(bytes, reparsed_bytes);
assert_eq!(reparsed_bytes[INDIRECT_BYTE_OFFSET], 1);
}
#[test]
fn test_roundtrip_indirect_patchset() {
let table = TestTable::new("t", 2, 0);
let insert = Insert::from(table)
.set(0, 1i64)
.unwrap()
.set(1, "a")
.unwrap()
.indirect(true);
let patchset: PatchSet<TestTable, String, Vec<u8>> = PatchSet::new().insert(insert);
let bytes = patchset.build();
let reparsed = crate::parser::ParsedDiffSet::try_from(bytes.as_slice()).unwrap();
let reparsed_bytes: Vec<u8> = reparsed.into();
assert_eq!(bytes, reparsed_bytes);
assert_eq!(reparsed_bytes[INDIRECT_BYTE_OFFSET], 1);
}
#[test]
fn test_indirect_full_pipeline_roundtrip() {
let table = TestTable::new("t", 2, 0);
let insert = Insert::from(table)
.set(0, 1i64)
.unwrap()
.set(1, "alice")
.unwrap()
.indirect(true);
let original = ChangesetBuilder::new().insert(insert);
let bytes_a = original.build();
assert_eq!(bytes_a[INDIRECT_BYTE_OFFSET], 1);
let parsed = crate::parser::ParsedDiffSet::try_from(bytes_a.as_slice()).unwrap();
let crate::parser::ParsedDiffSet::Changeset(parsed_set) = parsed else {
panic!("expected Changeset variant");
};
let parsed_builder: DiffSetBuilder<
ChangesetFormat,
crate::parser::TableSchema<String>,
String,
Vec<u8>,
> = parsed_set.into();
let empty: DiffSetBuilder<
ChangesetFormat,
crate::parser::TableSchema<String>,
String,
Vec<u8>,
> = DiffSetBuilder::new();
let doubled = parsed_builder.reverse().reverse();
let merged = doubled | empty;
let bytes_b = merged.build();
assert_eq!(bytes_a, bytes_b);
assert_eq!(bytes_b[INDIRECT_BYTE_OFFSET], 1);
}
fn changeset_insert(v: i64) -> Operation<ChangesetFormat, String, Vec<u8>> {
Operation::Insert {
values: vec![Value::Integer(v), Value::Text("a".into())],
indirect: false,
}
}
fn changeset_delete(v: i64) -> Operation<ChangesetFormat, String, Vec<u8>> {
Operation::Delete {
data: vec![Value::Integer(v), Value::Text("a".into())],
indirect: false,
}
}
fn changeset_update(old: i64, new: i64) -> Operation<ChangesetFormat, String, Vec<u8>> {
Operation::Update {
values: vec![
(Some(Value::Integer(old)), Some(Value::Integer(new))),
(Some(Value::Text("a".into())), Some(Value::Text("b".into()))),
],
indirect: false,
}
}
#[test]
fn test_add_changeset_insert_plus_update() {
let merged = (changeset_insert(1) + changeset_update(1, 2)).unwrap();
let Operation::Insert { values, .. } = merged else {
panic!("expected Insert");
};
assert_eq!(values[0], Value::Integer(2));
assert_eq!(values[1], Value::Text("b".into()));
}
#[test]
fn test_add_changeset_update_plus_insert() {
let merged = (changeset_update(1, 2) + changeset_insert(99)).unwrap();
assert!(matches!(merged, Operation::Update { .. }));
}
#[test]
fn test_add_changeset_update_plus_update() {
let merged = (changeset_update(1, 2) + changeset_update(2, 3)).unwrap();
let Operation::Update { values, .. } = merged else {
panic!("expected Update");
};
assert_eq!(values[0].0, Some(Value::Integer(1)));
assert_eq!(values[0].1, Some(Value::Integer(3)));
}
#[test]
fn test_add_changeset_update_plus_delete() {
let merged = (changeset_update(1, 2) + changeset_delete(99)).unwrap();
let Operation::Delete { data, .. } = merged else {
panic!("expected Delete");
};
assert_eq!(data[0], Value::Integer(1));
assert_eq!(data[1], Value::Text("a".into()));
}
#[test]
fn test_add_changeset_delete_plus_update_keeps_delete() {
let merged = (changeset_delete(1) + changeset_update(1, 2)).unwrap();
assert!(matches!(merged, Operation::Delete { .. }));
}
#[test]
fn test_add_changeset_delete_plus_delete_keeps_first() {
let merged = (changeset_delete(1) + changeset_delete(2)).unwrap();
let Operation::Delete { data, .. } = merged else {
panic!("expected Delete");
};
assert_eq!(data[0], Value::Integer(1));
}
fn patchset_insert(v: i64) -> Operation<PatchsetFormat, String, Vec<u8>> {
Operation::Insert {
values: vec![Value::Integer(v), Value::Text("a".into())],
indirect: false,
}
}
fn patchset_update(new: i64) -> Operation<PatchsetFormat, String, Vec<u8>> {
Operation::Update {
values: vec![
((), Some(Value::Integer(new))),
((), Some(Value::Text("b".into()))),
],
indirect: false,
}
}
fn patchset_delete() -> Operation<PatchsetFormat, String, Vec<u8>> {
Operation::Delete {
data: (),
indirect: false,
}
}
#[test]
fn test_add_patchset_insert_plus_update() {
let merged = (patchset_insert(1) + patchset_update(2)).unwrap();
let Operation::Insert { values, .. } = merged else {
panic!("expected Insert");
};
assert_eq!(values[0], Value::Integer(2));
assert_eq!(values[1], Value::Text("b".into()));
}
#[test]
fn test_add_patchset_update_plus_insert() {
let merged = (patchset_update(2) + patchset_insert(99)).unwrap();
assert!(matches!(merged, Operation::Update { .. }));
}
#[test]
fn test_add_patchset_update_plus_update() {
let merged = (patchset_update(2) + patchset_update(3)).unwrap();
let Operation::Update { values, .. } = merged else {
panic!("expected Update");
};
assert_eq!(values[0].1, Some(Value::Integer(3)));
}
#[test]
fn test_add_patchset_update_plus_delete() {
let merged = (patchset_update(2) + patchset_delete()).unwrap();
assert!(matches!(merged, Operation::Delete { .. }));
}
#[test]
fn test_add_patchset_delete_plus_insert_promotes_to_update() {
let merged = (patchset_delete() + patchset_insert(1)).unwrap();
assert!(matches!(merged, Operation::Update { .. }));
}
#[test]
fn test_add_patchset_delete_plus_update_keeps_delete() {
let merged = (patchset_delete() + patchset_update(2)).unwrap();
assert!(matches!(merged, Operation::Delete { .. }));
}
#[test]
fn test_add_patchset_delete_plus_delete_keeps_first() {
let merged = (patchset_delete() + patchset_delete()).unwrap();
assert!(matches!(merged, Operation::Delete { .. }));
}
#[test]
fn test_session_hash_real_pk() {
let table = TestTable::new("t", 2, 0);
let insert = Insert::from(table)
.set(0, 2.5f64)
.unwrap()
.set(1, "a")
.unwrap();
let bytes = ChangesetBuilder::new().insert(insert).build();
assert!(!bytes.is_empty());
}
#[test]
fn test_session_hash_text_pk() {
let table = TestTable::new("t", 2, 0);
let insert = Insert::from(table)
.set(0, "alice")
.unwrap()
.set(1, 42i64)
.unwrap();
let bytes = ChangesetBuilder::new().insert(insert).build();
assert!(!bytes.is_empty());
}
#[test]
fn test_session_hash_blob_pk() {
let table = TestTable::new("t", 2, 0);
let insert = Insert::from(table)
.set(0, alloc::vec![0xDE_u8, 0xAD, 0xBE, 0xEF])
.unwrap()
.set(1, "a")
.unwrap();
let bytes = ChangesetBuilder::new().insert(insert).build();
assert!(!bytes.is_empty());
}
#[test]
fn test_session_row_order_empty() {
let cs: ChangesetBuilder = ChangesetBuilder::new();
let bytes = cs.build();
assert!(bytes.is_empty());
}
#[test]
fn test_session_row_order_empty_rows_returns_empty_vec() {
let rows: RowMap<ChangesetFormat, String, Vec<u8>> = IndexMap::default();
assert!(session_row_order(&rows).is_empty());
}
#[test]
fn test_diffset_patchset_build_skips_empty_table() {
let table = TestTable::new("t", 2, 0);
let mut builder: PatchSet<TestTable, String, Vec<u8>> = PatchSet::new();
builder.add_table(&table);
let frozen: DiffSet<PatchsetFormat, TestTable, String, Vec<u8>> = builder.into();
let bytes = frozen.build();
assert!(bytes.is_empty());
}
#[test]
fn test_from_changeset_builder_into_vec() {
let table = TestTable::new("t", 2, 0);
let insert = Insert::from(table)
.set(0, 1i64)
.unwrap()
.set(1, "a")
.unwrap();
let builder = ChangesetBuilder::new().insert(insert);
let bytes_owned: Vec<u8> = builder.clone().into();
let bytes_ref: Vec<u8> = (&builder).into();
assert_eq!(bytes_owned, bytes_ref);
let frozen: DiffSet<ChangesetFormat, TestTable, String, Vec<u8>> = builder.into();
let bytes_frozen_owned: Vec<u8> = frozen.clone().into();
let bytes_frozen_ref: Vec<u8> = (&frozen).into();
assert_eq!(bytes_frozen_owned, bytes_frozen_ref);
}
#[test]
fn test_from_patchset_builder_into_vec() {
let table = TestTable::new("t", 2, 0);
let insert = Insert::from(table)
.set(0, 1i64)
.unwrap()
.set(1, "a")
.unwrap();
let builder: PatchSet<TestTable, String, Vec<u8>> = PatchSet::new().insert(insert);
let bytes_owned: Vec<u8> = builder.clone().into();
let bytes_ref: Vec<u8> = (&builder).into();
assert_eq!(bytes_owned, bytes_ref);
let frozen: DiffSet<PatchsetFormat, TestTable, String, Vec<u8>> = builder.into();
let bytes_frozen_owned: Vec<u8> = frozen.clone().into();
let bytes_frozen_ref: Vec<u8> = (&frozen).into();
assert_eq!(bytes_frozen_owned, bytes_frozen_ref);
}
#[test]
fn test_add_operation_insert_then_update_changes_pk() {
let table = TestTable::new("t", 2, 0);
let insert = Insert::from(table.clone())
.set(0, 1i64)
.unwrap()
.set(1, "alice")
.unwrap();
let update = Update::<TestTable, ChangesetFormat, String, Vec<u8>>::from(table.clone())
.set(0, 1i64, 2i64)
.unwrap()
.set(1, "alice", "bob")
.unwrap();
let builder = ChangesetBuilder::new().insert(insert).update(update);
let rows = builder.tables.get(&table).unwrap();
assert_eq!(rows.len(), 1);
let (pk, op) = rows.iter().next().unwrap();
assert_eq!(pk[0], Value::Integer(2));
let Operation::Insert { values, .. } = op else {
panic!("expected merged INSERT");
};
assert_eq!(values[0], Value::Integer(2));
assert_eq!(values[1], Value::Text("bob".into()));
}
#[test]
fn test_diffset_changeset_diffops_wrappers() {
let table = TestTable::new("t", 2, 0);
let initial = Insert::from(table.clone())
.set(0, 1i64)
.unwrap()
.set(1, "a")
.unwrap();
let frozen: DiffSet<ChangesetFormat, TestTable, String, Vec<u8>> =
ChangesetBuilder::new().insert(initial).into();
let insert2 = Insert::from(table.clone())
.set(0, 2i64)
.unwrap()
.set(1, "b")
.unwrap();
let after_insert = <_ as DiffOps<_, _, _>>::insert(frozen.clone(), insert2);
assert_eq!(after_insert.len(), 2);
let delete = ChangeDelete::from(table.clone())
.set(0, 1i64)
.unwrap()
.set(1, "a")
.unwrap();
let after_delete = <_ as DiffOps<_, _, _>>::delete(frozen.clone(), delete);
assert_eq!(after_delete.len(), 0);
let update = Update::<TestTable, ChangesetFormat, String, Vec<u8>>::from(table)
.set(0, 1i64, 1i64)
.unwrap()
.set(1, "a", "z")
.unwrap();
let after_update = <_ as DiffOps<_, _, _>>::update(frozen, update);
assert_eq!(after_update.len(), 1);
}
#[test]
fn test_diffset_patchset_diffops_wrappers() {
let table = TestTable::new("t", 2, 0);
let initial = Insert::from(table.clone())
.set(0, 1i64)
.unwrap()
.set(1, "a")
.unwrap();
let frozen: DiffSet<PatchsetFormat, TestTable, String, Vec<u8>> =
PatchSet::new().insert(initial).into();
let insert2 = Insert::from(table.clone())
.set(0, 2i64)
.unwrap()
.set(1, "b")
.unwrap();
let after_insert = <_ as DiffOps<_, _, _>>::insert(frozen.clone(), insert2);
assert_eq!(after_insert.len(), 2);
let delete: PatchDelete<TestTable, String, Vec<u8>> =
PatchDelete::new(table.clone(), vec![Value::Integer(1)]);
let after_delete = <_ as DiffOps<_, _, _>>::delete(frozen.clone(), delete);
assert_eq!(after_delete.len(), 0);
let update = Update::<TestTable, PatchsetFormat, String, Vec<u8>>::from(table)
.set(0, 1i64)
.unwrap()
.set(1, "z")
.unwrap();
let after_update = <_ as DiffOps<_, _, _>>::update(frozen, update);
assert_eq!(after_update.len(), 1);
}
#[test]
fn test_diffset_changeset_iter_yields_inserts_and_indirect_flag() {
let table = TestTable::new("t", 2, 0);
let direct = Insert::from(table.clone()).set(0, 1i64).unwrap();
let indirect = Insert::from(table.clone())
.set(0, 2i64)
.unwrap()
.indirect(true);
let frozen: DiffSet<ChangesetFormat, TestTable, String, Vec<u8>> = ChangesetBuilder::new()
.insert(direct)
.insert(indirect)
.into();
let ops: Vec<_> = frozen.iter().collect();
assert_eq!(ops.len(), 2);
for op in &ops {
assert_eq!(crate::DynTable::name(op.table()), "t");
}
assert!(matches!(
ops[0],
ChangesetOp::Insert {
indirect: false,
..
}
));
assert!(matches!(ops[1], ChangesetOp::Insert { indirect: true, .. }));
}
#[test]
fn test_diffset_changeset_iter_yields_update_and_delete() {
let table = TestTable::new("t", 2, 0);
let starting = Insert::from(table.clone())
.set(0, 1i64)
.unwrap()
.set(1, "a")
.unwrap();
let update = Update::<TestTable, ChangesetFormat, String, Vec<u8>>::from(table.clone())
.set(0, 2i64, 2i64)
.unwrap()
.set(1, "before", "after")
.unwrap();
let delete = ChangeDelete::from(table.clone())
.set(0, 3i64)
.unwrap()
.set(1, "gone")
.unwrap();
let frozen: DiffSet<ChangesetFormat, TestTable, String, Vec<u8>> = ChangesetBuilder::new()
.insert(starting)
.update(update)
.delete(delete)
.into();
let kinds: Vec<&'static str> = frozen
.iter()
.map(|op| match op {
ChangesetOp::Insert { .. } => "insert",
ChangesetOp::Update { .. } => "update",
ChangesetOp::Delete { .. } => "delete",
})
.collect();
assert_eq!(kinds, ["insert", "update", "delete"]);
}
#[test]
fn test_diffset_patchset_iter_exposes_pk_for_delete_and_update() {
let table = TestTable::new("t", 2, 0);
let starting = Insert::from(table.clone())
.set(0, 1i64)
.unwrap()
.set(1, "a")
.unwrap();
let update = Update::<TestTable, PatchsetFormat, String, Vec<u8>>::from(table.clone())
.set(0, 5i64)
.unwrap()
.set(1, "z")
.unwrap();
let delete: PatchDelete<TestTable, String, Vec<u8>> =
PatchDelete::new(table.clone(), vec![Value::Integer(7)]);
let frozen: DiffSet<PatchsetFormat, TestTable, String, Vec<u8>> = PatchSet::new()
.insert(starting)
.update(update)
.delete(delete)
.into();
let mut saw_insert = false;
let mut saw_update_pk: Option<i64> = None;
let mut saw_delete_pk: Option<i64> = None;
for op in frozen.iter() {
match op {
PatchsetOp::Insert { values, .. } => {
saw_insert = true;
assert!(matches!(values[0], Value::Integer(1)));
}
PatchsetOp::Update { pk, .. } => {
if let Value::Integer(id) = pk[0] {
saw_update_pk = Some(id);
}
}
PatchsetOp::Delete { pk, .. } => {
if let Value::Integer(id) = pk[0] {
saw_delete_pk = Some(id);
}
}
}
}
assert!(saw_insert);
assert_eq!(saw_update_pk, Some(5));
assert_eq!(saw_delete_pk, Some(7));
}
#[test]
fn test_diffset_tables_skips_empty() {
let t1 = TestTable::new("t1", 2, 0);
let t2 = TestTable::new("t2", 2, 0);
let insert_t1 = Insert::from(t1.clone()).set(0, 1i64).unwrap();
let insert_then_delete_t2 = Insert::from(t2.clone()).set(0, 9i64).unwrap();
let delete_t2 = ChangeDelete::from(t2.clone()).set(0, 9i64).unwrap();
let frozen: DiffSet<ChangesetFormat, TestTable, String, Vec<u8>> = ChangesetBuilder::new()
.insert(insert_t1)
.insert(insert_then_delete_t2)
.delete(delete_t2)
.into();
let names: Vec<&str> = frozen.tables().map(crate::DynTable::name).collect();
assert_eq!(names, ["t1"]);
}
}