Struct TxnsProgress 

pub struct TxnsProgress<'scope, T: Timestamp> {
    remap: StreamVec<'scope, T, DataRemapEntry<T>>,
    name: String,
    data_id: ShardId,
    unique_id: u64,
}

A subscription to one data shard’s remap information, from which any number of streams read off that shard can have their frontiers translated.

The subscription is the expensive half and does not depend on what is read, so it is rendered once by TxnsProgress::new. TxnsProgress::translate then renders one passthrough operator per stream, and is generic over the container so streams of different shapes can share a subscription.

Fields

remap: StreamVec<'scope, T, DataRemapEntry<T>>

Broadcast remap stream, one operator per call to translate reads it.

name: String

data_id: ShardId

unique_id: u64

Disambiguates the log lines of operators rendered for the same shard.

Implementations

impl<'scope, T> TxnsProgress<'scope, T>

where T: Timestamp + Lattice + TotalOrder + StepForward + Codec64 + Sync,

pub fn new<K, V, D, C, F>( scope: Scope<'scope, T>, name: &str, ctx: &TxnsContext, client_fn: impl Fn() -> F, txns_id: ShardId, data_id: ShardId, as_of: T, data_key_schema: Arc<K::Schema>, data_val_schema: Arc<V::Schema>, ) -> (Self, PressOnDropButton)

where K: Debug + Codec + Send + Sync, V: Debug + Codec + Send + Sync, D: Debug + Clone + 'static + Monoid + Ord + Codec64 + Send + Sync, C: TxnsCodec + 'static, F: Future<Output = PersistClient> + Send + 'static,

Subscribe to data_id’s remap information and broadcast it to every worker.

impl<'scope, T> TxnsProgress<'scope, T>

where T: Timestamp + Lattice + TotalOrder + StepForward + Codec64,

pub fn translate<C: Container>( &self, passthrough: Stream<'scope, T, C>, until: Antichain<T>, ) -> (Stream<'scope, T, C>, PressOnDropButton)

Delay passthrough’s capability by the subscription’s remap, translating the data shard’s physical frontier into the logical one.

Call once per stream read off the shard. Streams of different container types can share one subscription.

The block ordering inside the schedule closure is load-bearing: pending passthrough input is emitted at the pre-activation capability BEFORE any capability downgrade, which keeps the differential invariant send_time <= record_time and avoids dropping in-flight rows when the passthrough frontier crosses until in the same activation (SQL-299). Do not reorder.

Trait Implementations

impl<'scope, T: Debug + Timestamp> Debug for TxnsProgress<'scope, T>

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

Formats the value using the given formatter.