Struct mz_stash::Postgres

pub struct Postgres {
    url: String,
    schema: Option<String>,
    tls: MakeTlsConnector,
    client: Option<Client>,
    epoch: i64,
    nonce: [u8; 16],
}

A Stash whose data is stored in a Postgres database. The format of the tables are not specified and should not be relied upon. The only promise is stability. Any changes to the table schemas will be accompanied by a clear

Fields

url: String

schema: Option<String>

tls: MakeTlsConnector

client: Option<Client>

epoch: i64

nonce: [u8; 16]

Implementations

impl Postgres

pub async fn new(
    url: String,
    schema: Option<String>,
    tls: MakeTlsConnector
) -> Result<Postgres, StashError>

Opens the stash stored at the specified path.

async fn connect(&mut self) -> Result<(), StashError>

Sets client to a new connection to the Postgres server.

async fn transact<F, T>(&mut self, f: F) -> Result<T, StashError> where
    F: for<'a> Fn(&'a mut Transaction<'_>) -> BoxFuture<'a, Result<T, StashError>>, 

Construct a fenced transaction, which will cause this Stash to fail if another connection is opened to it. f may be called multiple times in a backoff-retry loop if the Postgres server is unavailable, so it should only call functions on its Transaction argument.

Examples

async fn x(&mut self) -> Result<(), StashError> {
    self.transact(move |tx| {
        Box::pin(async move {
            // Use tx.
        })
    })
    .await

async fn transact_inner<F, T>(&mut self, f: &F) -> Result<T, StashError> where
    F: for<'a> Fn(&'a mut Transaction<'_>) -> BoxFuture<'a, Result<T, StashError>>, 

async fn since_tx(
    tx: &mut Transaction<'_>,
    collection_id: Id
) -> Result<Antichain<Timestamp>, StashError>

async fn upper_tx(
    tx: &mut Transaction<'_>,
    collection_id: Id
) -> Result<Antichain<Timestamp>, StashError>

async fn seal_batch_tx<'a, I>(
    tx: &mut Transaction<'_>,
    seals: I
) -> Result<(), StashError> where
    I: Iterator<Item = (Id, &'a Antichain<Timestamp>)>, 

async fn update_many_tx<I>(
    tx: &mut Transaction<'_>,
    collection_id: Id,
    entries: I
) -> Result<(), StashError> where
    I: Iterator<Item = ((Vec<u8>, Vec<u8>), Timestamp, Diff)>, 

async fn compact_batch_tx<'a, I>(
    tx: &mut Transaction<'_>,
    compactions: I
) -> Result<(), StashError> where
    I: Iterator<Item = (Id, &'a Antichain<Timestamp>)>, 

async fn consolidate_batch_tx<I>(
    tx: &mut Transaction<'_>,
    collections: I
) -> Result<(), StashError> where
    I: Iterator<Item = Id>, 

Trait Implementations

impl Append for Postgres

fn append<'life0, 'async_trait, I>(
    &'life0 mut self,
    batches: I
) -> Pin<Box<dyn Future<Output = Result<(), StashError>> + Send + 'async_trait>> where
    I: IntoIterator<Item = AppendBatch> + Send + 'static,
    I::IntoIter: Send,
    I: 'async_trait,
    'life0: 'async_trait,
    Self: 'async_trait, 

Atomically adds entries, seals, and compacts multiple collections.

impl Debug for Postgres

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

Formats the value using the given formatter.

impl Stash for Postgres

fn since<'life0, 'async_trait, K, V>(
    &'life0 mut self,
    collection: StashCollection<K, V>
) -> Pin<Box<dyn Future<Output = Result<Antichain<Timestamp>, StashError>> + Send + 'async_trait>> where
    K: Data,
    V: Data,
    K: 'async_trait,
    V: 'async_trait,
    'life0: 'async_trait,
    Self: 'async_trait, 

Reports the current since frontier.

fn upper<'life0, 'async_trait, K, V>(
    &'life0 mut self,
    collection: StashCollection<K, V>
) -> Pin<Box<dyn Future<Output = Result<Antichain<Timestamp>, StashError>> + Send + 'async_trait>> where
    K: Data,
    V: Data,
    K: 'async_trait,
    V: 'async_trait,
    'life0: 'async_trait,
    Self: 'async_trait, 

Reports the current upper frontier.

fn collection<'life0, 'life1, 'async_trait, K, V>(
    &'life0 mut self,
    name: &'life1 str
) -> Pin<Box<dyn Future<Output = Result<StashCollection<K, V>, StashError>> + Send + 'async_trait>> where
    K: Data,
    V: Data,
    K: 'async_trait,
    V: 'async_trait,
    'life0: 'async_trait,
    'life1: 'async_trait,
    Self: 'async_trait, 

