Enum FlowTransaction 

pub enum FlowTransaction {
    Deferred {
        inner: FlowTransactionInner,
    },
    Transactional {
        inner: FlowTransactionInner,
        base_pending: Pending,
        view_overlay: Arc<Vec<Change>>,
    },
    Ephemeral {
        inner: FlowTransactionInner,
        state: HashMap<EncodedKey, EncodedRow>,
    },
}

Variants

Deferred

CDC-driven async flow processing. Reads only from committed storage + flow pending writes.

Fields

inner: FlowTransactionInner

Transactional

Inline flow processing within a committing transaction. Can additionally read uncommitted writes from the parent transaction.

Fields

inner: FlowTransactionInner

base_pending: Pending

Read-only snapshot of the committing transaction’s KV writes.

view_overlay: Arc<Vec<Change>>

View outputs produced by sibling flows in earlier execution levels of this pre-commit. Consulted by view-reading pull paths to overlay in-transaction writes on top of the read_version snapshot. Empty for the first level.

Ephemeral

Ephemeral subscription flow processing.

Operator state lives in an in-memory HashMap instead of the multi-version store; source reads go through query at the CDC version. No writes are committed to persistent storage.

Fields

inner: FlowTransactionInner

state: HashMap<EncodedKey, EncodedRow>

In-memory operator state, replacing state_query for FlowNodeState keys.

Implementations

impl FlowTransaction

pub fn get(&mut self, key: &EncodedKey) -> Result<Option<EncodedRow>>

Get a value by key, checking pending writes first, then (if transactional) base_pending, then querying multi-version store

pub fn contains_key(&mut self, key: &EncodedKey) -> Result<bool>

Check if a key exists

pub fn prefix(&mut self, prefix: &EncodedKey) -> Result<MultiVersionBatch>

Prefix scan

pub fn range( &mut self, range: EncodedKeyRange, batch_size: usize, ) -> Box<dyn Iterator<Item = Result<MultiVersionRow>> + Send + '_>

Create an iterator for forward range queries.

This properly handles high version density by scanning until batch_size unique logical keys are collected. The iterator yields individual entries and maintains cursor state internally. Pending writes are merged with committed storage data.

pub fn range_rev( &mut self, range: EncodedKeyRange, batch_size: usize, ) -> Box<dyn Iterator<Item = Result<MultiVersionRow>> + Send + '_>

Create an iterator for reverse range queries.

This properly handles high version density by scanning until batch_size unique logical keys are collected. The iterator yields individual entries in reverse key order and maintains cursor state internally.

impl FlowTransaction

pub fn state_get( &mut self, id: FlowNodeId, key: &EncodedKey, ) -> Result<Option<EncodedRow>>

pub fn state_set( &mut self, id: FlowNodeId, key: &EncodedKey, value: EncodedRow, ) -> Result<()>

pub fn state_remove(&mut self, id: FlowNodeId, key: &EncodedKey) -> Result<()>

pub fn state_scan(&mut self, id: FlowNodeId) -> Result<MultiVersionBatch>

pub fn state_range( &mut self, id: FlowNodeId, range: EncodedKeyRange, ) -> Result<MultiVersionBatch>

pub fn state_clear(&mut self, id: FlowNodeId) -> Result<()>

pub fn load_or_create_row( &mut self, id: FlowNodeId, key: &EncodedKey, shape: &RowShape, ) -> Result<EncodedRow>

pub fn save_row( &mut self, id: FlowNodeId, key: &EncodedKey, row: EncodedRow, ) -> Result<()>

impl FlowTransaction

pub fn set(&mut self, key: &EncodedKey, value: EncodedRow) -> Result<()>

Set a value, buffering it in pending writes

pub fn remove(&mut self, key: &EncodedKey) -> Result<()>

Remove a key, buffering the deletion in pending operations

impl FlowTransaction

pub fn deferred( parent: &AdminTransaction, version: CommitVersion, catalog: Catalog, interceptors: Interceptors, clock: Clock, ) -> Self

Create a deferred (CDC) FlowTransaction from a parent transaction.

Used by the async worker path. Reads only from committed storage + flow-generated pending writes - no base pending from a parent transaction.

pub fn deferred_from_parts( version: CommitVersion, pending: Pending, query: MultiReadTransaction, state_query: MultiReadTransaction, catalog: Catalog, interceptors: Interceptors, clock: Clock, ) -> Self

Create a deferred (CDC) FlowTransaction from pre-built parts.

Used by the worker actor which creates its own query transactions.

pub fn transactional(params: TransactionalParams) -> Self

Create a transactional (inline) FlowTransaction.

Used by the pre-commit interceptor path. base_pending is a read-only snapshot of the committing transaction’s KV writes so that flow operators can see uncommitted row data.

