Struct mz_persist_client::read::ReadHandle

pub struct ReadHandle<K, V, T, D>
where
    T: Timestamp + Lattice + Codec64,
    K: Debug + Codec,
    V: Debug + Codec,
    D: Semigroup + Codec64 + Send + Sync,
{ /* private fields */ }

A “capability” granting the ability to read the state of some shard at times greater or equal to self.since().

Production users should call Self::expire before dropping a ReadHandle so that it can expire its leases. If/when rust gets AsyncDrop, this will be done automatically.

All async methods on ReadHandle retry for as long as they are able, but the returned std::future::Futures implement “cancel on drop” semantics. This means that callers can add a timeout using tokio::time::timeout or tokio::time::timeout_at.

tokio::time::timeout(timeout, read.downgrade_since(&new_since)).await

Implementations

impl<K, V, T, D> ReadHandle<K, V, T, D>

where K: Debug + Codec, V: Debug + Codec, T: Timestamp + Lattice + Codec64, D: Semigroup + Codec64 + Send + Sync,

pub fn shard_id(&self) -> ShardId

This handle’s shard id.

pub fn since(&self) -> &Antichain<T>

This handle’s since frontier.

This will always be greater or equal to the shard-global since.

pub async fn downgrade_since(&mut self, new_since: &Antichain<T>)

Forwards the since frontier of this handle, giving up the ability to read at times not greater or equal to new_since.

This may trigger (asynchronous) compaction and consolidation in the system. A new_since of the empty antichain “finishes” this shard, promising that no more data will ever be read by this handle.

This also acts as a heartbeat for the reader lease (including if called with new_since equal to something like self.since() or the minimum timestamp, making the call a no-op).

pub async fn listen( self, as_of: Antichain<T> ) -> Result<Listen<K, V, T, D>, Since<T>>

Returns an ongoing subscription of updates to a shard.

The stream includes all data at times greater than as_of. Combined with Self::snapshot it will produce exactly correct results: the snapshot is the TVCs contents at as_of and all subsequent updates occur at exactly their indicated time. The recipient should only downgrade their read capability when they are certain they have all data through the frontier they would downgrade to.

This takes ownership of the ReadHandle so the Listen can use it to Self::downgrade_since as it progresses. If you need to keep this handle, then Self::clone it before calling listen.

The Since error indicates that the requested as_of cannot be served (the caller has out of date information) and includes the smallest as_of that would have been accepted.

pub async fn snapshot( &mut self, as_of: Antichain<T> ) -> Result<Vec<LeasedBatchPart<T>>, Since<T>>

Returns all of the contents of the shard TVC at as_of broken up into LeasedBatchPartes. These parts can be “turned in” via crate::fetch::fetch_batch_part to receive the data they contain.

This command returns the contents of this shard as of as_of once they are known. This may “block” (in an async-friendly way) if as_of is greater or equal to the current upper of the shard. The recipient should only downgrade their read capability when they are certain they have all data through the frontier they would downgrade to.

The Since error indicates that the requested as_of cannot be served (the caller has out of date information) and includes the smallest as_of that would have been accepted.

pub async fn subscribe( self, as_of: Antichain<T> ) -> Result<Subscribe<K, V, T, D>, Since<T>>

Returns a snapshot of all of a shard’s data using as_of, followed by listening to any future updates.

For more details on this operation’s semantics, see Self::snapshot and Self::listen.

pub fn process_returned_leased_part(&mut self, leased_part: LeasedBatchPart<T>)

Processes that a part issued from self has been consumed, and self can now process any internal bookkeeping.

Panics

• If self does not have record of issuing the LeasedBatchPart, e.g. it originated from from another ReadHandle.

pub async fn clone(&self, purpose: &str) -> Self

Returns an independent ReadHandle with a new LeasedReaderId but the same since.

pub async fn maybe_downgrade_since(&mut self, new_since: &Antichain<T>)

A rate-limited version of Self::downgrade_since.

This is an internally rate limited helper, designed to allow users to call it as frequently as they like. Call this Self::downgrade_since, or Self::maybe_heartbeat_reader on some interval that is “frequent” compared to PersistConfig::FAKE_READ_LEASE_DURATION.

This is communicating actual progress information, so is given preferential treatment compared to Self::maybe_heartbeat_reader.

pub async fn expire(self)

Politely expires this reader, releasing its lease.

There is a best-effort impl in Drop to expire a reader that wasn’t explictly expired with this method. When possible, explicit expiry is still preferred because the Drop one is best effort and is dependant on a tokio Handle being available in the TLC at the time of drop (which is a bit subtle). Also, explicit expiry allows for control over when it happens.

impl<K, V, T, D> ReadHandle<K, V, T, D>

where K: Debug + Codec + Ord, V: Debug + Codec + Ord, T: Timestamp + Lattice + Codec64, D: Semigroup + Codec64 + Send + Sync,

pub async fn snapshot_and_fetch( &mut self, as_of: Antichain<T> ) -> Result<Vec<((Result<K, String>, Result<V, String>), T, D)>, Since<T>>

Generates a Self::snapshot, and fetches all of the batches it contains.

The output is consolidated. Furthermore, to keep memory usage down when reading a snapshot that consolidates well, this consolidates as it goes.

Potential future improvements (if necessary):

• Accept something like a F: Fn(K,V) -> (K,V) argument, which looks like an MFP you might be pushing down. Reason being that if you are projecting or transforming in a way that allows further consolidation, amazing.
• Reuse any code we write to streaming-merge consolidate in persist_source here.

pub async fn snapshot_cursor( &mut self, as_of: Antichain<T>, should_fetch_part: impl for<'a> Fn(Option<&'a LazyPartStats>) -> bool ) -> Result<Cursor<K, V, T, D>, Since<T>>

Generates a Self::snapshot, and fetches all of the batches it contains.

To keep memory usage down when reading a snapshot that consolidates well, this consolidates as it goes. However, note that only the serialized data is consolidated: the deserialized data will only be consolidated if your K/V codecs are one-to-one.

pub async fn snapshot_parts_stats( &mut self, as_of: Antichain<T> ) -> Result<SnapshotPartsStats, Since<T>>

Returns aggregate statistics about the contents of the shard TVC at the given frontier.

This command returns the contents of this shard as of as_of once they are known. This may “block” (in an async-friendly way) if as_of is greater or equal to the current upper of the shard.

The Since error indicates that the requested as_of cannot be served (the caller has out of date information) and includes the smallest as_of that would have been accepted.

impl<K, V, T, D> ReadHandle<K, V, T, D>

where K: Debug + Codec + Ord + Default, V: Debug + Codec + Ord + Default, T: Timestamp + Lattice + Codec64, D: Semigroup + Codec64 + Send + Sync,

pub async fn snapshot_and_stream( &mut self, as_of: Antichain<T> ) -> Result<impl Stream<Item = ((Result<K, String>, Result<V, String>), T, D)>, Since<T>>

Generates a Self::snapshot, and streams out all of the updates it contains in bounded memory.

The output is not consolidated.

Trait Implementations

impl<K, V, T, D> Debug for ReadHandle<K, V, T, D>

where T: Timestamp + Lattice + Codec64 + Debug, K: Debug + Codec + Debug, V: Debug + Codec + Debug, D: Semigroup + Codec64 + Send + Sync + Debug,

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

Formats the value using the given formatter.

impl<K, V, T, D> Drop for ReadHandle<K, V, T, D>

where T: Timestamp + Lattice + Codec64, K: Debug + Codec, V: Debug + Codec, D: Semigroup + Codec64 + Send + Sync,

fn drop(&mut self)

Executes the destructor for this type.