Struct SourceExecutor

pub struct SourceExecutor<S: StateStore> {
    actor_ctx: ActorContextRef,
    stream_source_core: Option<StreamSourceCore<S>>,
    metrics: Arc<StreamingMetrics>,
    barrier_receiver: Option<UnboundedReceiver<Barrier>>,
    system_params: SystemParamsReaderRef,
    rate_limit_rps: Option<u32>,
    is_shared_non_cdc: bool,
}

Fields

actor_ctx: ActorContextRef

stream_source_core: Option<StreamSourceCore<S>>

Streaming source for external

metrics: Arc<StreamingMetrics>

Metrics for monitor.

barrier_receiver: Option<UnboundedReceiver<Barrier>>

Receiver of barrier channel.

system_params: SystemParamsReaderRef

System parameter reader to read barrier interval

rate_limit_rps: Option<u32>

Rate limit in rows/s.

is_shared_non_cdc: bool

Implementations

impl<S: StateStore> SourceExecutor<S>

pub fn new( actor_ctx: ActorContextRef, stream_source_core: Option<StreamSourceCore<S>>, metrics: Arc<StreamingMetrics>, barrier_receiver: UnboundedReceiver<Barrier>, system_params: SystemParamsReaderRef, rate_limit_rps: Option<u32>, is_shared_non_cdc: bool, ) -> Self

async fn spawn_wait_checkpoint_worker( core: &StreamSourceCore<S>, source_reader: SourceReader, ) -> StreamExecutorResult<Option<WaitCheckpointTaskBuilder>>

pub async fn build_stream_source_reader( &self, source_desc: &SourceDesc, state: ConnectorState, seek_to_latest: bool, ) -> StreamExecutorResult<(BoxChunkSourceStream, Option<Vec<SplitImpl>>)>

If seek_to_latest is true, will also return the latest splits after seek.

fn is_auto_schema_change_enable(&self) -> bool

fn get_metric_labels(&self) -> [String; 4]

source_id | source_name | actor_id | fragment_id

async fn apply_split_change<const BIASED: bool>( &mut self, source_desc: &SourceDesc, stream: &mut StreamReaderWithPause<BIASED, StreamChunk>, split_assignment: &HashMap<ActorId, Vec<SplitImpl>>, should_trim_state: bool, source_split_change_count_metrics: &LabelGuardedIntCounter<4>, ) -> StreamExecutorResult<()>

• should_trim_state: whether to trim state for dropped splits.

  For scaling, the connector splits can be migrated to other actors, but won’t be added or removed. Actors should not trim states for splits that are moved to other actors.

  For source split change, split will not be migrated and we can trim states for deleted splits.

async fn update_state_if_changed( &mut self, target_splits: Vec<SplitImpl>, should_trim_state: bool, ) -> StreamExecutorResult<bool>

Returns true if split changed. Otherwise false.

async fn rebuild_stream_reader_from_error<const BIASED: bool>( &mut self, source_desc: &SourceDesc, stream: &mut StreamReaderWithPause<BIASED, StreamChunk>, e: StreamExecutorError, ) -> StreamExecutorResult<()>

Rebuild stream if there is a err in stream

async fn rebuild_stream_reader<const BIASED: bool>( &mut self, source_desc: &SourceDesc, stream: &mut StreamReaderWithPause<BIASED, StreamChunk>, ) -> StreamExecutorResult<()>

async fn persist_state_and_clear_cache( &mut self, epoch: EpochPair, ) -> StreamExecutorResult<HashMap<SplitId, SplitImpl>>

async fn try_flush_data(&mut self) -> StreamExecutorResult<()>

try mem table spill

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

A source executor with a stream source receives:

1. Barrier messages
2. Data from external source and acts accordingly.

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

A source executor without stream source only receives barrier messages and sends them to the downstream executor.

Trait Implementations

impl<S: StateStore> Debug for SourceExecutor<S>

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

Formats the value using the given formatter.

impl<S: StateStore> Execute for SourceExecutor<S>

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,