Struct AppendOnlyWriteTask

struct AppendOnlyWriteTask<T>
where
    T: Lattice + Codec64 + From<EpochMillis> + TimestampManipulation,
{
    id: GlobalId,
    write_handle: WriteHandle<SourceData, (), T, StorageDiff>,
    read_only: bool,
    now: NowFn,
    user_batch_duration_ms: Arc<AtomicU64>,
    rx: UnboundedReceiver<(Vec<AppendOnlyUpdate>, Sender<Result<(), StorageError<T>>>)>,
    shutdown_rx: Receiver<()>,
    previous_statuses: Option<BTreeMap<(GlobalId, Option<ReplicaId>), Status>>,
}

A task that writes to an append only collection and continuously bumps the upper for the specified collection.

For status history collections, this task can deduplicate redundant Statuses.

Fields

id: GlobalId

The collection that we are writing to.

write_handle: WriteHandle<SourceData, (), T, StorageDiff>

read_only: bool

now: NowFn

user_batch_duration_ms: Arc<AtomicU64>

rx: UnboundedReceiver<(Vec<AppendOnlyUpdate>, Sender<Result<(), StorageError<T>>>)>

Receiver for write commands.

shutdown_rx: Receiver<()>

We have to shut down when receiving from this.

previous_statuses: Option<BTreeMap<(GlobalId, Option<ReplicaId>), Status>>

If this collection deduplicates statuses, this map is used to track the previous status.

Implementations

impl<T> AppendOnlyWriteTask<T>

where T: Lattice + Codec64 + From<EpochMillis> + TimestampManipulation,

fn spawn( id: GlobalId, write_handle: WriteHandle<SourceData, (), T, StorageDiff>, read_only: bool, now: NowFn, user_batch_duration_ms: Arc<AtomicU64>, introspection_config: Option<AppendOnlyIntrospectionConfig<T>>, ) -> (UnboundedSender<(Vec<AppendOnlyUpdate>, Sender<Result<(), StorageError<T>>>)>, AbortOnDropHandle<()>, Sender<()>)

Spawns an AppendOnlyWriteTask in an mz_ore::task that will continuously bump the upper for the specified collection, and append data that is sent via the provided mpsc::UnboundedSender.

TODO(parkmycar): One day if we want to customize the tick interval for each collection, that should be done here. TODO(parkmycar): Maybe add prometheus metrics for each collection?

async fn prepare( &mut self, introspection_config: Option<AppendOnlyIntrospectionConfig<T>>, )

Does any work that is required before the background task starts writing to the given append only introspection collection.

This might include consolidation or deleting older entries.

async fn run(self)

fn process_updates( &mut self, updates: Vec<AppendOnlyUpdate>, ) -> impl Iterator<Item = (Row, Diff)>

Deduplicate any mz_storage_client::client::StatusUpdate within updates and converts updates to rows and diffs.