Struct Transactor

pub struct Transactor {
    cads_token: u64,
    shard_id: ShardId,
    client: PersistClient,
    since: SinceHandle<MaelstromKey, MaelstromVal, u64, i64, u64>,
    write: WriteHandle<MaelstromKey, MaelstromVal, u64, i64>,
    read_ts: u64,
    long_lived_updates: Vec<((Result<MaelstromKey, String>, Result<MaelstromVal, String>), u64, i64)>,
    long_lived_listen: Listen<MaelstromKey, MaelstromVal, u64, i64>,
}

An implementation of read-write transactions on top of persist

This executes Maelstrom txn-list-append transactions. The timestamp selection is modeled after Materialize’s SQL implementation.

A persist shard is maintained that directly represents a key-value map as it evolves over time. (Our real SQL implementation would likely instead maintain a WAL of updates to multiple tables.) Each transaction has associated timestamps:

• read_ts: at which the contents of the map are read (inclusive)
• write_ts: at which txn mutations (if any) are written
• expected_upper: the upper bound of the previous txn
• new_upper: the upper bound after this transaction commits, if it does

To keep things simple, write_ts is always read_ts+1, expected_upper is antichain[write_ts], and new_upper is antichain[write_ts+1].

Transactions are “committed” by using [WriteHandle::compare_and_append], which atomically:

• verifies that expected_upper matches
• writes the updates
• advances the upper to new_upper.

This guarantees that all transactions are linearized with each txn’s read_ts equal to the previous txn’s write_ts. If two transactions race by reading at the same read_ts and writing at the same write_ts, the compare_and_append guarantees that only one of them succeeds and the other must retry with new timestamps.

Upon first use, the persist shard is initialized by advancing the upper to antichain[T::minimum() + 1]. This allows the first txn to use 0 as the read_ts and 1 as the write_ts.

Similarly modeling Materialize, the since of the shard is kept following the upper. To keep things simple, this is done by a fixed offset. This exercises persist compaction.

To ensure that both mz_persist_client::read::ReadHandle::snapshot and mz_persist_client::read::ReadHandle::listen are exercised, when a txn reads the state at read_ts, it artificially picks an as_of timestamp in [since, read_ts] and splits up the read data between snapshot and listen along this timestamp.

Fields

cads_token: u64

shard_id: ShardId

client: PersistClient

since: SinceHandle<MaelstromKey, MaelstromVal, u64, i64, u64>

write: WriteHandle<MaelstromKey, MaelstromVal, u64, i64>

read_ts: u64

long_lived_updates: Vec<((Result<MaelstromKey, String>, Result<MaelstromVal, String>), u64, i64)>

long_lived_listen: Listen<MaelstromKey, MaelstromVal, u64, i64>

Implementations

impl Transactor

pub async fn new( client: &PersistClient, node_id: NodeId, shard_id: ShardId, ) -> Result<Self, MaelstromError>

async fn maybe_init_shard( write: &mut WriteHandle<MaelstromKey, MaelstromVal, u64, i64>, ) -> Result<u64, MaelstromError>

Initializes the shard, if it hasn’t been already, and returns the read timestamp.

pub async fn transact( &mut self, req_ops: &[ReqTxnOp], ) -> Result<Vec<ResTxnOp>, MaelstromError>

async fn read_short_lived( &mut self, ) -> Result<(Vec<((Result<MaelstromKey, String>, Result<MaelstromVal, String>), u64, i64)>, Antichain<u64>), MaelstromError>

async fn read( &mut self, ) -> Result<BTreeMap<MaelstromKey, MaelstromVal>, MaelstromError>

async fn listen_through( listen: Listen<MaelstromKey, MaelstromVal, u64, i64>, frontier: &Antichain<u64>, ) -> Result<Vec<((Result<MaelstromKey, String>, Result<MaelstromVal, String>), u64, i64)>, ExternalError>

async fn read_long_lived( &mut self, as_of: &Antichain<u64>, ) -> Vec<((Result<MaelstromKey, String>, Result<MaelstromVal, String>), u64, i64)>

fn extract_state_map( read_ts: u64, updates: Vec<((Result<MaelstromKey, String>, Result<MaelstromVal, String>), u64, i64)>, ) -> Result<BTreeMap<MaelstromKey, MaelstromVal>, MaelstromError>

fn eval_txn( state: &BTreeMap<MaelstromKey, MaelstromVal>, req_ops: &[ReqTxnOp], ) -> (Vec<(MaelstromKey, MaelstromVal, i64)>, Vec<ResTxnOp>)

async fn advance_since(&mut self) -> Result<(), MaelstromError>

fn extract_ts<T: TotalOrder + Copy>( frontier: &Antichain<T>, ) -> Result<T, MaelstromError>

Trait Implementations

impl Debug for Transactor

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

Formats the value using the given formatter.