Struct mz_storage::upsert::types::UpsertState

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

    inner: S,
    pub 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<T, O>>,
    multi_get_scratch: Vec<UpsertKey>,
    shrink_upsert_unused_buffers_by_ratio: usize,
}

An UpsertStateBackend wrapper that supports consolidating 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<T, O>>

multi_get_scratch: Vec<UpsertKey>

shrink_upsert_unused_buffers_by_ratio: usize

Implementations

impl<'metrics, S, T, O> UpsertState<'metrics, S, T, 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, T, O> UpsertState<'_, S, T, O>

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

pub async fn consolidate_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. Updates provided to this method can be assumed to consolidate into a single value per-key, after all chunks of updates for a given timestamp have been processed,

Therefore, after all updates of a given timestamp have been consolidated, all values must be in the correct state (as determined by StateValue::ensure_decoded).

The completed boolean communicates whether or not this is the final chunk of updates for the initial “snapshot” from persist.

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_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. 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 consolidating_merge_function.

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

async fn consolidate_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. 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, update_per_record_stats: bool, puts: P, ) -> Result<(), Error>

where P: IntoIterator<Item = (UpsertKey, PutValue<Value<T, 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<T, 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.