pub fn view_overlay(&self) -> Option<Arc<Vec<Change>>>

Return a (cheap) clone of the in-transaction view overlay, if any. Returns None for Deferred / Ephemeral transactions (which read everything from committed storage). Operators reading from a view parent overlay these changes on top of their storage reads so sibling transactional view outputs produced earlier in the same pre-commit are visible.

pub fn ephemeral( version: CommitVersion, query: MultiReadTransaction, catalog: Catalog, state: HashMap<EncodedKey, EncodedRow>, clock: Clock, ) -> Self

Create an ephemeral (subscription) FlowTransaction.

Operator state is backed by an in-memory HashMap. Source data reads go through query at the specified version. State reads go to state instead of the multi-version store.

pub fn merge_state(&mut self)

Merge pending state writes back into the ephemeral state HashMap.

After flow processing, pending writes contain both state mutations and subscription output writes. This method merges state mutations (keys matching FlowNodeState/FlowNodeInternalState) back into state and clears pending.

Only applicable to the Ephemeral variant; no-op for others.

pub fn take_state(&mut self) -> HashMap<EncodedKey, EncodedRow>

Extract the ephemeral state HashMap, consuming the state from this transaction.

Used to persist ephemeral state across CDC batches. Only applicable to the Ephemeral variant; returns empty HashMap for others.

pub fn version(&self) -> CommitVersion

Get the transaction version.

pub fn take_pending(&mut self) -> Pending

Extract pending writes, replacing them with an empty buffer.

pub fn take_pending_shapes(&mut self) -> Vec<RowShape>

Extract pending shapes, replacing them with an empty buffer.

pub fn track_flow_change(&mut self, change: Change)

Track a view-level flow change in this transaction’s accumulator.

pub fn take_accumulator_entries(&mut self) -> Vec<(ShapeId, Diff)>

Drain the accumulator entries collected during flow processing.

pub fn update_version(&mut self, new_version: CommitVersion)

Update the transaction to read at a new version

pub fn catalog(&self) -> &Catalog

Get access to the catalog for reading metadata

pub fn clock(&self) -> &Clock

Get access to the clock for timestamp generation

pub fn operator_state<S, F>( &mut self, node: FlowNodeId, load: F, ) -> Result<&mut S>

where S: 'static + Send, F: FnOnce(&mut Self) -> Result<(S, PersistFn)>,

Get or initialise the cached operator state for node.

On first access in this txn, load is invoked to build (state, persist). state is held boxed, persist is stashed for flush_operator_states. On subsequent calls, the cached state is returned directly without re-decoding, regardless of how many batches have flowed through.

Operators that mutate the returned &mut S must call mark_state_dirty(node) so the slot is persisted at flush time.

pub fn mark_state_dirty(&mut self, node: FlowNodeId)

Mark the cached state for node as dirty so it is persisted on flush. No-op if the slot does not exist (e.g. operator never touched state).

pub fn take_operator_state<S, F>( &mut self, node: FlowNodeId, load: F, ) -> Result<(S, PersistFn)>

where S: 'static + Send, F: FnOnce(&mut Self) -> Result<(S, PersistFn)>,

Take the cached state for node, returning the owned value and the existing persist closure. Used by operators whose helpers need &mut FlowTransaction alongside &mut State (e.g. Take calling parent.pull(txn, ...) while mutating its TakeState). Pair every take_operator_state with put_operator_state before returning so the slot is restored for the next batch.

pub fn put_operator_state<S>( &mut self, node: FlowNodeId, state: S, persist: PersistFn, )

where S: 'static + Send,

Restore a state slot taken by take_operator_state, marking it dirty since the operator just mutated it.

pub fn flush_operator_states(&mut self) -> Result<()>

Drain dirty slots and persist them. Must be called before take_pending so the resulting state writes are included in the pending buffer that the caller later commits.

Trait Implementations

impl WithInterceptors for FlowTransaction

fn table_row_pre_insert_interceptors( &mut self, ) -> &mut Chain<dyn TableRowPreInsertInterceptor + Send + Sync>

fn table_row_post_insert_interceptors( &mut self, ) -> &mut Chain<dyn TableRowPostInsertInterceptor + Send + Sync>

fn table_row_pre_update_interceptors( &mut self, ) -> &mut Chain<dyn TableRowPreUpdateInterceptor + Send + Sync>

fn table_row_post_update_interceptors( &mut self, ) -> &mut Chain<dyn TableRowPostUpdateInterceptor + Send + Sync>

fn table_row_pre_delete_interceptors( &mut self, ) -> &mut Chain<dyn TableRowPreDeleteInterceptor + Send + Sync>

