Struct RegionWorkerLoop 

struct RegionWorkerLoop<S> {

    id: u32,
    config: Arc<MitoConfig>,
    regions: Arc<RegionMap>,
    dropping_regions: Arc<RegionMap>,
    opening_regions: Arc<OpeningRegions>,
    catchup_regions: Arc<CatchupRegions>,
    sender: Sender<WorkerRequestWithTime>,
    receiver: Receiver<WorkerRequestWithTime>,
    wal: Wal<S>,
    object_store_manager: ObjectStoreManagerRef,
    running: Arc<AtomicBool>,
    memtable_builder_provider: MemtableBuilderProvider,
    purge_scheduler: SchedulerRef,
    write_buffer_manager: WriteBufferManagerRef,
    index_build_scheduler: IndexBuildScheduler,
    flush_scheduler: FlushScheduler,
    compaction_scheduler: CompactionScheduler,
    stalled_requests: StalledRequests,
    listener: WorkerListener,
    cache_manager: CacheManagerRef,
    puffin_manager_factory: PuffinManagerFactory,
    intermediate_manager: IntermediateManager,
    time_provider: Arc<dyn TimeProvider>,
    last_periodical_check_millis: i64,
    flush_sender: Sender<()>,
    flush_receiver: Receiver<()>,
    stalling_count: IntGauge,
    region_count: IntGauge,
    request_wait_time: Histogram,
    region_edit_queues: HashMap<RegionId, RegionEditQueue>,
    schema_metadata_manager: SchemaMetadataManagerRef,
    file_ref_manager: FileReferenceManagerRef,
    partition_expr_fetcher: PartitionExprFetcherRef,
    flush_semaphore: Arc<Semaphore>,
}

Background worker loop to handle requests.

Fields

id: u32

Id of the worker.

config: Arc<MitoConfig>

Engine config.

regions: Arc<RegionMap>

Regions bound to the worker.

dropping_regions: Arc<RegionMap>

Regions that are not yet fully dropped.

opening_regions: Arc<OpeningRegions>

Regions that are opening.

catchup_regions: Arc<CatchupRegions>

Regions that are catching up.

sender: Sender<WorkerRequestWithTime>

Request sender.

receiver: Receiver<WorkerRequestWithTime>

Request receiver.

wal: Wal<S>

WAL of the engine.

object_store_manager: ObjectStoreManagerRef

Manages object stores for manifest and SSTs.

running: Arc<AtomicBool>

Whether the worker thread is still running.

memtable_builder_provider: MemtableBuilderProvider

Memtable builder provider for each region.

purge_scheduler: SchedulerRef

Background purge job scheduler.

write_buffer_manager: WriteBufferManagerRef

Engine write buffer manager.

index_build_scheduler: IndexBuildScheduler

Scheduler for index build task.

flush_scheduler: FlushScheduler

Schedules background flush requests.

compaction_scheduler: CompactionScheduler

Scheduler for compaction tasks.

stalled_requests: StalledRequests

Stalled write requests.

listener: WorkerListener

Event listener for tests.

cache_manager: CacheManagerRef

Cache.

puffin_manager_factory: PuffinManagerFactory

Puffin manager factory for index.

intermediate_manager: IntermediateManager

Intermediate manager for inverted index.

time_provider: Arc<dyn TimeProvider>

Provider to get current time.

last_periodical_check_millis: i64

Last time to check regions periodically.

flush_sender: Sender<()>

Watch channel sender to notify workers to handle stalled requests.

flush_receiver: Receiver<()>

Watch channel receiver to wait for background flush job.

stalling_count: IntGauge

Gauge of stalling request count.

region_count: IntGauge

Gauge of regions in the worker.

request_wait_time: Histogram

Histogram of request wait time for this worker.

region_edit_queues: HashMap<RegionId, RegionEditQueue>

Queues for region edit requests.

schema_metadata_manager: SchemaMetadataManagerRef

Database level metadata manager.

file_ref_manager: FileReferenceManagerRef

Datanode level file references manager.

partition_expr_fetcher: PartitionExprFetcherRef

Partition expr fetcher used to backfill partition expr on open for compatibility.

flush_semaphore: Arc<Semaphore>

Semaphore to control flush concurrency.

Implementations

impl<S: LogStore> RegionWorkerLoop<S>

pub(crate) async fn handle_alter_request( &mut self, region_id: RegionId, request: RegionAlterRequest, sender: OptionOutputTx, )

fn handle_alter_region_with_empty_memtable( &mut self, region: MitoRegionRef, version: Arc<Version>, request: RegionAlterRequest, sender: OptionOutputTx, )

Handles region metadata and format changes when the region memtable is empty.

fn handle_alter_region_options_fast( &mut self, region: &MitoRegionRef, version: Arc<Version>, options: Vec<SetRegionOption>, ) -> Result<Option<Arc<Version>>, MetadataError>

Handles requests that changes region options, like TTL. It only affects memory state since changes are persisted in the DatanodeTableValue in metasrv.

If the options require empty memtable, it only does validation.

Returns a new version with the updated options if it needs further alteration.

impl<S: LogStore> RegionWorkerLoop<S>

