Struct Spine 

pub struct Spine<B>
where
    B: Batch,
{ /* private fields */ }

Persistence optimized trace implementation based on collection and merging immutable batches of updates.

This spine works asynchronously. The batches exposed to cursors are maintained separately from the batches currently being merged by an asynchronous thread. When one or more merges complete, the spine fetches the new (smaller) collection of batches from the thread in the next step. (It could fetch them earlier, but it might be unfriendly to expose potentially one form of data to a given cursor and then a different form to the next one within a single step.)

Implementations

impl<B> Spine<B>

where B: Batch,

pub fn get_batches(&self) -> Vec<Arc<B>>

impl<B> Spine<B>

where B: Batch,

pub fn with_runtime( runtime: Runtime, worker_index: usize, factories: &B::Factories, name: Arc<String>, ) -> Self

Allocates a new Spine for worker_index in runtime.

In the common case, Spine::new uses the current thread’s runtime and worker index.

pub fn complete_merges(&mut self)

pub fn maybe_flush_batch<F>( batch: impl Into<Arc<B>>, factories: &B::Factories, filters: F, ) -> Arc<B>

where F: FnOnce() -> (Option<Filter<B::Key>>, Option<GroupFilter<B::Val>>),

Returns batch, first pushing it to storage if it exceeds the user-configured min_storage_bytes or if we’re under high memory pressure.

pub fn backpressure_waiter(&self) -> Option<OwnedNotified>

Returns an object that can be used to wait for backpressure to be relieved. Returns None if no backpressure is needed.

Trait Implementations

impl<B> Archive for Spine<B>

where B: Batch,

type Archived = ()

The archived representation of this type.

type Resolver = ()

The resolver for this type. It must contain all the additional information from serializing needed to make the archived type from the normal type.

unsafe fn resolve( &self, _pos: usize, _resolver: Self::Resolver, _out: *mut Self::Archived, )

Creates the archived version of this value at the given position and writes it to the given output.

impl<B> BatchReader for Spine<B>

where B: Batch,

type Key = <B as BatchReader>::Key

Key by which updates are indexed.

type Val = <B as BatchReader>::Val

Values associated with keys.

type Time = <B as BatchReader>::Time

Timestamps associated with updates

type R = <B as BatchReader>::R

Associated update.

type Factories = <B as BatchReader>::Factories

type Cursor<'s> = SpineCursor<B>

The type used to enumerate the batch’s contents.

fn factories(&self) -> Self::Factories

fn key_count(&self) -> usize

The number of keys in the batch.

fn len(&self) -> usize

The number of updates in the batch.

fn approximate_byte_size(&self) -> usize

The memory or storage size of the batch in bytes.

fn range_filter_stats(&self) -> FilterStats

Statistics of the in-memory range filter used by Cursor::seek_key_exact.

fn cursor(&self) -> Self::Cursor<'_>

Acquires a cursor to the batch’s contents.

fn sample_keys<RG>( &self, rng: &mut RG, sample_size: usize, sample: &mut DynVec<Self::Key>, )

where RG: Rng,

Returns a uniform random sample of distincts keys from the batch.

async fn fetch<KR>( &self, keys: &KR, ) -> Option<Box<dyn CursorFactory<Self::Key, Self::Val, Self::Time, Self::R>>>

where KR: BatchReader<Key = Self::Key, Time = ()>,

Creates and returns a new batch that is a subset of this one, containing only the key-value pairs whose keys are in keys. May also return None, the default implementation, if the batch doesn’t want to implement this method. In particular, a batch for which access through a cursor is fast should return None to avoid the expense of copying data.

fn push_cursor( &self, ) -> Box<dyn PushCursor<Self::Key, Self::Val, Self::Time, Self::R> + Send + '_>

Acquires a PushCursor for the batch’s contents.

fn merge_cursor( &self, key_filter: Option<Filter<Self::Key>>, value_filter: Option<GroupFilter<Self::Val>>, ) -> Box<dyn MergeCursor<Self::Key, Self::Val, Self::Time, Self::R> + Send + '_>

Acquires a MergeCursor for the batch’s contents.

fn merge_cursor_with_snapshot<'a, S>( &'a self, key_filter: Option<Filter<Self::Key>>, value_filter: Option<GroupFilter<Self::Val>>, snapshot: &'a Option<Arc<S>>, ) -> Box<dyn MergeCursor<Self::Key, Self::Val, Self::Time, Self::R> + Send + 'a>

where S: BatchReader<Key = Self::Key, Val = Self::Val, Time = Self::Time, R = Self::R>,

Similar to merge_cursor, but invoked in the context of a spine merger. Takes the current spine snapshot as an extra argument and uses it to evaluate value_filter precisely.

fn consuming_cursor( &mut self, key_filter: Option<Filter<Self::Key>>, value_filter: Option<GroupFilter<Self::Val>>, ) -> Box<dyn MergeCursor<Self::Key, Self::Val, Self::Time, Self::R> + Send + '_>

Acquires a merge cursor for the batch’s contents.

fn membership_filter_stats(&self) -> FilterStats

Statistics of the secondary membership filter used by Cursor::seek_key_exact after the range filter.

fn membership_filter_kind(&self) -> FilterKind

Filter kind for the secondary membership filter used by Cursor::seek_key_exact.

fn location(&self) -> BatchLocation

Where the batch’s data is stored.

fn cache_stats(&self) -> CacheStats

