Struct StatementExecutor 

pub struct StatementExecutor {
    catalog_manager: CatalogManagerRef,
    query_engine: QueryEngineRef,
    procedure_executor: ProcedureExecutorRef,
    table_metadata_manager: TableMetadataManagerRef,
    flow_metadata_manager: FlowMetadataManagerRef,
    view_info_manager: ViewInfoManagerRef,
    partition_manager: PartitionRuleManagerRef,
    cache_invalidator: CacheInvalidatorRef,
    inserter: InserterRef,
    process_manager: Option<ProcessManagerRef>,
}

Fields

catalog_manager: CatalogManagerRef

query_engine: QueryEngineRef

procedure_executor: ProcedureExecutorRef

table_metadata_manager: TableMetadataManagerRef

flow_metadata_manager: FlowMetadataManagerRef

view_info_manager: ViewInfoManagerRef

partition_manager: PartitionRuleManagerRef

cache_invalidator: CacheInvalidatorRef

inserter: InserterRef

process_manager: Option<ProcessManagerRef>

Implementations

impl StatementExecutor

pub(super) async fn execute_admin_command( &self, stmt: Admin, query_ctx: QueryContextRef, ) -> Result<Output>

Execute the Admin statement and returns the output.

impl StatementExecutor

pub(crate) async fn copy_database_to( &self, req: CopyDatabaseRequest, ctx: QueryContextRef, ) -> Result<Output>

pub(crate) async fn copy_database_from( &self, req: CopyDatabaseRequest, ctx: QueryContextRef, ) -> Result<Output>

Imports data to database from a given location and returns total rows imported.

impl StatementExecutor

pub(crate) async fn copy_query_to( &self, req: CopyQueryToRequest, query_output: Output, ) -> Result<usize>

impl StatementExecutor

async fn list_copy_from_entries( &self, req: &CopyTableRequest, ) -> Result<(ObjectStore, Vec<Entry>)>

async fn collect_metadata( &self, object_store: &ObjectStore, format: Format, path: String, ) -> Result<FileMetadata>

async fn build_file_stream( &self, store: &ObjectStore, filename: &str, file_schema: SchemaRef, file_source: Arc<dyn FileSource>, projection: Option<Vec<usize>>, ) -> Result<DfSendableRecordBatchStream>

async fn build_read_stream( &self, compat_schema: SchemaRef, object_store: &ObjectStore, file_metadata: &FileMetadata, projection: Vec<usize>, filters: Vec<Expr>, ) -> Result<DfSendableRecordBatchStream>

pub async fn copy_table_from( &self, req: CopyTableRequest, query_ctx: QueryContextRef, ) -> Result<Output>

impl StatementExecutor

async fn stream_to_file( &self, stream: SendableRecordBatchStream, format: &Format, object_store: ObjectStore, path: &str, ) -> Result<usize>

pub(crate) async fn copy_table_to( &self, req: CopyTableRequest, query_ctx: QueryContextRef, ) -> Result<usize>

pub(crate) async fn copy_to_file( &self, format: &Format, output: Output, location: &str, connection: &HashMap<String, String>, ) -> Result<usize>

impl StatementExecutor

pub(super) async fn declare_cursor( &self, declare_cursor: DeclareCursor, query_ctx: QueryContextRef, ) -> Result<Output>

pub(super) async fn fetch_cursor( &self, fetch_cursor: FetchCursor, query_ctx: QueryContextRef, ) -> Result<Output>

pub(super) async fn close_cursor( &self, close_cursor: CloseCursor, query_ctx: QueryContextRef, ) -> Result<Output>

impl StatementExecutor

pub fn catalog_manager(&self) -> CatalogManagerRef

pub async fn create_table( &self, stmt: CreateTable, ctx: QueryContextRef, ) -> Result<TableRef>

pub async fn create_table_like( &self, stmt: CreateTableLike, ctx: QueryContextRef, ) -> Result<TableRef>

pub async fn create_external_table( &self, create_expr: CreateExternalTable, ctx: QueryContextRef, ) -> Result<TableRef>

pub async fn create_table_inner( &self, create_table: &mut CreateTableExpr, partitions: Option<Partitions>, query_ctx: QueryContextRef, ) -> Result<TableRef>

pub async fn create_non_logic_table( &self, create_table: &mut CreateTableExpr, partitions: Option<Partitions>, query_ctx: QueryContextRef, ) -> Result<TableRef>

pub async fn create_logical_tables( &self, create_table_exprs: &[CreateTableExpr], query_context: QueryContextRef, ) -> Result<Vec<TableRef>>

pub async fn create_flow( &self, stmt: CreateFlow, query_context: QueryContextRef, ) -> Result<Output>

pub async fn create_flow_inner( &self, expr: CreateFlowExpr, query_context: QueryContextRef, ) -> Result<Output>

async fn create_flow_procedure( &self, expr: CreateFlowExpr, query_context: QueryContextRef, ) -> Result<SubmitDdlTaskResponse>

