Struct Instance 

pub struct Instance {

    catalog_manager: CatalogManagerRef,
    pipeline_operator: Arc<PipelineOperator>,
    statement_executor: Arc<StatementExecutor>,
    query_engine: QueryEngineRef,
    plugins: Plugins,
    inserter: InserterRef,
    deleter: DeleterRef,
    table_metadata_manager: TableMetadataManagerRef,
    event_recorder: Option<EventRecorderRef>,
    process_manager: ProcessManagerRef,
    slow_query_options: SlowQueryOptions,
    suspend: Arc<AtomicBool>,
    otlp_metrics_table_legacy_cache: DashMap<String, DashMap<String, bool>>,
}

The frontend instance contains necessary components, and implements many traits, like servers::query_handler::grpc::GrpcQueryHandler, servers::query_handler::sql::SqlQueryHandler, etc.

Fields

catalog_manager: CatalogManagerRef

pipeline_operator: Arc<PipelineOperator>

statement_executor: Arc<StatementExecutor>

query_engine: QueryEngineRef

plugins: Plugins

inserter: InserterRef

deleter: DeleterRef

table_metadata_manager: TableMetadataManagerRef

event_recorder: Option<EventRecorderRef>

process_manager: ProcessManagerRef

slow_query_options: SlowQueryOptions

suspend: Arc<AtomicBool>

otlp_metrics_table_legacy_cache: DashMap<String, DashMap<String, bool>>

Implementations

impl Instance

fn build_dashboard_schema() -> (String, Vec<String>, Vec<ColumnDef>)

Build a schema for dashboard table. Returns the (time index, primary keys, column) definitions.

fn build_dashboard_insert_column_schemas() -> Vec<PbColumnSchema>

Build a column schemas for inserting a row into the dashboard table.

fn dashboard_query_ctx(table_info: &TableInfo) -> QueryContextRef

async fn create_dashboard_table_if_not_exists( &self, ctx: QueryContextRef, ) -> Result<TableRef>

async fn insert_dashboard( &self, name: &str, definition: &str, query_ctx: QueryContextRef, ) -> Result<()>

Insert a dashboard into the dashboard table.

async fn list_dashboards( &self, query_ctx: QueryContextRef, ) -> Result<Vec<DashboardDefinition>>

List all dashboards.

async fn delete_dashboard( &self, name: &str, query_ctx: QueryContextRef, ) -> Result<()>

Delete a dashboard by name.

impl Instance

fn handle_put_record_batch_stream_inner( &self, stream: PutRecordBatchRequestStream, ctx: QueryContextRef, ) -> Pin<Box<dyn Stream<Item = Result<DoPutResponse>> + Send>>

async fn handle_insert_plan( &self, insert: InsertIntoPlan, ctx: QueryContextRef, ) -> Result<Output>

pub async fn handle_inserts( &self, requests: InsertRequests, ctx: QueryContextRef, ) -> Result<Output>

pub async fn handle_row_inserts( &self, requests: RowInsertRequests, ctx: QueryContextRef, accommodate_existing_schema: bool, is_single_value: bool, ) -> Result<Output>

pub async fn handle_influx_row_inserts( &self, requests: RowInsertRequests, ctx: QueryContextRef, ) -> Result<Output>

pub async fn handle_metric_row_inserts( &self, requests: RowInsertRequests, ctx: QueryContextRef, physical_table: String, ) -> Result<Output>

pub async fn handle_deletes( &self, requests: DeleteRequests, ctx: QueryContextRef, ) -> Result<Output>

pub async fn handle_row_deletes( &self, requests: RowDeleteRequests, ctx: QueryContextRef, ) -> Result<Output>

impl Instance

pub async fn handle_log_inserts( &self, log: RowInsertRequests, ctx: QueryContextRef, ) -> ServerResult<Output>

pub async fn handle_trace_inserts( &self, rows: RowInsertRequests, ctx: QueryContextRef, ) -> ServerResult<Output>