fn table_row_post_delete_interceptors( &mut self, ) -> &mut Chain<dyn TableRowPostDeleteInterceptor + Send + Sync>

fn ringbuffer_row_pre_insert_interceptors( &mut self, ) -> &mut Chain<dyn RingBufferRowPreInsertInterceptor + Send + Sync>

fn ringbuffer_row_post_insert_interceptors( &mut self, ) -> &mut Chain<dyn RingBufferRowPostInsertInterceptor + Send + Sync>

fn ringbuffer_row_pre_update_interceptors( &mut self, ) -> &mut Chain<dyn RingBufferRowPreUpdateInterceptor + Send + Sync>

fn ringbuffer_row_post_update_interceptors( &mut self, ) -> &mut Chain<dyn RingBufferRowPostUpdateInterceptor + Send + Sync>

fn ringbuffer_row_pre_delete_interceptors( &mut self, ) -> &mut Chain<dyn RingBufferRowPreDeleteInterceptor + Send + Sync>

fn ringbuffer_row_post_delete_interceptors( &mut self, ) -> &mut Chain<dyn RingBufferRowPostDeleteInterceptor + Send + Sync>

fn pre_commit_interceptors( &mut self, ) -> &mut Chain<dyn PreCommitInterceptor + Send + Sync>

fn post_commit_interceptors( &mut self, ) -> &mut Chain<dyn PostCommitInterceptor + Send + Sync>

fn namespace_post_create_interceptors( &mut self, ) -> &mut Chain<dyn NamespacePostCreateInterceptor + Send + Sync>

fn namespace_pre_update_interceptors( &mut self, ) -> &mut Chain<dyn NamespacePreUpdateInterceptor + Send + Sync>

fn namespace_post_update_interceptors( &mut self, ) -> &mut Chain<dyn NamespacePostUpdateInterceptor + Send + Sync>

fn namespace_pre_delete_interceptors( &mut self, ) -> &mut Chain<dyn NamespacePreDeleteInterceptor + Send + Sync>

fn table_post_create_interceptors( &mut self, ) -> &mut Chain<dyn TablePostCreateInterceptor + Send + Sync>

fn table_pre_update_interceptors( &mut self, ) -> &mut Chain<dyn TablePreUpdateInterceptor + Send + Sync>

fn table_post_update_interceptors( &mut self, ) -> &mut Chain<dyn TablePostUpdateInterceptor + Send + Sync>

fn table_pre_delete_interceptors( &mut self, ) -> &mut Chain<dyn TablePreDeleteInterceptor + Send + Sync>

fn view_row_pre_insert_interceptors( &mut self, ) -> &mut Chain<dyn ViewRowPreInsertInterceptor + Send + Sync>

fn view_row_post_insert_interceptors( &mut self, ) -> &mut Chain<dyn ViewRowPostInsertInterceptor + Send + Sync>

fn view_row_pre_update_interceptors( &mut self, ) -> &mut Chain<dyn ViewRowPreUpdateInterceptor + Send + Sync>

fn view_row_post_update_interceptors( &mut self, ) -> &mut Chain<dyn ViewRowPostUpdateInterceptor + Send + Sync>

fn view_row_pre_delete_interceptors( &mut self, ) -> &mut Chain<dyn ViewRowPreDeleteInterceptor + Send + Sync>

fn view_row_post_delete_interceptors( &mut self, ) -> &mut Chain<dyn ViewRowPostDeleteInterceptor + Send + Sync>

fn view_post_create_interceptors( &mut self, ) -> &mut Chain<dyn ViewPostCreateInterceptor + Send + Sync>

fn view_pre_update_interceptors( &mut self, ) -> &mut Chain<dyn ViewPreUpdateInterceptor + Send + Sync>

fn view_post_update_interceptors( &mut self, ) -> &mut Chain<dyn ViewPostUpdateInterceptor + Send + Sync>

fn view_pre_delete_interceptors( &mut self, ) -> &mut Chain<dyn ViewPreDeleteInterceptor + Send + Sync>

fn ringbuffer_post_create_interceptors( &mut self, ) -> &mut Chain<dyn RingBufferPostCreateInterceptor + Send + Sync>

fn ringbuffer_pre_update_interceptors( &mut self, ) -> &mut Chain<dyn RingBufferPreUpdateInterceptor + Send + Sync>

fn ringbuffer_post_update_interceptors( &mut self, ) -> &mut Chain<dyn RingBufferPostUpdateInterceptor + Send + Sync>

fn ringbuffer_pre_delete_interceptors( &mut self, ) -> &mut Chain<dyn RingBufferPreDeleteInterceptor + Send + Sync>

fn dictionary_row_pre_insert_interceptors( &mut self, ) -> &mut Chain<dyn DictionaryRowPreInsertInterceptor + Send + Sync>

