Enum mz_storage::upsert::types::StateValue

pub enum StateValue<T, O> {
    Consolidating(Consolidating),
    Value(Value<T, O>),
}

UpsertState has 2 modes:

• Normal operation
• Consolidation.

This struct and its substructs are helpers to simplify the logic that individual UpsertState implementations need to do to manage these 2 modes.

Normal operation is simple, we just store an ordinary UpsertValue, and allow the implementer to store it any way they want. During consolidation, the logic is more complex. See the docs on StateValue::merge_update for more information.

Note also that this type is designed to support partial updates. All values are associated with an order key O that can be used to determine if a value existing in the UpsertStateBackend occurred before or after a value being considered for insertion.

O typically required to be : Default, with the default value sorting below all others. During consolidation, values consolidate correctly (as they are actual differential updates with diffs), so order keys are not required.

Variants

Consolidating(Consolidating)

Value(Value<T, O>)

Implementations

impl<T, O> StateValue<T, O>

pub fn finalized_value(value: Result<Row, UpsertError>, order: O) -> Self

A finalized, that is (assumed) persistent, value occurring at some order key.

pub fn tombstone(order: O) -> Self

A tombstoned value occurring at some order key.

pub fn is_tombstone(&self) -> bool

Whether the value is a tombstone.

pub fn order(&self) -> &O

Pull out the order for the given Value, assuming ensure_decoded has been called.

pub fn into_decoded(self) -> Value<T, O>

Pull out the Value value for a StateValue, after ensure_decoded has been called.

pub fn memory_size(&self) -> u64

The size of a StateValue, in memory. This is:

1. only used in the InMemoryHashMap implementation.
2. An estimate (it only looks at value sizes, and not errors)

Other implementations may use more accurate accounting.

impl<T: Eq, O> StateValue<T, O>

pub fn new_provisional_value( provisional_value: Result<Row, UpsertError>, provisional_ts: T, order: O, ) -> Self

Creates a new provisional value, occurring at some order key, observed at the given timestamp.

pub fn into_provisional_value( self, provisional_value: Result<Row, UpsertError>, provisional_ts: T, provisional_order: O, ) -> Self

Creates a provisional value, that retains the finalized value along with its order in this StateValue, if any.

We record the finalized value, so that we can present it when needed or when trying to read a provisional value at a different timestamp.

pub fn new_provisional_tombstone(provisional_ts: T, order: O) -> Self

Creates a new provisional tombstone occurring at some order key, observed at the given timestamp.

pub fn into_provisional_tombstone( self, provisional_ts: T, provisional_order: O, ) -> Self

Creates a provisional tombstone, that retains the finalized value along with its order in this StateValue, if any.

We record the current finalized value, so that we can present it when needed or when trying to read a provisional value at a different timestamp.

pub fn provisional_order(&self, ts: &T) -> Option<&O>

Returns the order of a provisional value at the given timestamp. If that doesn’t exist, the order of the finalized value.

Returns None if none of the above exist.

pub fn provisional_value_ref(&self, ts: &T) -> Option<&Result<Row, UpsertError>>

Returns the provisional value, if one is present at the given timestamp. Falls back to the finalized value, or None if there is neither.

pub fn into_finalized_value(self) -> Option<(Result<Row, UpsertError>, O)>

Returns the the finalized value, if one is present.

impl<T: Eq, O: Default> StateValue<T, O>

pub fn merge_update( &mut self, value: Result<Row, UpsertError>, diff: Diff, bincode_opts: DefaultOptions, bincode_buffer: &mut Vec<u8>, ) -> bool

We use a XOR trick in order to accumulate the values without having to store the full unconsolidated history in memory. For all (value, diff) updates of a key we track:

• diff_sum = SUM(diff)
• checksum_sum = SUM(checksum(bincode(value)) * diff)
• len_sum = SUM(len(bincode(value)) * diff)
• value_xor = XOR(bincode(value))

Return value

Returns a bool indicating whether or not the current merged value is able to be deleted.

Correctness

The method is correct because a well formed upsert collection at a given timestamp will have for each key:

• Zero or one updates of the form (cur_value, +1)
• Zero or more pairs of updates of the form (prev_value, +1), (prev_value, -1)

We are interested in extracting the cur_value of each key and discard all prev_values that might be included in the stream. Since the history of prev_values always comes in pairs, computing the XOR of those is always going to cancel their effects out. Also, since XOR is commutative this property is true independent of the order. The same is true for the summations of the length and checksum since the sum will contain the unrelated values zero times.

Therefore the accumulators will end up precisely in one of two states:

1. diff == 0, checksum == 0, value == [0..] => the key is not present
2. diff == 1, checksum == checksum(cur_value) value == cur_value => the key is present

Robustness

In the absense of bugs, accumulating the diff and checksum is not required since we know that a well formed collection always satisfies XOR(bincode(values)) == bincode(cur_value). However bugs may happen and so storing 16 more bytes per key to have a very high guarantee that we’re not decoding garbage is more than worth it. The main key->value used to store previous values.

pub fn merge_update_state(&mut self, other: &Self)

Merge an existing StateValue into this one, using the same method described in merge_update. See the docstring above for more information on correctness and robustness.

pub fn ensure_decoded(&mut self, bincode_opts: DefaultOptions)

During and after consolidation, we assume that values in the UpsertStateBackend implementation can be Self::Consolidating, with a diff_sum of 1 (or 0, if they have been deleted). Afterwards, if we need to retract one of these values, we need to assert that its in this correct state, then mutate it to its Value state, so the upsert operator can use it.

Trait Implementations

impl<T: Clone, O: Clone> Clone for StateValue<T, O>

fn clone(&self) -> StateValue<T, O>

Returns a copy of the value.

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

Performs copy-assignment from source.

impl<T, O> Debug for StateValue<T, O>

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

Formats the value using the given formatter.

impl<T, O> Default for StateValue<T, O>

fn default() -> Self

Returns the “default value” for a type.

impl<'de, T, O> Deserialize<'de> for StateValue<T, O>

where T: Deserialize<'de>, O: Deserialize<'de>,

fn deserialize<__D>(__deserializer: __D) -> Result<Self, __D::Error>

where __D: Deserializer<'de>,

Deserialize this value from the given Serde deserializer.

impl<T, O> Serialize for StateValue<T, O>

where T: Serialize, O: Serialize,

fn serialize<__S>(&self, __serializer: __S) -> Result<__S::Ok, __S::Error>

where __S: Serializer,

Serialize this value into the given Serde serializer.