Storage cache access statistics for this batch only.

fn is_empty(&self) -> bool

True if the batch is empty.

fn partition_keys(&self, num_partitions: usize, bounds: &mut DynVec<Self::Key>)

where Self::Time: PartialEq<()>,

Returns num_partitions-1 keys from the batch that partition the batch into num_partitions approximately equal size ranges 0..key1, key1..key2, … , key_num_partitions-1..last_key_in_the_batch.

fn keys(&self) -> Option<&DynVec<Self::Key>>

impl<B> Clone for Spine<B>

where B: Batch,

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl<B> Debug for Spine<B>

where B: Batch,

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

Formats the value using the given formatter.

impl<B: Batch, D: Fallible> Deserialize<Spine<B>, D> for Archived<Spine<B>>

fn deserialize(&self, _deserializer: &mut D) -> Result<Spine<B>, D::Error>

Deserializes using the given deserializer

impl<B> Display for Spine<B>

where B: Batch + Display,

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

Formats the value using the given formatter.

impl<B> From<&Spine<B>> for SpineSnapshot<B>

where B: Batch + Send + Sync,

fn from(spine: &Spine<B>) -> Self

Converts to this type from the input type.

impl<B> NumEntries for Spine<B>

where B: Batch,

const CONST_NUM_ENTRIES: Option<usize> = None

Returns Some(n) if Self has constant size or None otherwise.

fn num_entries_shallow(&self) -> usize

Returns the number of entries in self.

fn num_entries_deep(&self) -> usize

Recursively computes the number of entries in a container by calling this method on each entry in self.

impl<B: Batch, S: Serializer + ?Sized> Serialize<S> for Spine<B>

fn serialize(&self, _serializer: &mut S) -> Result<Self::Resolver, S::Error>

Writes the dependencies for the object and returns a resolver that can create the archived type.

impl<B> SizeOf for Spine<B>

where B: Batch,

fn size_of_children(&self, context: &mut Context)

Gets the size of all “children” owned by this value, not including the size of the value itself.

fn size_of(&self) -> TotalSize

Gets the total size of the current value

fn size_of_with_context(&self, context: &mut Context)

Adds the size of the current value to the given Context, including both the size of the value itself and all of its children

impl<B> Trace for Spine<B>

where B: Batch,

fn insert_without_blocking(&mut self, batch: impl Into<Arc<B>>) -> bool

Inserts a batch into the spine without blocking. Thus, this omits:

• Spilling the batch to storage when that is a good idea. The caller may do this beforehand by calling Spine::maybe_flush_batch.

• Waiting until the number of batches in the spine falls below the level at which we impose backpressure. The function returns true if backpressure is warranted. The caller may do so afterward by calling Spine::backpressure_wait.

async fn backpressure_wait(&self)

Waits for the number of batches in the spine to fall below the level at which we impose backpressure.

type Batch = B

The type of an immutable collection of updates.

fn new(factories: &B::Factories, name: Arc<String>) -> Self

Allocates a new empty trace associated with name, which should identify the operator or other origin of the trace.

fn set_name(&mut self, name: Arc<String>)

Updates the name of the trace to name.

fn set_frontier(&mut self, frontier: &B::Time)

Sets a compaction frontier, i.e., a timestamp such that timestamps below the frontier are indistinguishable to DBSP, therefore any ts in the trace can be safely replaced with ts.join(frontier) without affecting the output of the circuit. By applying this replacement, updates to the same (key, value) pairs applied during different steps can be merged or discarded.

fn exert(&mut self, _effort: &mut isize)

Exert merge effort, even without updates.

fn consolidate(self) -> Option<B>

Merge all updates in a trace into a single batch.

async fn insert(&mut self, batch: impl Into<Arc<Self::Batch>>)

Introduces a batch of updates to the trace.

fn clear_dirty_flag(&mut self)

Clears the value of the “dirty” flag to false.

fn dirty(&self) -> bool

Returns the value of the dirty flag.

fn retain_keys(&mut self, filter: Filter<Self::Key>)

Informs the trace that keys that don’t pass the filter are no longer used and can be removed from the trace.

fn retain_values(&mut self, filter: GroupFilter<Self::Val>)

Informs the trace that values that don’t pass the filter are no longer used and can be removed from the trace.

fn key_filter(&self) -> &Option<Filter<Self::Key>>

fn value_filter(&self) -> &Option<GroupFilter<Self::Val>>

fn save( &mut self, base: &StoragePath, persistent_id: &str, files: &mut Vec<Arc<dyn FileCommitter>>, ) -> Result<(), Error>

Writes this trace to storage beneath base, using pid as a file name prefix. Adds the files that were written to files so that they can be committed later.

fn restore( &mut self, base: &StoragePath, persistent_id: &str, ) -> Result<(), Error>

Reads this trace back from storage under base with pid as the prefix.

fn metadata(&self, meta: &mut OperatorMeta)

Allows the trace to report additional metadata.

fn initiate_compaction(&self)

fn is_compaction_complete(&self) -> bool

Returns true when compaction has fully converged: all compaction requests have been processed, no background merge is in progress, and the trace has been reduced to at most one batch.

impl<B> WithSnapshot for Spine<B>

where B: Batch,

type Batch = B

fn ro_snapshot(&self) -> SpineSnapshot<B>

Returns a read-only, non-merging snapshot of the current trace state.

fn into_ro_snapshot(self) -> SpineSnapshot<Self::Batch>