Struct CollectionManager

pub struct CollectionManager<T>
where
    T: Timestamp + Lattice + Codec64 + TimestampManipulation,
{
    read_only: bool,
    differential_collections: Arc<Mutex<BTreeMap<GlobalId, (UnboundedSender<(StorageWriteOp, Sender<Result<(), StorageError<T>>>)>, AbortOnDropHandle<()>, Sender<()>)>>>,
    append_only_collections: Arc<Mutex<BTreeMap<GlobalId, (UnboundedSender<(Vec<AppendOnlyUpdate>, Sender<Result<(), StorageError<T>>>)>, AbortOnDropHandle<()>, Sender<()>)>>>,
    user_batch_duration_ms: Arc<AtomicU64>,
    now: NowFn,
}

Fields

read_only: bool

When a CollectionManager is in read-only mode it must not affect any changes to external state.

differential_collections: Arc<Mutex<BTreeMap<GlobalId, (UnboundedSender<(StorageWriteOp, Sender<Result<(), StorageError<T>>>)>, AbortOnDropHandle<()>, Sender<()>)>>>

These are collections that we write to by adding/removing updates to an internal desired collection. The CollectionManager continually makes sure that collection contents (in persist) match the desired state.

append_only_collections: Arc<Mutex<BTreeMap<GlobalId, (UnboundedSender<(Vec<AppendOnlyUpdate>, Sender<Result<(), StorageError<T>>>)>, AbortOnDropHandle<()>, Sender<()>)>>>

Collections that we only append to using blind-writes.

Every write succeeds at some timestamp, and we never check what the actual contents of the collection (in persist) are.

user_batch_duration_ms: Arc<AtomicU64>

Amount of time we’ll wait before sending a batch of inserts to Persist, for user collections.

now: NowFn

Implementations

impl<T> CollectionManager<T>

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

The CollectionManager provides two complementary functions:

• Providing an API to append values to a registered set of collections. For this usecase:
  • The CollectionManager expects to be the only writer.
  • Appending to a closed collection panics
• Automatically advancing the timestamp of managed collections every second. For this usecase:
  • The CollectionManager handles contention by permitting and ignoring errors.
  • Closed collections will not panic if they continue receiving these requests.

pub(crate) fn new(read_only: bool, now: NowFn) -> CollectionManager<T>

pub fn update_user_batch_duration(&self, duration: Duration)

Updates the duration we’ll wait to batch events for user owned collections.

pub(crate) fn register_differential_collection<R>( &self, id: GlobalId, write_handle: WriteHandle<SourceData, (), T, StorageDiff>, read_handle_fn: R, force_writable: bool, introspection_config: DifferentialIntrospectionConfig<T>, )

where R: FnMut() -> Pin<Box<dyn Future<Output = ReadHandle<SourceData, (), T, StorageDiff>> + Send>> + Send + Sync + 'static,

Registers a new differential collection.

The CollectionManager will automatically advance the upper of every registered collection every second.

Update the desired state of a differential collection using Self::differential_write.

pub(crate) fn register_append_only_collection( &self, id: GlobalId, write_handle: WriteHandle<SourceData, (), T, StorageDiff>, force_writable: bool, introspection_config: Option<AppendOnlyIntrospectionConfig<T>>, )

Registers a new append-only collection.

The CollectionManager will automatically advance the upper of every registered collection every second.

pub(crate) fn unregister_collection( &self, id: GlobalId, ) -> BoxFuture<'static, ()>

Unregisters the given collection.

Also waits until the CollectionManager has completed all outstanding work to ensure that it has stopped referencing the provided id.

pub(crate) fn append_only_write_sender( &self, id: GlobalId, ) -> UnboundedSender<(Vec<AppendOnlyUpdate>, Sender<Result<(), StorageError<T>>>)>

Returns a sender for writes to the given append-only collection.

Panics

• If id does not belong to an append-only collections.

pub(crate) fn differential_write_sender( &self, id: GlobalId, ) -> UnboundedSender<(StorageWriteOp, Sender<Result<(), StorageError<T>>>)>

Returns a sender for writes to the given differential collection.

Panics

• If id does not belong to a differential collections.

pub(crate) fn blind_write(&self, id: GlobalId, updates: Vec<AppendOnlyUpdate>)

Appends updates to the append-only collection identified by id, at some timestamp. Does not wait for the append to complete.

Panics

• If id does not belong to an append-only collections.
• If this CollectionManager is in read-only mode.
• If the collection closed.

pub(crate) fn differential_write(&self, id: GlobalId, op: StorageWriteOp)

Updates the desired collection state of the differential collection identified by id. The underlying persist shard will reflect this change at _some_point. Does not wait for the change to complete.

Panics

• If id does not belong to a differential collection.
• If the collection closed.

pub(crate) fn differential_append( &self, id: GlobalId, updates: Vec<(Row, Diff)>, )

Appends the given updates to the differential collection identified by id.

Panics

• If id does not belong to a differential collection.
• If the collection closed.

pub(crate) fn monotonic_appender( &self, id: GlobalId, ) -> Result<MonotonicAppender<T>, StorageError<T>>

Returns a MonotonicAppender that can be used to monotonically append updates to the collection correlated with id.

fn get_read_only(&self, id: GlobalId, force_writable: bool) -> bool

Trait Implementations

impl<T> Clone for CollectionManager<T>

where T: Timestamp + Lattice + Codec64 + TimestampManipulation + Clone,

fn clone(&self) -> CollectionManager<T>

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl<T> Debug for CollectionManager<T>

where T: Timestamp + Lattice + Codec64 + TimestampManipulation + Debug,

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

Formats the value using the given formatter.