pub(crate) async fn handle_bulk_insert_batch( &mut self, region_metadata: RegionMetadataRef, request: RegionBulkInsertsRequest, pending_bulk_request: &mut Vec<SenderBulkRequest>, sender: OptionOutputTx, )

impl<S: LogStore> RegionWorkerLoop<S>

pub(crate) async fn handle_catchup_request( &mut self, region_id: RegionId, request: RegionCatchupRequest, entry_receiver: Option<WalEntryReceiver>, sender: OptionOutputTx, )

pub(crate) async fn reopen_region( &mut self, region: &Arc<MitoRegion>, ) -> Result<Arc<MitoRegion>>

Reopens a region.

async fn reopen_region_if_needed( &mut self, region: Arc<MitoRegion>, ) -> Result<Arc<MitoRegion>>

impl<S> RegionWorkerLoop<S>

pub(crate) async fn handle_close_request( &mut self, region_id: RegionId, ) -> Result<AffectedRows>

impl<S> RegionWorkerLoop<S>

pub(crate) async fn handle_compaction_request( &mut self, region_id: RegionId, req: RegionCompactRequest, sender: OptionOutputTx, )

Handles compaction request submitted to region worker.

pub(crate) async fn handle_compaction_finished( &mut self, region_id: RegionId, request: CompactionFinished, )

Handles compaction finished, update region version and manifest, deleted compacted files.

pub(crate) async fn handle_compaction_failure(&mut self, req: CompactionFailed)

When compaction fails, we simply log the error.

pub(crate) async fn schedule_compaction(&mut self, region: &MitoRegionRef)

Schedule compaction for the region if necessary.

impl<S: LogStore> RegionWorkerLoop<S>

pub(crate) async fn handle_create_request( &mut self, region_id: RegionId, request: RegionCreateRequest, ) -> Result<AffectedRows>

impl<S> RegionWorkerLoop<S>

where S: LogStore,

pub(crate) async fn handle_drop_request( &mut self, region_id: RegionId, ) -> Result<AffectedRows>

impl<S: LogStore> RegionWorkerLoop<S>

pub(crate) async fn handle_enter_staging_request( &mut self, region_id: RegionId, partition_expr: String, sender: OptionOutputTx, )

async fn enter_staging( region: &MitoRegionRef, partition_expr: String, ) -> Result<()>

fn handle_enter_staging( &self, region: MitoRegionRef, partition_expr: String, sender: OptionOutputTx, )

pub(crate) async fn handle_enter_staging_result( &mut self, enter_staging_result: EnterStagingResult, )

Handles enter staging result.

fn update_region_staging_partition_expr( region: &MitoRegionRef, partition_expr: String, )

impl<S: LogStore> RegionWorkerLoop<S>

pub(crate) async fn handle_flush_failed( &mut self, region_id: RegionId, request: FlushFailed, )

On region flush job failed.

pub(crate) fn maybe_flush_worker(&mut self)

Checks whether the engine reaches flush threshold. If so, finds regions in this worker to flush.

fn flush_regions_on_engine_full(&mut self) -> Result<()>

Finds some regions to flush to reduce write buffer usage.

pub(crate) fn new_flush_task( &self, region: &MitoRegionRef, reason: FlushReason, row_group_size: Option<usize>, engine_config: Arc<MitoConfig>, is_staging: bool, ) -> RegionFlushTask

Creates a flush task with specific reason for the region.

impl<S: LogStore> RegionWorkerLoop<S>

pub(crate) fn handle_flush_request( &mut self, region_id: RegionId, request: RegionFlushRequest, sender: OptionOutputTx, )

Handles manual flush request.

pub(crate) fn flush_periodically(&mut self) -> Result<()>

Flushes regions periodically.

pub(crate) async fn handle_flush_finished( &mut self, region_id: RegionId, request: FlushFinished, )

On region flush job finished.

pub(crate) fn update_topic_latest_entry_id(&mut self, region: &MitoRegionRef)

Updates the latest entry id since flush of the region. This is only used for remote WAL pruning.

impl<S: LogStore> RegionWorkerLoop<S>

pub(crate) async fn handle_manifest_region_change_result( &mut self, change_result: RegionChangeResult, )

Handles region change result.

pub(crate) async fn handle_region_sync(&mut self, request: RegionSyncRequest)

Handles region sync request.

Updates the manifest to at least the given version. Note: The installed version may be greater than the given version.

impl<S> RegionWorkerLoop<S>

pub(crate) async fn handle_region_edit(&mut self, request: RegionEditRequest)

Handles region edit request.

pub(crate) async fn handle_region_edit_result( &mut self, edit_result: RegionEditResult, )

Handles region edit result.

pub(crate) fn handle_manifest_truncate_action( &self, region: MitoRegionRef, truncate: RegionTruncate, sender: OptionOutputTx, )

Writes truncate action to the manifest and then applies it to the region in background.

pub(crate) fn handle_manifest_region_change( &self, region: MitoRegionRef, change: RegionChange, need_index: bool, new_options: Option<RegionOptions>, sender: OptionOutputTx, )

