Struct WriteBatch 

pub struct WriteBatch { /* private fields */ }

Atomic group of writes that may span multiple buckets.

Methods without a bucket suffix target the built-in default bucket. Methods ending in _bucket target an optional named bucket returned by Db::bucket.

A batch is committed with crate::Db::write_sync or crate::Db::write. Trine assigns one commit sequence to the entire batch, appends the accepted operations to the WAL for persistent databases, and publishes the batch to the affected memtables atomically from the caller’s point of view.

Examples

use trine_kv::{Db, WriteBatch, WriteOptions};

let db = Db::open_sync(trine_kv::DbOptions::memory())?;
let users = db.bucket_sync("users")?;

let mut batch = WriteBatch::new();
batch.put(b"system:ready", b"yes");
batch.put_bucket(users.name().as_str(), b"1", b"Ada")?;

let commit = db.write_sync(batch, WriteOptions::sync_all())?;
assert!(commit.sequence().get() > 0);

Implementations

impl WriteBatch

pub const fn new() -> Self

Creates an empty batch.

The batch does not reserve a commit sequence until it is passed to a database write method.

Examples found in repository

examples/event_index.rs (line 58)

    fn append(&self, event: &Event) -> Result<()> {
        let mut batch = WriteBatch::new();
        batch.put_bucket("events", event_key(&event.id), event.encode()?)?;
        batch.put_bucket(
            "events_by_account",
            account_event_key(&event.account_id, &event.id),
            event.id.as_bytes(),
        )?;
        self.db.write_sync(batch, WriteOptions::default())?;
        Ok(())
    }

examples/sync_quickstart.rs (line 15)

fn main() -> trine_kv::Result<()> {
    let path =
        std::env::temp_dir().join(format!("trine-kv-sync-quickstart-{}", std::process::id()));
    if path.exists() {
        std::fs::remove_dir_all(&path)?;
    }

    let db = Db::open_sync(&path)?;
    let users = db.bucket_sync("users")?;

    users.put_sync(b"user:001", b"Ada")?;

    let mut batch = WriteBatch::new();
    batch.put_bucket("users", b"user:002", b"Lin")?;
    batch.put_bucket("users", b"team:core", b"database")?;
    db.write_sync(batch, WriteOptions::default())?;

    assert_eq!(users.get_sync(b"user:001")?, Some(b"Ada".to_vec()));

    let snapshot = db.snapshot();
    users.put_sync(b"user:003", b"Grace")?;
    assert_eq!(snapshot.get_sync(&users, b"user:003")?, None);
    assert_eq!(users.get_sync(b"user:003")?, Some(b"Grace".to_vec()));

    let user_prefix_keys = users
        .prefix_sync(b"user:")?
        .map(|item| item.map(|key_value| display_key(&key_value.key)))
        .collect::<trine_kv::Result<Vec<_>>>()?;
    assert_eq!(user_prefix_keys, ["user:001", "user:002", "user:003"]);

    let range = KeyRange::half_open(b"user:001", b"user:004");
    let range_values = users
        .range_sync(&range)?
        .map(|item| item.map(|key_value| display_value(&key_value.value)))
        .collect::<trine_kv::Result<Vec<_>>>()?;
    assert_eq!(range_values, ["Ada", "Lin", "Grace"]);

    let mut txn = db.transaction(TransactionOptions::default());
    assert_eq!(
        txn.get_bucket_sync("users", b"user:001")?,
        Some(b"Ada".to_vec())
    );
    txn.put_bucket("users", b"user:004", b"Barbara")?;
    txn.commit_sync()?;

    db.flush_sync()?;
    drop(users);
    drop(snapshot);
    drop(db);

    let reopened = Db::open_sync(&path)?;
    let users = reopened.bucket_sync("users")?;
    assert_eq!(users.get_sync(b"user:004")?, Some(b"Barbara".to_vec()));

    let stats = reopened.stats();
    assert_eq!(stats.live_buckets, 2);
    assert!(stats.total_tables > 0);

    drop(users);
    drop(reopened);
    std::fs::remove_dir_all(path)?;
    Ok(())
}

examples/quickstart.rs (line 31)

