Struct mz_storage::source::reclock::compat::PersistHandle

pub struct PersistHandle<FromTime: SourceTimestamp, IntoTime: Timestamp + Lattice + Codec64> {
    events: LocalBoxStream<'static, ListenEvent<IntoTime, ((Result<SourceData, String>, Result<(), String>), IntoTime, Diff)>>,
    write_handle: WriteHandle<SourceData, (), IntoTime, Diff>,
    read_only_rx: Receiver<bool>,
    pending_batch: Vec<(FromTime, IntoTime, Diff)>,
    shared_write_frontier: Rc<RefCell<Antichain<IntoTime>>>,
}

A handle to a persist shard that stores remap bindings

Fields

events: LocalBoxStream<'static, ListenEvent<IntoTime, ((Result<SourceData, String>, Result<(), String>), IntoTime, Diff)>>

write_handle: WriteHandle<SourceData, (), IntoTime, Diff>

read_only_rx: Receiver<bool>

Whether or not this handle is in read-only mode.

pending_batch: Vec<(FromTime, IntoTime, Diff)>

shared_write_frontier: Rc<RefCell<Antichain<IntoTime>>>

Implementations

impl<FromTime, IntoTime> PersistHandle<FromTime, IntoTime>

where FromTime: SourceTimestamp + Timestamp, IntoTime: Timestamp + Lattice + Codec64 + Sync,

pub async fn new( persist_clients: Arc<PersistClientCache>, read_only_rx: Receiver<bool>, metadata: CollectionMetadata, as_of: Antichain<IntoTime>, shared_write_frontier: Rc<RefCell<Antichain<IntoTime>>>, id: GlobalId, operator: &str, worker_id: usize, worker_count: usize, remap_relation_desc: RelationDesc, remap_collection_id: GlobalId, ) -> Result<Self>

Trait Implementations

impl<FromTime, IntoTime> RemapHandle for PersistHandle<FromTime, IntoTime>

where FromTime: SourceTimestamp, IntoTime: Timestamp + Lattice + Codec64 + Sync,

fn compare_and_append<'life0, 'async_trait>( &'life0 mut self, updates: Vec<(Self::FromTime, Self::IntoTime, Diff)>, upper: Antichain<Self::IntoTime>, new_upper: Antichain<Self::IntoTime>, ) -> Pin<Box<dyn Future<Output = Result<(), UpperMismatch<Self::IntoTime>>> + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Attempt to write the batch of remap collection updates to the collection. If the remap collection was already written by some other process an error will return with the current upper.

fn upper(&self) -> &Antichain<Self::IntoTime>

impl<FromTime, IntoTime> RemapHandleReader for PersistHandle<FromTime, IntoTime>

where FromTime: SourceTimestamp, IntoTime: Timestamp + Lattice + Codec64,

type FromTime = FromTime

type IntoTime = IntoTime

fn next<'life0, 'async_trait>( &'life0 mut self, ) -> Pin<Box<dyn Future<Output = Option<(Vec<(Self::FromTime, Self::IntoTime, Diff)>, Antichain<Self::IntoTime>)>> + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Produces the next batch of data contained in the remap collection and the upper frontier of that batch. The return batch should contain all the updates that happened at times not beyond ther returned upper.