Struct RegionWorkerLoop

struct RegionWorkerLoop<S> {

    id: u32,
    config: Arc<MitoConfig>,
    regions: Arc<RegionMap>,
    dropping_regions: Arc<RegionMap>,
    opening_regions: Arc<OpeningRegions>,
    sender: Sender<WorkerRequest>,
    receiver: Receiver<WorkerRequest>,
    wal: Wal<S>,
    object_store_manager: ObjectStoreManagerRef,
    running: Arc<AtomicBool>,
    memtable_builder_provider: MemtableBuilderProvider,
    purge_scheduler: SchedulerRef,
    write_buffer_manager: WriteBufferManagerRef,
    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<()>,
    stalled_count: IntGauge,
    region_count: IntGauge,
    region_edit_queues: HashMap<RegionId, RegionEditQueue>,
    schema_metadata_manager: SchemaMetadataManagerRef,
}

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.

sender: Sender<WorkerRequest>

Request sender.

receiver: Receiver<WorkerRequest>

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.

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.

stalled_count: IntGauge

Gauge of stalled request count.

region_count: IntGauge

Gauge of regions in the worker.

region_edit_queues: HashMap<RegionId, RegionEditQueue>

Queues for region edit requests.

schema_metadata_manager: SchemaMetadataManagerRef

Database level metadata manager.

Implementations

impl<S> RegionWorkerLoop<S>

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

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

Handles region metadata changes.

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

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

impl<S: LogStore> RegionWorkerLoop<S>

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

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

Reopens a region.

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: &Arc<MitoRegion>)

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> RegionWorkerLoop<S>

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

Handles manual flush request.

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 flush_periodically(&mut self) -> Result<()>

Flushes regions periodically.

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

Creates a flush task with specific reason for the region.

impl<S: LogStore> RegionWorkerLoop<S>

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

On region flush job finished.

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: Arc<MitoRegion>, 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: Arc<MitoRegion>, change: RegionChange, sender: OptionOutputTx, )

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

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: LogStore> RegionWorkerLoop<S>

pub(crate) async fn handle_truncate_request( &mut self, region_id: RegionId, 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>, 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>, ) -> HashMap<RegionId, RegionWriteCtx>

Validates and groups requests by region.

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>, )

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.