Struct DdlController

pub struct DdlController {
    pub(crate) env: MetaSrvEnv,
    pub(crate) metadata_manager: MetadataManager,
    pub(crate) stream_manager: GlobalStreamManagerRef,
    pub(crate) source_manager: SourceManagerRef,
    barrier_manager: BarrierManagerRef,
    pub(crate) creating_streaming_job_permits: Arc<CreatingStreamingJobPermit>,
}

Fields

env: MetaSrvEnv

metadata_manager: MetadataManager

stream_manager: GlobalStreamManagerRef

source_manager: SourceManagerRef

barrier_manager: BarrierManagerRef

creating_streaming_job_permits: Arc<CreatingStreamingJobPermit>

Implementations

impl DdlController

pub async fn new( env: MetaSrvEnv, metadata_manager: MetadataManager, stream_manager: GlobalStreamManagerRef, source_manager: SourceManagerRef, barrier_manager: BarrierManagerRef, ) -> Self

pub async fn run_command( &self, command: DdlCommand, ) -> MetaResult<Option<WaitVersion>>

run_command spawns a tokio coroutine to execute the target ddl command. When the client has been interrupted during executing, the request will be cancelled by tonic. Since we have a lot of logic for revert, status management, notification and so on, ensuring consistency would be a huge hassle and pain if we don’t spawn here.

Though returning Option, it’s always Some, to simplify the handling logic

pub async fn get_ddl_progress(&self) -> MetaResult<Vec<DdlProgress>>

async fn create_database( &self, database: Database, ) -> MetaResult<NotificationVersion>

pub async fn alter_parallelism( &self, table_id: u32, parallelism: PbTableParallelism, deferred: bool, ) -> MetaResult<()>

async fn drop_database( &self, database_id: DatabaseId, ) -> MetaResult<NotificationVersion>

async fn create_schema(&self, schema: Schema) -> MetaResult<NotificationVersion>

async fn drop_schema( &self, schema_id: SchemaId, ) -> MetaResult<NotificationVersion>

async fn create_source_without_streaming_job( &self, source: Source, ) -> MetaResult<NotificationVersion>

Shared source is handled in Self::create_streaming_job

async fn drop_source( &self, source_id: SourceId, drop_mode: DropMode, ) -> MetaResult<NotificationVersion>

async fn alter_source(&self, source: Source) -> MetaResult<NotificationVersion>

This replaces the source in the catalog. Note: StreamSourceInfo in downstream MVs’ SourceExecutors are not updated.

async fn create_function( &self, function: Function, ) -> MetaResult<NotificationVersion>

async fn drop_function( &self, function_id: FunctionId, ) -> MetaResult<NotificationVersion>

async fn create_view(&self, view: View) -> MetaResult<NotificationVersion>

async fn drop_view( &self, view_id: ViewId, drop_mode: DropMode, ) -> MetaResult<NotificationVersion>

async fn create_connection( &self, connection: Connection, ) -> MetaResult<NotificationVersion>

async fn drop_connection( &self, connection_id: ConnectionId, ) -> MetaResult<NotificationVersion>

async fn create_secret(&self, secret: Secret) -> MetaResult<NotificationVersion>

async fn drop_secret( &self, secret_id: SecretId, ) -> MetaResult<NotificationVersion>

async fn create_subscription( &self, subscription: Subscription, ) -> MetaResult<NotificationVersion>

async fn drop_subscription( &self, subscription_id: SubscriptionId, drop_mode: DropMode, ) -> MetaResult<NotificationVersion>

pub(crate) async fn validate_cdc_table( table: &Table, table_fragments: &TableFragments, ) -> MetaResult<()>

Validates the connect properties in the cdc_table_desc stored in the StreamCdcScan node