Loads or creates the named collection.

fn iter<'life0, 'async_trait, K, V>(
    &'life0 mut self,
    collection: StashCollection<K, V>
) -> Pin<Box<dyn Future<Output = Result<Vec<((K, V), Timestamp, Diff)>, StashError>> + Send + 'async_trait>> where
    K: Data,
    V: Data,
    K: 'async_trait,
    V: 'async_trait,
    'life0: 'async_trait,
    Self: 'async_trait, 

Iterates over all entries in the stash.

fn iter_key<'life0, 'life1, 'async_trait, K, V>(
    &'life0 mut self,
    collection: StashCollection<K, V>,
    key: &'life1 K
) -> Pin<Box<dyn Future<Output = Result<Vec<(V, Timestamp, Diff)>, StashError>> + Send + 'async_trait>> where
    K: Data,
    V: Data,
    K: 'async_trait,
    V: 'async_trait,
    'life0: 'async_trait,
    'life1: 'async_trait,
    Self: 'async_trait, 

Iterates over entries in the stash for the given key.

fn update_many<'life0, 'async_trait, K, V, I>(
    &'life0 mut self,
    collection: StashCollection<K, V>,
    entries: I
) -> Pin<Box<dyn Future<Output = Result<(), StashError>> + Send + 'async_trait>> where
    K: Data,
    V: Data,
    I: IntoIterator<Item = ((K, V), Timestamp, Diff)> + Send,
    I::IntoIter: Send,
    K: 'async_trait,
    V: 'async_trait,
    I: 'async_trait,
    'life0: 'async_trait,
    Self: 'async_trait, 

Atomically adds multiple entries to the arrangement.

fn seal<'life0, 'life1, 'async_trait, K, V>(
    &'life0 mut self,
    collection: StashCollection<K, V>,
    new_upper: AntichainRef<'life1, Timestamp>
) -> Pin<Box<dyn Future<Output = Result<(), StashError>> + Send + 'async_trait>> where
    K: Data,
    V: Data,
    K: 'async_trait,
    V: 'async_trait,
    'life0: 'async_trait,
    'life1: 'async_trait,
    Self: 'async_trait, 

Atomically advances the upper frontier to the specified value.

fn seal_batch<'life0, 'life1, 'async_trait, K, V>(
    &'life0 mut self,
    seals: &'life1 [(StashCollection<K, V>, Antichain<Timestamp>)]
) -> Pin<Box<dyn Future<Output = Result<(), StashError>> + Send + 'async_trait>> where
    K: Data,
    V: Data,
    K: 'async_trait,
    V: 'async_trait,
    'life0: 'async_trait,
    'life1: 'async_trait,
    Self: 'async_trait, 

Performs multiple seals at once, potentially in a more performant way than performing the individual seals one by one.

fn compact<'a, 'async_trait, K, V>(
    &'a mut self,
    collection: StashCollection<K, V>,
    new_since: AntichainRef<'a, Timestamp>
) -> Pin<Box<dyn Future<Output = Result<(), StashError>> + Send + 'async_trait>> where
    K: Data,
    V: Data,
    'a: 'async_trait,
    K: 'async_trait,
    V: 'async_trait,
    Self: 'async_trait, 

Atomically advances the since frontier to the specified value.

