Struct mz_storage::upsert::types::UpsertState

pub struct UpsertState<'metrics, S, O> {

    inner: S,
    snapshot_start: Instant,
    snapshot_stats: SnapshotStats,
    snapshot_completed: bool,
    metrics: Arc<UpsertSharedMetrics>,
    worker_metrics: &'metrics UpsertMetrics,
    stats: SourceStatistics,
    bincode_opts: DefaultOptions,
    bincode_buffer: Vec<u8>,
    consolidate_scratch: Vec<(UpsertKey, Result<Row, UpsertError>, Diff)>,
    consolidate_upsert_scratch: IndexMap<UpsertKey, UpsertValueAndSize<O>>,
    multi_get_scratch: Vec<UpsertKey>,
    shrink_upsert_unused_buffers_by_ratio: usize,
}

An UpsertStateBackend wrapper that supports snapshot merging, and reports basic metrics about the usage of the UpsertStateBackend.

Fields

inner: S

snapshot_start: Instant

snapshot_stats: SnapshotStats

snapshot_completed: bool

metrics: Arc<UpsertSharedMetrics>

worker_metrics: &'metrics UpsertMetrics

stats: SourceStatistics

bincode_opts: DefaultOptions

bincode_buffer: Vec<u8>

consolidate_scratch: Vec<(UpsertKey, Result<Row, UpsertError>, Diff)>

consolidate_upsert_scratch: IndexMap<UpsertKey, UpsertValueAndSize<O>>

multi_get_scratch: Vec<UpsertKey>

shrink_upsert_unused_buffers_by_ratio: usize

Implementations

impl<'metrics, S, O> UpsertState<'metrics, S, O>

pub(crate) fn new( inner: S, metrics: Arc<UpsertSharedMetrics>, worker_metrics: &'metrics UpsertMetrics, stats: SourceStatistics, shrink_upsert_unused_buffers_by_ratio: usize, ) -> Self

impl<S, O> UpsertState<'_, S, O>

where S: UpsertStateBackend<O>, O: Default + Clone + Send + Sync + Serialize + DeserializeOwned + 'static,

pub async fn consolidate_snapshot_chunk<U>( &mut self, updates: U, completed: bool, ) -> Result<(), Error>

where U: IntoIterator<Item = (UpsertKey, Result<Row, UpsertError>, Diff)> + ExactSizeIterator,

Consolidate the following differential updates into the state, during snapshotting. Updates provided to this method can be assumed to consolidate into a single value per-key, after all chunks have been processed.

Therefore, after an entire snapshot has been consolidated, all values must be in the correct state (as determined by StateValue::ensure_decoded), and consolidate_snapshot_chunk must NOT be called again.

The completed boolean communicates whether or not this is the final chunk of updates to be consolidated, to assert correct usage.

If the backend supports it, this method will use multi_merge to consolidate the updates to avoid having to read the existing value for each key first. On some backends (like RocksDB), this can be significantly faster than the read-then-write consolidation strategy.

Also note that we use self.inner.multi_*, not self.multi_*. This is to avoid erroneously changing metric and stats values.

async fn consolidate_snapshot_merge_inner<U>( &mut self, updates: U, ) -> Result<SnapshotStats, Error>

where U: IntoIterator<Item = (UpsertKey, Result<Row, UpsertError>, Diff)> + ExactSizeIterator,

Consolidate the updates into the state during snapshotting. This method requires the backend has support for the multi_merge operation, and will panic if self.inner.supports_merge() was not checked before calling this method. multi_merge will write the updates as ‘merge operands’ to the backend, and then the backend will consolidate those updates with any existing state using the snapshot_merge_function.

This method can have significant performance benefits over the read-then-write method of consolidate_snapshot_read_write_inner.

async fn consolidate_snapshot_read_write_inner<U>( &mut self, updates: U, ) -> Result<SnapshotStats, Error>

where U: IntoIterator<Item = (UpsertKey, Result<Row, UpsertError>, Diff)> + ExactSizeIterator,

Consolidates the updates into the state during snapshotting. This method reads the existing values for each key, consolidates the updates, and writes the new values back to the state.

pub async fn multi_put<P>(&mut self, puts: P) -> Result<(), Error>

where P: IntoIterator<Item = (UpsertKey, PutValue<Value<O>>)>,

Insert or delete for all puts keys, prioritizing the last value for repeated keys.

pub async fn multi_get<'r, G, R>( &mut self, gets: G, results_out: R, ) -> Result<(), Error>

where G: IntoIterator<Item = UpsertKey>, R: IntoIterator<Item = &'r mut UpsertValueAndSize<O>>, O: 'r,

Get the gets keys, which must be unique, placing the results in results_out.

Panics if gets and results_out are not the same length.