impl Instance

async fn ingest_trace_spans( &self, pipeline_handler: PipelineHandlerRef, pipeline: &PipelineWay, pipeline_params: &GreptimePipelineParams, table_name: String, groups: Vec<TraceSpanGroup>, ctx: QueryContextRef, ) -> ServerResult<TraceIngestOutcome>

Ingest OTLP trace spans with chunk-level writes and span-level fallback on deterministic chunk failures.

async fn ingest_trace_chunk( &self, ingest_ctx: &TraceChunkIngestContext<'_>, chunk: Vec<TraceSpan>, ctx: QueryContextRef, ingest_state: &mut TraceIngestState, ) -> ServerResult<()>

Ingest one owned trace chunk so successful spans can be moved into the accepted set without extra cloning.

async fn ingest_trace_chunk_span_by_span( &self, ingest_ctx: &TraceChunkIngestContext<'_>, chunk: Vec<TraceSpan>, ctx: QueryContextRef, ingest_state: &mut TraceIngestState, ) -> ServerResult<()>

Retry spans one by one only after a deterministic chunk failure.

async fn insert_trace_requests( &self, requests: RowInsertRequests, is_trace_v1_model: bool, ctx: QueryContextRef, ) -> ServerResult<Output>

Reconcile and insert one trace request batch.

fn classify_trace_chunk_failure(status: StatusCode) -> ChunkFailureReaction

fn should_propagate_trace_span_failure(status: StatusCode) -> bool

fn add_trace_write_cost(outcome: &mut TraceIngestOutcome, cost: usize)

fn push_trace_failure_message( messages: &mut Vec<String>, label: &str, message: String, )

fn finish_trace_failure_message( accepted_spans: usize, rejected_spans: usize, messages: Vec<String>, ) -> Option<String>

async fn alter_trace_table_columns_to_float64( &self, ctx: &QueryContextRef, table_name: &str, column_names: &[String], ) -> ServerResult<()>

Widen existing trace table columns to Float64 before request rewrite.

async fn reconcile_trace_column_types( &self, requests: &mut RowInsertRequests, ctx: &QueryContextRef, ) -> ServerResult<()>

Coerce request column types and values to match the existing table schema for compatible type pairs. Existing table schema wins when present; otherwise the full request batch decides a stable target type.

impl Instance

async fn handle_remote_query( &self, ctx: &QueryContextRef, catalog_name: &str, schema_name: &str, table_name: &str, query: &Query, ) -> Result<Output>

async fn handle_remote_queries( &self, ctx: QueryContextRef, queries: &[Query], ) -> ServerResult<Vec<(String, Output)>>

impl Instance

async fn create_metric_physical_table_if_missing( &self, catalog: &str, schema: &str, physical_table: &str, ctx: QueryContextRef, ) -> ServerResult<()>

impl Instance

pub(crate) async fn handle_query_metric_names( &self, matchers: Vec<Matcher>, ctx: &QueryContextRef, ) -> Result<Vec<String>>

Handles metric names query request, returns the names.

pub(crate) async fn handle_query_label_values( &self, metric: String, label_name: String, matchers: Vec<Matcher>, start: SystemTime, end: SystemTime, ctx: &QueryContextRef, ) -> Result<Vec<String>>

Handles label values query request, returns the values.

impl Instance

pub fn catalog_manager(&self) -> &CatalogManagerRef

pub fn query_engine(&self) -> &QueryEngineRef

pub fn plugins(&self) -> &Plugins

pub fn statement_executor(&self) -> &StatementExecutorRef

pub fn table_metadata_manager(&self) -> &TableMetadataManagerRef

pub fn inserter(&self) -> &InserterRef

pub fn process_manager(&self) -> &ProcessManagerRef

pub fn node_manager(&self) -> &NodeManagerRef

pub fn partition_manager(&self) -> &PartitionRuleManagerRef

pub fn cache_invalidator(&self) -> &CacheInvalidatorRef

pub fn procedure_executor(&self) -> &ProcedureExecutorRef

pub fn suspend_state(&self) -> Arc<AtomicBool>

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

impl Instance

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

async fn exec_statement_with_timeout( &self, stmt: Statement, query_ctx: QueryContextRef, query_interceptor: Option<&SqlQueryInterceptorRef<Error>>, ) -> Result<Output>

async fn exec_statement( &self, stmt: Statement, query_ctx: QueryContextRef, query_interceptor: Option<&SqlQueryInterceptorRef<Error>>, ) -> Result<Output>

async fn plan_and_exec_sql( &self, stmt: Statement, query_ctx: &QueryContextRef, query_interceptor: Option<&SqlQueryInterceptorRef<Error>>, ) -> Result<Output>

async fn plan_and_exec_tql( &self, query_ctx: &QueryContextRef, query_interceptor: Option<&SqlQueryInterceptorRef<Error>>, tql: Tql, ) -> Result<Output>

async fn check_otlp_legacy( &self, names: &[&String], ctx: QueryContextRef, ) -> Result<bool>

impl Instance

async fn do_query_inner( &self, query: &str, query_ctx: QueryContextRef, ) -> Vec<Result<Output>>

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

async fn exec_plan_with_timeout( &self, stmt: Option<Statement>, plan: LogicalPlan, query_ctx: QueryContextRef, ) -> Result<Output>

async fn do_exec_plan_inner( &self, stmt: Option<Statement>, plan: LogicalPlan, query_ctx: QueryContextRef, ) -> Result<Output>

async fn do_promql_query_inner( &self, query: &PromQuery, query_ctx: QueryContextRef, ) -> Vec<Result<Output>>

async fn do_describe_inner( &self, stmt: Statement, query_ctx: QueryContextRef, ) -> Result<Option<DescribeResult>>

async fn is_valid_schema_inner( &self, catalog: &str, schema: &str, ) -> Result<bool>

Trait Implementations

impl Clone for Instance

fn clone(&self) -> Instance

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl DashboardHandler for Instance

fn save<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, name: &'life1 str, definition: &'life2 str, ctx: QueryContextRef, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

fn list<'life0, 'async_trait>( &'life0 self, ctx: QueryContextRef, ) -> Pin<Box<dyn Future<Output = Result<Vec<DashboardDefinition>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

fn delete<'life0, 'life1, 'async_trait>( &'life0 self, name: &'life1 str, ctx: QueryContextRef, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

impl GrpcQueryHandler for Instance

fn do_query<'life0, 'async_trait>( &'life0 self, request: Request, ctx: QueryContextRef, ) -> Pin<Box<dyn Future<Output = Result<Output>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

fn handle_put_record_batch_stream( &self, stream: PutRecordBatchRequestStream, ctx: QueryContextRef, ) -> Pin<Box<dyn Stream<Item = Result<DoPutResponse>> + Send>>

impl InfluxdbLineProtocolHandler for Instance

fn exec<'life0, 'async_trait>( &'life0 self, request: InfluxdbRequest, ctx: QueryContextRef, ) -> Pin<Box<dyn Future<Output = Result<Output>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

A successful request will not return a response. Only on error will the socket return a line of data.

impl JaegerQueryHandler for Instance

fn get_services<'life0, 'async_trait>( &'life0 self, ctx: QueryContextRef, ) -> Pin<Box<dyn Future<Output = ServerResult<Output>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Get trace services. It’s used for /api/services API.

fn get_operations<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, ctx: QueryContextRef, service_name: &'life1 str, span_kind: Option<&'life2 str>, ) -> Pin<Box<dyn Future<Output = ServerResult<Output>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Get Jaeger operations. It’s used for /api/operations and /api/services/{service_name}/operations API.

fn get_trace<'life0, 'life1, 'async_trait>( &'life0 self, ctx: QueryContextRef, trace_id: &'life1 str, start_time: Option<i64>, end_time: Option<i64>, limit: Option<usize>, ) -> Pin<Box<dyn Future<Output = ServerResult<Output>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Retrieves a trace by its unique identifier.

fn find_traces<'life0, 'async_trait>( &'life0 self, ctx: QueryContextRef, query_params: QueryTraceParams, ) -> Pin<Box<dyn Future<Output = ServerResult<Output>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Find traces by query params. It’s used for /api/traces API.

impl LogQueryHandler for Instance

fn query<'life0, 'async_trait>( &'life0 self, request: LogQuery, ctx: QueryContextRef, ) -> Pin<Box<dyn Future<Output = ServerResult<Output>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Execute a log query.

fn catalog_manager( &self, _ctx: &QueryContext, ) -> ServerResult<&dyn CatalogManager>

Get catalog manager.

impl OpenTelemetryProtocolHandler for Instance

fn metrics<'life0, 'async_trait>( &'life0 self, request: ExportMetricsServiceRequest, ctx: QueryContextRef, ) -> Pin<Box<dyn Future<Output = ServerResult<Output>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Handling opentelemetry metrics request

fn traces<'life0, 'async_trait>( &'life0 self, pipeline_handler: PipelineHandlerRef, request: ExportTraceServiceRequest, pipeline: PipelineWay, pipeline_params: GreptimePipelineParams, table_name: String, ctx: QueryContextRef, ) -> Pin<Box<dyn Future<Output = ServerResult<TraceIngestOutcome>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Handling opentelemetry traces request

fn logs<'life0, 'async_trait>( &'life0 self, pipeline_handler: PipelineHandlerRef, request: ExportLogsServiceRequest, pipeline: PipelineWay, pipeline_params: GreptimePipelineParams, table_name: String, ctx: QueryContextRef, ) -> Pin<Box<dyn Future<Output = ServerResult<Vec<Output>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

impl OpentsdbProtocolHandler for Instance

fn exec<'life0, 'async_trait>( &'life0 self, data_points: Vec<DataPoint>, ctx: QueryContextRef, ) -> Pin<Box<dyn Future<Output = Result<usize>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

A successful request will not return a response. Only on error will the socket return a line of data.

impl PendingRowsSchemaAlterer for Instance

fn create_tables_if_missing_batch<'life0, 'life1, 'life2, 'life3, 'life4, 'life5, 'async_trait>( &'life0 self, catalog: &'life1 str, schema: &'life2 str, tables: &'life3 [(&'life4 str, &'life5 [ColumnSchema])], with_metric_engine: bool, ctx: QueryContextRef, ) -> Pin<Box<dyn Future<Output = ServerResult<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait, 'life4: 'async_trait, 'life5: 'async_trait,

Batch-create multiple logical tables that are missing. Each entry is (table_name, request_schema).

fn add_missing_prom_tag_columns_batch<'life0, 'life1, 'life2, 'life3, 'life4, 'life5, 'async_trait>( &'life0 self, catalog: &'life1 str, schema: &'life2 str, tables: &'life3 [(&'life4 str, &'life5 [String])], ctx: QueryContextRef, ) -> Pin<Box<dyn Future<Output = ServerResult<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait, 'life4: 'async_trait, 'life5: 'async_trait,

Batch-alter multiple logical tables to add missing tag columns. Each entry is (table_name, missing_column_names).

impl PipelineHandler for Instance

fn insert<'life0, 'async_trait>( &'life0 self, log: RowInsertRequests, ctx: QueryContextRef, ) -> Pin<Box<dyn Future<Output = ServerResult<Output>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

fn get_pipeline<'life0, 'life1, 'async_trait>( &'life0 self, name: &'life1 str, version: PipelineVersion, query_ctx: QueryContextRef, ) -> Pin<Box<dyn Future<Output = ServerResult<Arc<Pipeline>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

fn insert_pipeline<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, name: &'life1 str, content_type: &'life2 str, pipeline: &'life3 str, query_ctx: QueryContextRef, ) -> Pin<Box<dyn Future<Output = ServerResult<PipelineInfo>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait,

fn delete_pipeline<'life0, 'life1, 'async_trait>( &'life0 self, name: &'life1 str, version: PipelineVersion, ctx: QueryContextRef, ) -> Pin<Box<dyn Future<Output = ServerResult<Option<()>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

fn get_table<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, table: &'life1 str, query_ctx: &'life2 QueryContext, ) -> Pin<Box<dyn Future<Output = Result<Option<Arc<Table>>, Error>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

fn build_pipeline(&self, pipeline: &str) -> ServerResult<Pipeline>

fn get_pipeline_str<'life0, 'life1, 'async_trait>( &'life0 self, name: &'life1 str, version: PipelineVersion, query_ctx: QueryContextRef, ) -> Pin<Box<dyn Future<Output = ServerResult<(String, TimestampNanosecond)>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Get a original pipeline by name.

impl PromStoreProtocolHandler for Instance

fn pre_write<'life0, 'life1, 'async_trait>( &'life0 self, request: &'life1 RowInsertRequests, ctx: QueryContextRef, ) -> Pin<Box<dyn Future<Output = ServerResult<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Runs pre-write checks/hooks for prometheus remote write requests.

fn write<'life0, 'async_trait>( &'life0 self, request: RowInsertRequests, ctx: QueryContextRef, with_metric_engine: bool, ) -> Pin<Box<dyn Future<Output = ServerResult<Output>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Handling prometheus remote write requests

fn read<'life0, 'async_trait>( &'life0 self, request: ReadRequest, ctx: QueryContextRef, ) -> Pin<Box<dyn Future<Output = ServerResult<PromStoreResponse>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Handling prometheus remote read requests

fn ingest_metrics<'life0, 'async_trait>( &'life0 self, _metrics: Metrics, ) -> Pin<Box<dyn Future<Output = ServerResult<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Handling push gateway requests

impl PrometheusHandler for Instance

fn do_query<'life0, 'life1, 'async_trait>( &'life0 self, query: &'life1 PromQuery, query_ctx: QueryContextRef, ) -> Pin<Box<dyn Future<Output = Result<Output>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

fn query_metric_names<'life0, 'life1, 'async_trait>( &'life0 self, matchers: Vec<Matcher>, ctx: &'life1 QueryContextRef, ) -> Pin<Box<dyn Future<Output = Result<Vec<String>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Query metric table names by the __name__ matchers.

fn query_label_values<'life0, 'life1, 'async_trait>( &'life0 self, metric: String, label_name: String, matchers: Vec<Matcher>, start: SystemTime, end: SystemTime, ctx: &'life1 QueryContextRef, ) -> Pin<Box<dyn Future<Output = Result<Vec<String>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

fn catalog_manager(&self) -> CatalogManagerRef

impl SqlQueryHandler for Instance

fn do_query<'life0, 'life1, 'async_trait>( &'life0 self, query: &'life1 str, query_ctx: QueryContextRef, ) -> Pin<Box<dyn Future<Output = Vec<Result<Output>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

fn do_exec_plan<'life0, 'async_trait>( &'life0 self, stmt: Option<Statement>, plan: LogicalPlan, query_ctx: QueryContextRef, ) -> Pin<Box<dyn Future<Output = Result<Output>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

fn do_promql_query<'life0, 'life1, 'async_trait>( &'life0 self, query: &'life1 PromQuery, query_ctx: QueryContextRef, ) -> Pin<Box<dyn Future<Output = Vec<Result<Output>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

fn do_describe<'life0, 'async_trait>( &'life0 self, stmt: Statement, query_ctx: QueryContextRef, ) -> Pin<Box<dyn Future<Output = Result<Option<DescribeResult>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

fn is_valid_schema<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, catalog: &'life1 str, schema: &'life2 str, ) -> Pin<Box<dyn Future<Output = Result<bool>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,