async fn determine_flow_type( &self, expr: &CreateFlowExpr, query_ctx: QueryContextRef, ) -> Result<FlowType>

Determine the flow type based on the SQL query

If it contains aggregation or distinct, then it is a batch flow, otherwise it is a streaming flow

pub async fn create_view( &self, create_view: CreateView, ctx: QueryContextRef, ) -> Result<TableRef>

pub async fn create_view_by_expr( &self, expr: CreateViewExpr, ctx: QueryContextRef, ) -> Result<TableRef>

pub async fn drop_flow( &self, catalog_name: String, flow_name: String, drop_if_exists: bool, query_context: QueryContextRef, ) -> Result<Output>

async fn drop_flow_procedure( &self, expr: DropFlowTask, query_context: QueryContextRef, ) -> Result<SubmitDdlTaskResponse>

pub(crate) async fn drop_view( &self, catalog: String, schema: String, view: String, drop_if_exists: bool, query_context: QueryContextRef, ) -> Result<Output>

Drop a view

async fn drop_view_procedure( &self, expr: DropViewTask, query_context: QueryContextRef, ) -> Result<SubmitDdlTaskResponse>

Submit DropViewTask to procedure executor.

pub async fn alter_logical_tables( &self, alter_table_exprs: Vec<AlterTableExpr>, query_context: QueryContextRef, ) -> Result<Output>

pub async fn drop_table( &self, table_name: TableName, drop_if_exists: bool, query_context: QueryContextRef, ) -> Result<Output>

pub async fn drop_tables( &self, table_names: &[TableName], drop_if_exists: bool, query_context: QueryContextRef, ) -> Result<Output>

pub async fn drop_database( &self, catalog: String, schema: String, drop_if_exists: bool, query_context: QueryContextRef, ) -> Result<Output>

pub async fn truncate_table( &self, table_name: TableName, time_ranges: Vec<(Timestamp, Timestamp)>, query_context: QueryContextRef, ) -> Result<Output>

pub async fn alter_table( &self, alter_table: AlterTable, query_context: QueryContextRef, ) -> Result<Output>

pub async fn alter_table_inner( &self, expr: AlterTableExpr, query_context: QueryContextRef, ) -> Result<Output>

pub async fn alter_database( &self, alter_expr: AlterDatabase, query_context: QueryContextRef, ) -> Result<Output>

pub async fn alter_database_inner( &self, alter_expr: AlterDatabaseExpr, query_context: QueryContextRef, ) -> Result<Output>

async fn create_table_procedure( &self, create_table: CreateTableExpr, partitions: Vec<PartitionExpr>, table_info: RawTableInfo, query_context: QueryContextRef, ) -> Result<SubmitDdlTaskResponse>

async fn create_logical_tables_procedure( &self, tables_data: Vec<(CreateTableExpr, RawTableInfo)>, query_context: QueryContextRef, ) -> Result<SubmitDdlTaskResponse>

async fn alter_logical_tables_procedure( &self, tables_data: Vec<AlterTableExpr>, query_context: QueryContextRef, ) -> Result<SubmitDdlTaskResponse>

async fn drop_table_procedure( &self, table_name: &TableName, table_id: TableId, drop_if_exists: bool, query_context: QueryContextRef, ) -> Result<SubmitDdlTaskResponse>

async fn drop_database_procedure( &self, catalog: String, schema: String, drop_if_exists: bool, query_context: QueryContextRef, ) -> Result<SubmitDdlTaskResponse>

async fn alter_database_procedure( &self, alter_expr: AlterDatabaseExpr, query_context: QueryContextRef, ) -> Result<SubmitDdlTaskResponse>

async fn truncate_table_procedure( &self, table_name: &TableName, table_id: TableId, time_ranges: Vec<(Timestamp, Timestamp)>, query_context: QueryContextRef, ) -> Result<SubmitDdlTaskResponse>

pub async fn create_database( &self, database: &str, create_if_not_exists: bool, options: HashMap<String, String>, query_context: QueryContextRef, ) -> Result<Output>

async fn create_database_procedure( &self, catalog: String, database: String, create_if_not_exists: bool, options: HashMap<String, String>, query_context: QueryContextRef, ) -> Result<SubmitDdlTaskResponse>

impl StatementExecutor

pub(super) async fn describe_table( &self, stmt: DescribeTable, query_ctx: QueryContextRef, ) -> Result<Output>

impl StatementExecutor

pub async fn insert( &self, insert: Box<Insert>, query_ctx: QueryContextRef, ) -> Result<Output>

impl StatementExecutor

pub async fn execute_kill( &self, query_ctx: QueryContextRef, kill: Kill, ) -> Result<Output>

async fn kill_process_id( &self, pm: &ProcessManagerRef, query_ctx: QueryContextRef, process_id: String, ) -> Result<bool>

Handles KILL <PROCESS_ID> statements.

pub async fn kill_connection_id( &self, pm: &ProcessManagerRef, query_ctx: QueryContextRef, connection_id: u32, ) -> Result<bool>

