Struct Arrangement

pub struct Arrangement {
    name: Vec<String>,
    spine: BTreeMap<i64, BTreeMap<Row, SmallVec<[(Row, i64, i64); 2]>>>,
    full_arrangement: bool,
    is_written: bool,
    expire_state: Option<KeyExpiryManager>,
    last_compaction_time: Option<i64>,
    estimated_size: usize,
    last_size_update: Instant,
    size_update_interval: Duration,
}

A shared state of key-value pair for various state in dataflow execution.

i.e: Mfp operator with temporal filter need to store it’s future output so that it can add now, and delete later. To get all needed updates in a time span, use [get_updates_in_range].

And reduce operator need full state of it’s output, so that it can query (and modify by calling [apply_updates]) existing state, also need a way to expire keys. To get a key’s current value, use [get] with time being now so it’s like: mfp operator -> arrange(store futures only, no expire) -> reduce operator <-> arrange(full, with key expiring time) -> output

Note the two way arrow between reduce operator and arrange, it’s because reduce operator need to query existing state and also need to update existing state.

Fields

name: Vec<String>

A name or identifier for the arrangement which can be used for debugging or logging purposes. This field is not critical to the functionality but aids in monitoring and management of arrangements.

spine: BTreeMap<i64, BTreeMap<Row, SmallVec<[(Row, i64, i64); 2]>>>

Manages a collection of pending updates in a BTreeMap where each key is a timestamp and each value is a Batch of updates. Updates are grouped into batched based on their timestamps. Each batch covers a range of time from the last key (exclusive) to the current key (inclusive).

• Updates with a timestamp (update_ts) that falls between two keys are placed in the batch of the higher key. For example, if the keys are 1, 5, 7, 9 and update_ts is 6, the update goes into the batch with key 7.
• Updates with a timestamp before the first key are categorized under the first key.
• Updates with a timestamp greater than the highest key result in a new batch being created with that timestamp as the key.

The first key represents the current state and includes consolidated updates from the past. It is always set to now. Each key should have only one update per batch with a diff=1 for the batch representing the current time (now).

Since updates typically occur as a delete followed by an insert, a small vector of size 2 is used to store updates for efficiency.

TODO(discord9): Consider balancing the batch size?

full_arrangement: bool

Indicates whether the arrangement maintains a complete history of updates.

• true: Maintains all past and future updates, necessary for full state reconstruction at any point in time.
• false: Only future updates are retained, optimizing for scenarios where past state is irrelevant and conserving resources. Useful for case like map -> arrange -> reduce.

is_written: bool

Indicates whether the arrangement has been modified since its creation.

• true: The arrangement has been written to, meaning it has received updates. Cloning this arrangement is generally unsafe as it may lead to inconsistencies if the clone is modified independently. However, cloning is safe when both the original and the clone require a full arrangement, as this ensures consistency.
• false: The arrangement is in its initial state and has not been modified. It can be safely cloned and shared without concerns of carrying over unintended state changes.

expire_state: Option<KeyExpiryManager>

Manage the expire state of the arrangement.

last_compaction_time: Option<i64>

The time that the last compaction happened, also known as the current time.

estimated_size: usize

Estimated size of the arrangement in heap size.

last_size_update: Instant

size_update_interval: Duration

Implementations

impl Arrangement

fn compute_size(&self) -> usize

fn update_and_fetch_size(&mut self) -> usize

impl Arrangement

pub fn new_with_name(name: Vec<String>) -> Self

pub fn get_expire_state(&self) -> Option<&KeyExpiryManager>

pub fn set_expire_state(&mut self, expire_state: KeyExpiryManager)

pub fn apply_updates( &mut self, now: i64, updates: Vec<((Row, Row), i64, i64)>, ) -> Result<Option<i64>, EvalError>

Apply updates into spine, with no respect of whether the updates are in futures, past, or now.

Return the maximum expire time (already expire by how much time) of all updates if any keys is already expired.

pub fn get_next_update_time(&self, now: &i64) -> Option<i64>

Find out the time of next update in the future that is the next update with timestamp > now.

pub fn last_compaction_time(&self) -> Option<i64>

Get the last compaction time.

fn split_spine_le( &mut self, split_ts: &i64, ) -> BTreeMap<i64, BTreeMap<Row, SmallVec<[(Row, i64, i64); 2]>>>

Split spine off at split_ts, and return the spine that’s before split_ts (including split_ts).

fn split_batch_at(&mut self, split_ts: &i64)

Split the batch at split_ts into two parts.

pub fn compact_to(&mut self, now: i64) -> Result<Option<i64>, EvalError>

Advance time to now and consolidate all older (now included) updates to the first key.

Return the maximum expire time(already expire by how much time) of all updates if any keys is already expired.

pub fn get_updates_in_range<R: RangeBounds<i64> + Clone>( &self, range: R, ) -> Vec<((Row, Row), i64, i64)>

Get the updates of the arrangement from the given range of time.

pub fn truncate_expired_keys(&mut self, now: i64)

Expire keys in now that are older than expire_time, intended for reducing memory usage and limit late data arrive

pub fn get(&self, now: i64, key: &Row) -> Option<(Row, i64, i64)>

Get current state of things.

Useful for query existing keys (i.e. reduce and join operator need to query existing state)

Trait Implementations

impl Clone for Arrangement

fn clone(&self) -> Arrangement

Returns a copy of the value.

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

Performs copy-assignment from source.

impl Debug for Arrangement

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

Formats the value using the given formatter.

impl Default for Arrangement

fn default() -> Self

Returns the “default value” for a type.

impl GetSize for Arrangement

fn get_heap_size(&self) -> usize

Determines how many bytes this object occupies inside the heap.

fn get_stack_size() -> usize

Determines how may bytes this object occupies inside the stack.

fn get_heap_size_with_tracker<T>(&self, tracker: T) -> (usize, T)

where T: GetSizeTracker,

Determines how many bytes this object occupies inside the heap while using a tracker.

fn get_size(&self) -> usize

Determines the total size of the object.

fn get_size_with_tracker<T>(&self, tracker: T) -> (usize, T)

where T: GetSizeTracker,

Determines the total size of the object while using a tracker.

impl Ord for Arrangement

fn cmp(&self, other: &Arrangement) -> Ordering

This method returns an Ordering between self and other.

fn max(self, other: Self) -> Self

where Self: Sized,

Compares and returns the maximum of two values.

fn min(self, other: Self) -> Self

where Self: Sized,

Compares and returns the minimum of two values.

fn clamp(self, min: Self, max: Self) -> Self

where Self: Sized,

Restrict a value to a certain interval.

impl PartialEq for Arrangement

fn eq(&self, other: &Arrangement) -> bool

Tests for self and other values to be equal, and is used by ==.

fn ne(&self, other: &Rhs) -> bool

Tests for !=. The default implementation is almost always sufficient, and should not be overridden without very good reason.

impl PartialOrd for Arrangement

fn partial_cmp(&self, other: &Arrangement) -> Option<Ordering>

This method returns an ordering between self and other values if one exists.

fn lt(&self, other: &Rhs) -> bool

Tests less than (for self and other) and is used by the < operator.

fn le(&self, other: &Rhs) -> bool

Tests less than or equal to (for self and other) and is used by the <= operator.

fn gt(&self, other: &Rhs) -> bool

Tests greater than (for self and other) and is used by the > operator.

fn ge(&self, other: &Rhs) -> bool

Tests greater than or equal to (for self and other) and is used by the >= operator.

impl Eq for Arrangement

impl StructuralPartialEq for Arrangement