pub(crate) async fn inject_replace_table_job_for_table_sink( &self, tmp_id: u32, mgr: &MetadataManager, stream_ctx: StreamContext, sink: Option<&Sink>, creating_sink_table_fragments: Option<&TableFragments>, dropping_sink_id: Option<SinkId>, streaming_job: &StreamingJob, fragment_graph: StreamFragmentGraph, ) -> MetaResult<(ReplaceTableContext, TableFragments)>

pub(crate) fn inject_replace_table_plan_for_sink( sink_id: Option<u32>, sink_fragment: &PbFragment, table: &Table, replace_table_ctx: &mut ReplaceTableContext, table_fragments: &mut TableFragments, target_fragment_id: FragmentId, unique_identity: Option<&str>, )

pub async fn create_streaming_job( &self, streaming_job: StreamingJob, fragment_graph: StreamFragmentGraphProto, affected_table_replace_info: Option<ReplaceTableInfo>, dependencies: HashSet<ObjectId>, ) -> MetaResult<NotificationVersion>

For CreateType::Foreground, the function will only return after backfilling finishes (crate::manager::MetadataManager::wait_streaming_job_finished).

async fn create_streaming_job_inner( &self, ctx: StreamContext, streaming_job: StreamingJob, fragment_graph: StreamFragmentGraphProto, affected_table_replace_info: Option<ReplaceTableInfo>, ) -> MetaResult<NotificationVersion>

pub async fn drop_object( &self, object_type: ObjectType, object_id: ObjectId, drop_mode: DropMode, target_replace_info: Option<ReplaceTableInfo>, ) -> MetaResult<NotificationVersion>

pub async fn replace_table( &self, streaming_job: StreamingJob, fragment_graph: StreamFragmentGraphProto, table_col_index_mapping: Option<ColIndexMapping>, ) -> MetaResult<NotificationVersion>

This is used for ALTER TABLE ADD/DROP COLUMN.

async fn drop_streaming_job( &self, job_id: StreamingJobId, drop_mode: DropMode, target_replace_info: Option<ReplaceTableInfo>, ) -> MetaResult<NotificationVersion>

fn resolve_stream_parallelism( &self, specified: Option<NonZeroUsize>, max: NonZeroUsize, cluster_info: &StreamingClusterInfo, ) -> MetaResult<NonZeroUsize>

Resolve the parallelism of the stream job based on the given information.

Returns error if user specifies a parallelism that cannot be satisfied.

pub(crate) async fn build_stream_job( &self, stream_ctx: StreamContext, stream_job: StreamingJob, fragment_graph: StreamFragmentGraph, affected_table_replace_info: Option<(StreamingJob, StreamFragmentGraph)>, ) -> MetaResult<(CreateStreamingJobContext, TableFragments)>

Builds the actor graph:

• Add the upstream fragments to the fragment graph
• Schedule the fragments based on their distribution
• Expand each fragment into one or several actors

pub(crate) async fn build_replace_table( &self, stream_ctx: StreamContext, stream_job: &StreamingJob, fragment_graph: StreamFragmentGraph, table_col_index_mapping: Option<ColIndexMapping>, tmp_table_id: TableId, ) -> MetaResult<(ReplaceTableContext, TableFragments)>

build_replace_table builds a table replacement and returns the context and new table fragments.

Note that we use a dummy ID for the new table fragments and replace it with the real one after replacement is finished.

async fn alter_name( &self, relation: Object, new_name: &str, ) -> MetaResult<NotificationVersion>

async fn alter_swap_rename( &self, object: Object, ) -> MetaResult<NotificationVersion>

async fn alter_owner( &self, object: Object, owner_id: UserId, ) -> MetaResult<NotificationVersion>

async fn alter_set_schema( &self, object: Object, new_schema_id: SchemaId, ) -> MetaResult<NotificationVersion>

pub async fn wait(&self) -> MetaResult<()>

async fn comment_on(&self, comment: Comment) -> MetaResult<NotificationVersion>

Trait Implementations

impl Clone for DdlController

fn clone(&self) -> DdlController

Returns a copy of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.