fn dictionary_row_post_insert_interceptors( &mut self, ) -> &mut Chain<dyn DictionaryRowPostInsertInterceptor + Send + Sync>

fn dictionary_row_pre_update_interceptors( &mut self, ) -> &mut Chain<dyn DictionaryRowPreUpdateInterceptor + Send + Sync>

fn dictionary_row_post_update_interceptors( &mut self, ) -> &mut Chain<dyn DictionaryRowPostUpdateInterceptor + Send + Sync>

fn dictionary_row_pre_delete_interceptors( &mut self, ) -> &mut Chain<dyn DictionaryRowPreDeleteInterceptor + Send + Sync>

fn dictionary_row_post_delete_interceptors( &mut self, ) -> &mut Chain<dyn DictionaryRowPostDeleteInterceptor + Send + Sync>

fn dictionary_post_create_interceptors( &mut self, ) -> &mut Chain<dyn DictionaryPostCreateInterceptor + Send + Sync>

fn dictionary_pre_update_interceptors( &mut self, ) -> &mut Chain<dyn DictionaryPreUpdateInterceptor + Send + Sync>

fn dictionary_post_update_interceptors( &mut self, ) -> &mut Chain<dyn DictionaryPostUpdateInterceptor + Send + Sync>

fn dictionary_pre_delete_interceptors( &mut self, ) -> &mut Chain<dyn DictionaryPreDeleteInterceptor + Send + Sync>

fn series_row_pre_insert_interceptors( &mut self, ) -> &mut Chain<dyn SeriesRowPreInsertInterceptor + Send + Sync>

fn series_row_post_insert_interceptors( &mut self, ) -> &mut Chain<dyn SeriesRowPostInsertInterceptor + Send + Sync>

fn series_row_pre_update_interceptors( &mut self, ) -> &mut Chain<dyn SeriesRowPreUpdateInterceptor + Send + Sync>

fn series_row_post_update_interceptors( &mut self, ) -> &mut Chain<dyn SeriesRowPostUpdateInterceptor + Send + Sync>

fn series_row_pre_delete_interceptors( &mut self, ) -> &mut Chain<dyn SeriesRowPreDeleteInterceptor + Send + Sync>

fn series_row_post_delete_interceptors( &mut self, ) -> &mut Chain<dyn SeriesRowPostDeleteInterceptor + Send + Sync>

fn series_post_create_interceptors( &mut self, ) -> &mut Chain<dyn SeriesPostCreateInterceptor + Send + Sync>

fn series_pre_update_interceptors( &mut self, ) -> &mut Chain<dyn SeriesPreUpdateInterceptor + Send + Sync>

fn series_post_update_interceptors( &mut self, ) -> &mut Chain<dyn SeriesPostUpdateInterceptor + Send + Sync>

fn series_pre_delete_interceptors( &mut self, ) -> &mut Chain<dyn SeriesPreDeleteInterceptor + Send + Sync>

fn identity_post_create_interceptors( &mut self, ) -> &mut Chain<dyn IdentityPostCreateInterceptor + Send + Sync>

fn identity_pre_update_interceptors( &mut self, ) -> &mut Chain<dyn IdentityPreUpdateInterceptor + Send + Sync>

fn identity_post_update_interceptors( &mut self, ) -> &mut Chain<dyn IdentityPostUpdateInterceptor + Send + Sync>

fn identity_pre_delete_interceptors( &mut self, ) -> &mut Chain<dyn IdentityPreDeleteInterceptor + Send + Sync>

fn role_post_create_interceptors( &mut self, ) -> &mut Chain<dyn RolePostCreateInterceptor + Send + Sync>

fn role_pre_update_interceptors( &mut self, ) -> &mut Chain<dyn RolePreUpdateInterceptor + Send + Sync>

fn role_post_update_interceptors( &mut self, ) -> &mut Chain<dyn RolePostUpdateInterceptor + Send + Sync>

fn role_pre_delete_interceptors( &mut self, ) -> &mut Chain<dyn RolePreDeleteInterceptor + Send + Sync>

fn granted_role_post_create_interceptors( &mut self, ) -> &mut Chain<dyn GrantedRolePostCreateInterceptor + Send + Sync>

fn granted_role_pre_delete_interceptors( &mut self, ) -> &mut Chain<dyn GrantedRolePreDeleteInterceptor + Send + Sync>

fn authentication_post_create_interceptors( &mut self, ) -> &mut Chain<dyn AuthenticationPostCreateInterceptor + Send + Sync>

fn authentication_pre_delete_interceptors( &mut self, ) -> &mut Chain<dyn AuthenticationPreDeleteInterceptor + Send + Sync>