fn compact_batch<'life0, 'life1, 'async_trait, K, V>(
    &'life0 mut self,
    compactions: &'life1 [(StashCollection<K, V>, Antichain<Timestamp>)]
) -> Pin<Box<dyn Future<Output = Result<(), StashError>> + Send + 'async_trait>> where
    K: Data,
    V: Data,
    K: 'async_trait,
    V: 'async_trait,
    'life0: 'async_trait,
    'life1: 'async_trait,
    Self: 'async_trait, 

Performs multiple compactions at once, potentially in a more performant way than performing the individual compactions one by one.

fn consolidate<'a, 'async_trait, K, V>(
    &'a mut self,
    collection: StashCollection<K, V>
) -> Pin<Box<dyn Future<Output = Result<(), StashError>> + Send + 'async_trait>> where
    K: Data,
    V: Data,
    'a: 'async_trait,
    K: 'async_trait,
    V: 'async_trait,
    Self: 'async_trait, 

Atomically consolidates entries less than the since frontier.

fn consolidate_batch<'life0, 'life1, 'async_trait, K, V>(
    &'life0 mut self,
    collections: &'life1 [StashCollection<K, V>]
) -> Pin<Box<dyn Future<Output = Result<(), StashError>> + Send + 'async_trait>> where
    K: Data,
    V: Data,
    K: 'async_trait,
    V: 'async_trait,
    'life0: 'async_trait,
    'life1: 'async_trait,
    Self: 'async_trait, 

Performs multiple consolidations at once, potentially in a more performant way than performing the individual consolidations one by one.

fn peek_timestamp<'life0, 'async_trait, K, V>(
    &'life0 mut self,
    collection: StashCollection<K, V>
) -> Pin<Box<dyn Future<Output = Result<Timestamp, StashError>> + Send + 'async_trait>> where
    K: Data,
    V: Data,
    K: 'async_trait,
    V: 'async_trait,
    'life0: 'async_trait,
    Self: 'async_trait, 

Returns the most recent timestamp at which sealed entries can be read.

fn peek<'life0, 'async_trait, K, V>(
    &'life0 mut self,
    collection: StashCollection<K, V>
) -> Pin<Box<dyn Future<Output = Result<Vec<(K, V, Diff)>, StashError>> + Send + 'async_trait>> where
    K: Data,
    V: Data,
    K: 'async_trait,
    V: 'async_trait,
    'life0: 'async_trait,
    Self: 'async_trait, 

Returns the current value of sealed entries.

fn peek_one<'life0, 'async_trait, K, V>(
    &'life0 mut self,
    collection: StashCollection<K, V>
) -> Pin<Box<dyn Future<Output = Result<BTreeMap<K, V>, StashError>> + Send + 'async_trait>> where
    K: Data + Hash,
    V: Data,
    K: 'async_trait,
    V: 'async_trait,
    'life0: 'async_trait,
    Self: 'async_trait, 

Returns the current k,v pairs of sealed entries, erroring if there is more than one entry for a given key or the multiplicity is not 1 for each key.

fn peek_key_one<'life0, 'life1, 'async_trait, K, V>(
    &'life0 mut self,
    collection: StashCollection<K, V>,
    key: &'life1 K
) -> Pin<Box<dyn Future<Output = Result<Option<V>, StashError>> + Send + 'async_trait>> where
    K: Data,
    V: Data,
    K: 'async_trait,
    V: 'async_trait,
    'life0: 'async_trait,
    'life1: 'async_trait,
    Self: 'async_trait, 

Returns the current sealed value for the given key, erroring if there is more than one entry for the key or its multiplicity is not 1.

fn update<'life0, 'async_trait, K, V>(
    &'life0 mut self,
    collection: StashCollection<K, V>,
    data: (K, V),
    time: Timestamp,
    diff: Diff
) -> Pin<Box<dyn Future<Output = Result<(), StashError>> + Send + 'async_trait>> where
    K: Data,
    V: Data,
    K: 'async_trait,
    V: 'async_trait,
    'life0: 'async_trait,
    Self: 'async_trait, 

Atomically adds a single entry to the arrangement.