Struct BatchIcebergFetchExecutor 

pub struct BatchIcebergFetchExecutor<S: StateStore> {
    actor_ctx: ActorContextRef,
    stream_source_core: Option<StreamSourceCore<S>>,
    upstream: Option<Executor>,
    barrier_manager: LocalBarrierManager,
    streaming_config: Arc<StreamingConfig>,
    associated_table_id: TableId,
}

Executor that fetches data from Iceberg files discovered by an upstream list executor.

Refresh Cycle

1. Receives RefreshStart mutation - clears state and starts new cycle
2. Receives file chunks from upstream list executor - queues files for processing
3. On each barrier, starts batch reader if files are pending
4. Receives ListFinish mutation - marks listing as complete
5. When all files processed, reports load finished

At-Least-Once Semantics

On fetch errors, in-flight files are re-queued to ensure no file is skipped. This may cause duplicate reads, but guarantees data completeness.

Fields

actor_ctx: ActorContextRef

stream_source_core: Option<StreamSourceCore<S>>

Core component for managing external streaming source state

upstream: Option<Executor>

Upstream list executor that provides file scan tasks

barrier_manager: LocalBarrierManager

Barrier manager for reporting load finished

streaming_config: Arc<StreamingConfig>

associated_table_id: TableId

Implementations

impl<S: StateStore> BatchIcebergFetchExecutor<S>

pub fn new( actor_ctx: ActorContextRef, stream_source_core: StreamSourceCore<S>, upstream: Executor, barrier_manager: LocalBarrierManager, streaming_config: Arc<StreamingConfig>, associated_table_id: Option<TableId>, ) -> Self

impl<S: StateStore> BatchIcebergFetchExecutor<S>

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

fn handle_barrier_mutations( barrier: &Barrier, core: &StreamSourceCore<S>, state: &mut FetchState, stream: &mut StreamReaderWithPause<true, ChunksWithState>, ) -> bool

Handle barrier mutations and return whether reader needs to be rebuilt.

fn parse_file_assignments(chunk: &StreamChunk) -> Vec<(String, JsonbVal)>

Parse file assignments from an upstream chunk.

fn start_batch_reader( state: &mut FetchState, stream: &mut StreamReaderWithPause<true, ChunksWithState>, source_desc: SourceDesc, streaming_config: &StreamingConfig, ) -> StreamExecutorResult<()>

Start a new batch reader for pending files.

fn build_batched_stream_reader( source_desc: SourceDesc, tasks: Vec<FileScanTask>, chunk_size: usize, batch_finished: Arc<RwLock<bool>>, ) -> impl Stream<Item = Result<ChunksWithState, StreamExecutorError>>

Build a stream reader that reads multiple Iceberg files in sequence.

Trait Implementations

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

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

Formats the value using the given formatter.

impl<S: StateStore> Execute for BatchIcebergFetchExecutor<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,