Writes region change action to the manifest and then applies it to the region in background.

fn update_region_version( version_control: &Arc<VersionControl>, new_meta: RegionMetadataRef, new_options: Option<RegionOptions>, memtable_builder_provider: &MemtableBuilderProvider, )

impl<S: LogStore> RegionWorkerLoop<S>

async fn check_and_cleanup_region( &self, region_id: RegionId, request: &RegionOpenRequest, ) -> Result<()>

pub(crate) async fn handle_open_request( &mut self, region_id: RegionId, request: RegionOpenRequest, wal_entry_receiver: Option<WalEntryReceiver>, sender: OptionOutputTx, )

impl<S> RegionWorkerLoop<S>

pub(crate) fn new_index_build_task( &self, region: &MitoRegionRef, file: FileHandle, build_type: IndexBuildType, result_sender: ResultMpscSender, ) -> IndexBuildTask

pub(crate) async fn handle_build_index_request( &mut self, region_id: RegionId, _req: RegionBuildIndexRequest, sender: OptionOutputTx, )

Handles manual build index requests.

pub(crate) async fn handle_rebuild_index( &mut self, request: BuildIndexRequest, sender: OptionOutputTx, )

pub(crate) async fn handle_index_build_finished( &mut self, region_id: RegionId, request: IndexBuildFinished, )

pub(crate) async fn handle_index_build_failed( &mut self, region_id: RegionId, request: IndexBuildFailed, )

pub(crate) async fn handle_index_build_stopped( &mut self, region_id: RegionId, request: IndexBuildStopped, )

impl<S> RegionWorkerLoop<S>

pub(crate) fn handle_remap_manifests_request( &mut self, request: RemapManifestsRequest, )

async fn fetch_and_remap_manifests( region: MitoRegionRef, region_metadata_loader: RegionMetadataLoader, input_regions: Vec<RegionId>, new_partition_exprs: HashMap<RegionId, PartitionExpr>, region_mapping: HashMap<RegionId, Vec<RegionId>>, ) -> Result<HashMap<RegionId, RegionManifest>>

impl<S: LogStore> RegionWorkerLoop<S>

pub(crate) async fn handle_truncate_request( &mut self, region_id: RegionId, req: RegionTruncateRequest, sender: OptionOutputTx, )

pub(crate) async fn handle_truncate_result( &mut self, truncate_result: TruncateResult, )

Handles truncate result.

impl<S: LogStore> RegionWorkerLoop<S>

pub(crate) async fn handle_write_requests( &mut self, write_requests: &mut Vec<SenderWriteRequest>, bulk_requests: &mut Vec<SenderBulkRequest>, allow_stall: bool, )

Takes and handles all write requests.

pub(crate) async fn handle_stalled_requests(&mut self)

Handles all stalled write requests.

pub(crate) fn reject_stalled_requests(&mut self)

Rejects all stalled requests.

pub(crate) fn reject_region_stalled_requests(&mut self, region_id: &RegionId)

Rejects a specific region’s stalled requests.

pub(crate) async fn handle_region_stalled_requests( &mut self, region_id: &RegionId, )

Handles a specific region’s stalled requests.

impl<S> RegionWorkerLoop<S>

fn prepare_region_write_ctx( &mut self, write_requests: &mut Vec<SenderWriteRequest>, bulk_requests: &mut Vec<SenderBulkRequest>, ) -> HashMap<RegionId, RegionWriteCtx>

Validates and groups requests by region.

fn process_write_requests( &mut self, region_ctxs: &mut HashMap<RegionId, RegionWriteCtx>, write_requests: &mut Vec<SenderWriteRequest>, )

fn process_bulk_requests( &mut self, region_ctxs: &mut HashMap<RegionId, RegionWriteCtx>, requests: &mut Vec<SenderBulkRequest>, )

Processes bulk insert requests.

pub(crate) fn should_reject_write(&self) -> bool

Returns true if the engine needs to reject some write requests.

impl<S: LogStore> RegionWorkerLoop<S>

async fn run(&mut self)

Starts the worker loop.

async fn handle_requests( &mut self, write_requests: &mut Vec<SenderWriteRequest>, ddl_requests: &mut Vec<SenderDdlRequest>, general_requests: &mut Vec<WorkerRequest>, bulk_requests: &mut Vec<SenderBulkRequest>, )

Dispatches and processes requests.

buffer should be empty.

async fn handle_ddl_requests( &mut self, ddl_requests: &mut Vec<SenderDdlRequest>, )

Takes and handles all ddl requests.

fn handle_periodical_tasks(&mut self)

Handle periodical tasks such as region auto flush.

async fn handle_background_notify( &mut self, region_id: RegionId, notify: BackgroundNotify, )

Handles region background request

async fn set_role_state_gracefully( &mut self, region_id: RegionId, region_role_state: SettableRegionRoleState, sender: Sender<SetRegionRoleStateResponse>, )

Handles set_region_role_gracefully.

impl<S> RegionWorkerLoop<S>

async fn clean(&self)

Cleans up the worker.

fn notify_group(&mut self)

Notifies the whole group that a flush job is finished so other workers can handle stalled requests.