async fn run(path: &Path) -> Result<()> {
    let mut options = DbOptions::new(path);
    options.background_worker_count = 0;

    let db = Db::open(options).await?;
    let users = db.bucket("users").await?;

    users.put(b"user:001".to_vec(), b"Ada".to_vec()).await?;

    let mut batch = WriteBatch::new();
    batch.put_bucket("users", b"user:002".to_vec(), b"Lin".to_vec())?;
    batch.put_bucket("users", b"team:core".to_vec(), b"database".to_vec())?;
    db.write(batch, WriteOptions::default()).await?;

    assert_eq!(users.get(b"user:001").await?, Some(b"Ada".to_vec()));

    let snapshot = db.snapshot();
    users.put(b"user:003".to_vec(), b"Grace".to_vec()).await?;
    assert_eq!(
        users.get_at(&snapshot, b"user:003").await?,
        None,
        "snapshot reads stay pinned to their original sequence",
    );
    assert_eq!(users.get(b"user:003").await?, Some(b"Grace".to_vec()));

    let prefix_rows = collect_lazy_rows(users.prefix_lazy(b"user:".to_vec()).await?).await?;
    assert_eq!(
        prefix_rows,
        [
            ("user:001".to_owned(), "Ada".to_owned()),
            ("user:002".to_owned(), "Lin".to_owned()),
            ("user:003".to_owned(), "Grace".to_owned()),
        ],
    );

    let mut transaction = db.transaction(TransactionOptions::default());
    assert_eq!(
        transaction.get_bucket("users", b"user:001").await?,
        Some(b"Ada".to_vec()),
    );
    transaction
        .read_range_bucket("users", KeyRange::half_open(b"user:001", b"user:004"))
        .await?;
    transaction.put_bucket("users", b"user:004".to_vec(), b"Barbara".to_vec())?;
    transaction.commit().await?;

    db.flush().await?;
    drop(users);
    drop(snapshot);
    db.close().await?;

    let reopened = Db::open(DbOptions::new(path).read_only()).await?;
    let users = reopened.bucket("users").await?;
    assert_eq!(users.get(b"user:004").await?, Some(b"Barbara".to_vec()));

    let stats = reopened.stats();
    assert_eq!(stats.live_buckets, 2);
    assert!(stats.total_tables > 0);
    assert!(stats.storage_uses_sync_adapter);
    assert!(!stats.storage_uses_platform_async_io);

    drop(users);
    reopened.close().await
}

pub fn put(&mut self, key: impl Into<Vec<u8>>, value: impl Into<Value>)

Adds a key/value write to the default bucket.

Parameters

• key: user key bytes for the built-in default bucket.
• value: value bytes to store.

pub fn put_bucket( &mut self, bucket: impl Into<String>, key: impl Into<Vec<u8>>, value: impl Into<Value>, ) -> Result<()>

Adds a key/value write for a named bucket.

The bucket name must refer to an optional named bucket. Use WriteBatch::put for the built-in default bucket.

Parameters

• bucket: target named bucket.
• key: user key bytes.
• value: value bytes to store.

Errors

Returns Error::InvalidOptions if bucket is empty or is the reserved default bucket name.

Examples found in repository

examples/event_index.rs (line 59)

    fn append(&self, event: &Event) -> Result<()> {
        let mut batch = WriteBatch::new();
        batch.put_bucket("events", event_key(&event.id), event.encode()?)?;
        batch.put_bucket(
            "events_by_account",
            account_event_key(&event.account_id, &event.id),
            event.id.as_bytes(),
        )?;
        self.db.write_sync(batch, WriteOptions::default())?;
        Ok(())
    }

examples/sync_quickstart.rs (line 16)

fn main() -> trine_kv::Result<()> {
    let path =
        std::env::temp_dir().join(format!("trine-kv-sync-quickstart-{}", std::process::id()));
    if path.exists() {
        std::fs::remove_dir_all(&path)?;
    }

    let db = Db::open_sync(&path)?;
    let users = db.bucket_sync("users")?;

    users.put_sync(b"user:001", b"Ada")?;

    let mut batch = WriteBatch::new();
    batch.put_bucket("users", b"user:002", b"Lin")?;
    batch.put_bucket("users", b"team:core", b"database")?;
    db.write_sync(batch, WriteOptions::default())?;

    assert_eq!(users.get_sync(b"user:001")?, Some(b"Ada".to_vec()));

    let snapshot = db.snapshot();
    users.put_sync(b"user:003", b"Grace")?;
    assert_eq!(snapshot.get_sync(&users, b"user:003")?, None);
    assert_eq!(users.get_sync(b"user:003")?, Some(b"Grace".to_vec()));

    let user_prefix_keys = users
        .prefix_sync(b"user:")?
        .map(|item| item.map(|key_value| display_key(&key_value.key)))
        .collect::<trine_kv::Result<Vec<_>>>()?;
    assert_eq!(user_prefix_keys, ["user:001", "user:002", "user:003"]);

    let range = KeyRange::half_open(b"user:001", b"user:004");
    let range_values = users
        .range_sync(&range)?
        .map(|item| item.map(|key_value| display_value(&key_value.value)))
        .collect::<trine_kv::Result<Vec<_>>>()?;
    assert_eq!(range_values, ["Ada", "Lin", "Grace"]);

    let mut txn = db.transaction(TransactionOptions::default());
    assert_eq!(
        txn.get_bucket_sync("users", b"user:001")?,
        Some(b"Ada".to_vec())
    );
    txn.put_bucket("users", b"user:004", b"Barbara")?;
    txn.commit_sync()?;

    db.flush_sync()?;
    drop(users);
    drop(snapshot);
    drop(db);

    let reopened = Db::open_sync(&path)?;
    let users = reopened.bucket_sync("users")?;
    assert_eq!(users.get_sync(b"user:004")?, Some(b"Barbara".to_vec()));

    let stats = reopened.stats();
    assert_eq!(stats.live_buckets, 2);
    assert!(stats.total_tables > 0);

    drop(users);
    drop(reopened);
    std::fs::remove_dir_all(path)?;
    Ok(())
}