Handles MySQL KILL QUERY <CONNECTION_ID> statements.

impl StatementExecutor

pub(super) async fn show_databases( &self, stmt: ShowDatabases, query_ctx: QueryContextRef, ) -> Result<Output>

pub(super) async fn show_tables( &self, stmt: ShowTables, query_ctx: QueryContextRef, ) -> Result<Output>

pub(super) async fn show_table_status( &self, stmt: ShowTableStatus, query_ctx: QueryContextRef, ) -> Result<Output>

pub(super) async fn show_columns( &self, stmt: ShowColumns, query_ctx: QueryContextRef, ) -> Result<Output>

pub(super) async fn show_index( &self, stmt: ShowIndex, query_ctx: QueryContextRef, ) -> Result<Output>

pub(super) async fn show_region( &self, stmt: ShowRegion, query_ctx: QueryContextRef, ) -> Result<Output>

pub async fn show_create_database( &self, database_name: &str, opts: OptionMap, ) -> Result<Output>

pub async fn show_create_table( &self, table_name: TableName, table: TableRef, query_ctx: QueryContextRef, ) -> Result<Output>

pub async fn show_create_table_for_pg( &self, table_name: TableName, table: TableRef, query_ctx: QueryContextRef, ) -> Result<Output>

pub async fn show_create_view( &self, show: ShowCreateView, query_ctx: QueryContextRef, ) -> Result<Output>

pub(super) async fn show_views( &self, stmt: ShowViews, query_ctx: QueryContextRef, ) -> Result<Output>

pub(super) async fn show_flows( &self, stmt: ShowFlows, query_ctx: QueryContextRef, ) -> Result<Output>

pub async fn show_create_flow( &self, show: ShowCreateFlow, query_ctx: QueryContextRef, ) -> Result<Output>

pub fn show_variable( &self, stmt: ShowVariables, query_ctx: QueryContextRef, ) -> Result<Output>

pub async fn show_collation( &self, kind: ShowKind, query_ctx: QueryContextRef, ) -> Result<Output>

pub async fn show_charset( &self, kind: ShowKind, query_ctx: QueryContextRef, ) -> Result<Output>

pub async fn show_status(&self, query_ctx: QueryContextRef) -> Result<Output>

pub async fn show_search_path( &self, query_ctx: QueryContextRef, ) -> Result<Output>

pub async fn show_processlist( &self, stmt: ShowProcessList, query_ctx: QueryContextRef, ) -> Result<Output>

impl StatementExecutor

pub async fn plan_tql( &self, tql: Tql, query_ctx: &QueryContextRef, ) -> Result<LogicalPlan>

Plan the given Tql query and return the [LogicalPlan].

pub(super) async fn execute_tql( &self, tql: Tql, query_ctx: QueryContextRef, ) -> Result<Output>

Execute the given Tql query and return the result.

impl StatementExecutor

pub fn new( catalog_manager: CatalogManagerRef, query_engine: QueryEngineRef, procedure_executor: ProcedureExecutorRef, kv_backend: KvBackendRef, cache_invalidator: CacheInvalidatorRef, inserter: InserterRef, table_route_cache: TableRouteCacheRef, process_manager: Option<ProcessManagerRef>, ) -> Self

pub async fn execute_stmt( &self, stmt: QueryStatement, query_ctx: QueryContextRef, ) -> Result<Output>

pub async fn execute_sql( &self, stmt: Statement, query_ctx: QueryContextRef, ) -> Result<Output>

pub async fn use_database( &self, db: String, query_ctx: QueryContextRef, ) -> Result<Output>

fn set_variables( &self, set_var: SetVariables, query_ctx: QueryContextRef, ) -> Result<Output>

pub async fn plan( &self, stmt: &QueryStatement, query_ctx: QueryContextRef, ) -> Result<LogicalPlan>

pub async fn exec_plan( &self, plan: LogicalPlan, query_ctx: QueryContextRef, ) -> Result<Output>

Execute [LogicalPlan] directly.

pub fn optimize_logical_plan(&self, plan: LogicalPlan) -> Result<LogicalPlan>

async fn plan_exec( &self, stmt: QueryStatement, query_ctx: QueryContextRef, ) -> Result<Output>

async fn get_table(&self, table_ref: &TableReference<'_>) -> Result<TableRef>

pub fn procedure_executor(&self) -> &ProcedureExecutorRef

pub fn cache_invalidator(&self) -> &CacheInvalidatorRef

pub async fn convert_truncate_time_ranges( &self, table_name: &TableName, sql_values_time_range: &[(Value, Value)], query_ctx: &QueryContextRef, ) -> Result<Vec<(Timestamp, Timestamp)>>

Convert truncate time ranges for the given table from sql values to timestamps

pub(crate) fn inserter(&self) -> &InserterRef

Returns the inserter for the statement executor.

Trait Implementations

impl Clone for StatementExecutor

fn clone(&self) -> StatementExecutor

Returns a duplicate of the value.

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

Performs copy-assignment from source.