Struct MaterializeExecutor

pub struct MaterializeExecutor<S: StateStore, SD: ValueRowSerde> {

    input: Executor,
    schema: Schema,
    state_table: StateTableInner<S, SD>,
    arrange_key_indices: Vec<usize>,
    actor_context: ActorContextRef,
    materialize_cache: MaterializeCache<SD>,
    conflict_behavior: ConflictBehavior,
    version_column_indices: Vec<u32>,
    may_have_downstream: bool,
    depended_subscription_ids: HashSet<u32>,
    metrics: MaterializeMetrics,
    is_dummy_table: bool,
    toastable_column_indices: Option<Vec<usize>>,
    refresh_args: Option<RefreshableMaterializeArgs<S, SD>>,
    local_barrier_manager: LocalBarrierManager,
}

MaterializeExecutor materializes changes in stream into a materialized view on storage.

Fields

input: Executor

schema: Schema

state_table: StateTableInner<S, SD>

arrange_key_indices: Vec<usize>

Columns of arrange keys (including pk, group keys, join keys, etc.)

actor_context: ActorContextRef

materialize_cache: MaterializeCache<SD>

conflict_behavior: ConflictBehavior

version_column_indices: Vec<u32>

may_have_downstream: bool

depended_subscription_ids: HashSet<u32>

metrics: MaterializeMetrics

is_dummy_table: bool

No data will be written to hummock table. This Materialize is just a dummy node. Used for APPEND ONLY table with iceberg engine. All data will be written to iceberg table directly.

toastable_column_indices: Option<Vec<usize>>

Indices of TOAST-able columns for PostgreSQL CDC tables. None means either non-CDC table or CDC table without TOAST-able columns.

refresh_args: Option<RefreshableMaterializeArgs<S, SD>>

Optional refresh arguments and state for refreshable materialized views

local_barrier_manager: LocalBarrierManager

Local barrier manager for reporting barrier events

Implementations

impl<S: StateStore, SD: ValueRowSerde> MaterializeExecutor<S, SD>

pub async fn new( input: Executor, schema: Schema, store: S, arrange_key: Vec<ColumnOrder>, actor_context: ActorContextRef, vnodes: Option<Arc<Bitmap>>, table_catalog: &Table, watermark_epoch: AtomicU64Ref, conflict_behavior: ConflictBehavior, version_column_indices: Vec<u32>, metrics: Arc<StreamingMetrics>, refresh_args: Option<RefreshableMaterializeArgs<S, SD>>, local_barrier_manager: LocalBarrierManager, ) -> Self

Create a new MaterializeExecutor with distribution specified with distribution_keys and vnodes. For singleton distribution, distribution_keys should be empty and vnodes should be None.

fn execute_inner( self, ) -> impl Stream<Item = Result<Message, StreamExecutorError>>

fn make_mergesort_stream<'a>( main_table: &'a StateTableInner<S, SD>, staging_table: &'a StateTableInner<S, SD>, progress_table: &'a mut RefreshProgressTable<S>, ) -> impl Stream<Item = Result<Option<(VirtualNode, OwnedRow)>, StreamExecutorError>> + 'a

Stream that yields rows to be deleted from main table. Yields Some((vnode, row)) for rows that exist in main but not in staging. Yields None when finished processing all vnodes.

fn may_update_depended_subscriptions( depended_subscriptions: &mut HashSet<u32>, barrier: &Barrier, mv_table_id: TableId, )

return true when changed

fn init_refresh_progress( state_table: &StateTableInner<S, SD>, progress_table: &mut RefreshProgressTable<S>, _epoch: u64, ) -> StreamExecutorResult<()>

Initialize refresh progress tracking for all VNodes

impl<S: StateStore> MaterializeExecutor<S, BasicSerde>

pub async fn for_test( input: Executor, store: S, table_id: TableId, keys: Vec<ColumnOrder>, column_ids: Vec<ColumnId>, watermark_epoch: AtomicU64Ref, conflict_behavior: ConflictBehavior, ) -> Self

Create a new MaterializeExecutor without distribution info for test purpose.

Trait Implementations

impl<S: StateStore, SD: ValueRowSerde> Debug for MaterializeExecutor<S, SD>

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

Formats the value using the given formatter.

impl<S: StateStore, SD: ValueRowSerde> Execute for MaterializeExecutor<S, SD>

fn execute(self: Box<Self>) -> BoxedMessageStream

fn execute_with_epoch(self: Box<Self>, _epoch: u64) -> BoxedMessageStream

fn boxed(self) -> Box<dyn Execute>

where Self: Sized + Send + 'static,