examples/quickstart.rs (line 32)

async fn run(path: &Path) -> Result<()> {
    let mut options = DbOptions::new(path);
    options.background_worker_count = 0;

    let db = Db::open(options).await?;
    let users = db.bucket("users").await?;

    users.put(b"user:001".to_vec(), b"Ada".to_vec()).await?;

    let mut batch = WriteBatch::new();
    batch.put_bucket("users", b"user:002".to_vec(), b"Lin".to_vec())?;
    batch.put_bucket("users", b"team:core".to_vec(), b"database".to_vec())?;
    db.write(batch, WriteOptions::default()).await?;

    assert_eq!(users.get(b"user:001").await?, Some(b"Ada".to_vec()));

    let snapshot = db.snapshot();
    users.put(b"user:003".to_vec(), b"Grace".to_vec()).await?;
    assert_eq!(
        users.get_at(&snapshot, b"user:003").await?,
        None,
        "snapshot reads stay pinned to their original sequence",
    );
    assert_eq!(users.get(b"user:003").await?, Some(b"Grace".to_vec()));

    let prefix_rows = collect_lazy_rows(users.prefix_lazy(b"user:".to_vec()).await?).await?;
    assert_eq!(
        prefix_rows,
        [
            ("user:001".to_owned(), "Ada".to_owned()),
            ("user:002".to_owned(), "Lin".to_owned()),
            ("user:003".to_owned(), "Grace".to_owned()),
        ],
    );

    let mut transaction = db.transaction(TransactionOptions::default());
    assert_eq!(
        transaction.get_bucket("users", b"user:001").await?,
        Some(b"Ada".to_vec()),
    );
    transaction
        .read_range_bucket("users", KeyRange::half_open(b"user:001", b"user:004"))
        .await?;
    transaction.put_bucket("users", b"user:004".to_vec(), b"Barbara".to_vec())?;
    transaction.commit().await?;

    db.flush().await?;
    drop(users);
    drop(snapshot);
    db.close().await?;

    let reopened = Db::open(DbOptions::new(path).read_only()).await?;
    let users = reopened.bucket("users").await?;
    assert_eq!(users.get(b"user:004").await?, Some(b"Barbara".to_vec()));

    let stats = reopened.stats();
    assert_eq!(stats.live_buckets, 2);
    assert!(stats.total_tables > 0);
    assert!(stats.storage_uses_sync_adapter);
    assert!(!stats.storage_uses_platform_async_io);

    drop(users);
    reopened.close().await
}

pub fn delete(&mut self, key: impl Into<Vec<u8>>)

Adds a point delete to the default bucket.

The delete hides older values for the same key after the batch commits. Snapshots older than the commit sequence can still see earlier values.

pub fn delete_bucket( &mut self, bucket: impl Into<String>, key: impl Into<Vec<u8>>, ) -> Result<()>

Adds a point delete for a named bucket.

pub fn delete_range(&mut self, range: KeyRange)

Adds a range delete to the default bucket.

The delete hides all keys in range for read sequences after the batch commits. The operation is stored as a range tombstone and can conflict with optimistic transactions that read overlapping keys or ranges.

pub fn delete_range_bucket( &mut self, bucket: impl Into<String>, range: KeyRange, ) -> Result<()>

Adds a range delete for a named bucket.

pub fn operations(&self) -> &[BatchOperation]

Returns the operations in insertion order.

pub fn into_operations(self) -> Vec<BatchOperation>

Consumes the batch and returns its operations in insertion order.

pub fn len(&self) -> usize

Returns the number of operations in the batch.

pub fn is_empty(&self) -> bool

Returns true when the batch contains no operations.

Trait Implementations

impl Clone for WriteBatch

fn clone(&self) -> WriteBatch

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for WriteBatch

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl Default for WriteBatch

fn default() -> WriteBatch

Returns the “default value” for a type.

impl Eq for WriteBatch

impl PartialEq for WriteBatch

fn eq(&self, other: &WriteBatch) -> bool

Tests for self and other values to be equal, and is used by ==.

fn ne(&self, other: &Rhs) -> bool

Tests for !=. The default implementation is almost always sufficient, and should not be overridden without very good reason.

impl StructuralPartialEq for WriteBatch