Struct Instance 

pub struct Instance {

    frontend_peer_addr: String,
    experimental_metric_export: bool,
    catalog_manager: CatalogManagerRef,
    pipeline_operator: Arc<PipelineOperator>,
    statement_executor: Arc<StatementExecutor>,
    query_engine: QueryEngineRef,
    plugins: Plugins,
    inserter: InserterRef,
    logical_batcher: Arc<OnceLock<Option<Arc<LogicalTablePendingRowsBatcher>>>>,
    deleter: DeleterRef,
    table_metadata_manager: TableMetadataManagerRef,
    event_recorder: EventRecorderRef,
    slow_query_recorder: EventRecorderRef,
    process_manager: ProcessManagerRef,
    slow_query_options: SlowQueryOptions,
    influxdb_default_merge_mode: InfluxdbMergeMode,
    trace_ingest_chunk_size: usize,
    trace_aux_cache: TraceAuxCache,
    trace_aux_limiter: Option<RequestLimiter>,
    otlp_resource_info: bool,
    suspend: Arc<AtomicBool>,
    otlp_metrics_table_legacy_cache: Arc<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

frontend_peer_addr: String

experimental_metric_export: bool

catalog_manager: CatalogManagerRef

pipeline_operator: Arc<PipelineOperator>

statement_executor: Arc<StatementExecutor>

query_engine: QueryEngineRef

plugins: Plugins

inserter: InserterRef

logical_batcher: Arc<OnceLock<Option<Arc<LogicalTablePendingRowsBatcher>>>>

deleter: DeleterRef

table_metadata_manager: TableMetadataManagerRef

event_recorder: EventRecorderRef

slow_query_recorder: EventRecorderRef

process_manager: ProcessManagerRef

slow_query_options: SlowQueryOptions

influxdb_default_merge_mode: InfluxdbMergeMode

trace_ingest_chunk_size: usize

trace_aux_cache: TraceAuxCache

trace_aux_limiter: Option<RequestLimiter>

otlp_resource_info: bool

suspend: Arc<AtomicBool>

otlp_metrics_table_legacy_cache: Arc<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

pub(crate) fn show_metric_export_capability(&self) -> Result<Output>

pub(crate) async fn copy_metric_database( &self, arg: CopyDatabaseArgument, ctx: QueryContextRef, ) -> Result<Output>

async fn export_database( &self, req: CopyDatabaseRequest, names: Option<&[String]>, cancellation: &CancellationToken, ctx: QueryContextRef, ) -> Result<DatabaseExportSummary>

async fn prepare_database_export( &self, req: CopyDatabaseRequest, names: Option<&[String]>, ctx: &QueryContextRef, ) -> Result<PreparedDatabaseExport>

impl Instance

pub(crate) fn check_table_permission( &self, ctx: &QueryContextRef, req: PermissionReq<'_>, targets: PermissionTableTargets, ) -> Result<PermissionResp>

pub(crate) fn check_row_insert_permission( &self, requests: &RowInsertRequests, ctx: &QueryContextRef, req: PermissionReq<'_>, ) -> Result<PermissionResp>

Checks every logical table targeted by normalized row inserts.

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(crate) async fn copy_packed_database( &self, arg: CopyDatabaseArgument, stmt: &Statement, ctx: QueryContextRef, ) -> Result<Output>

impl Instance

async fn check_jaeger_query_permission( &self, ctx: &QueryContextRef, ) -> ServerResult<()>

impl Instance

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

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

pub fn init_logical_batcher(self: &Arc<Self>, options: &FrontendOptions)

Initializes the shared logical batcher for enabled ingestion endpoints. Disabled endpoints leave it uninitialized so downstream routers can enable it.

pub(crate) fn logical_batcher( &self, ) -> Option<&Arc<LogicalTablePendingRowsBatcher>>

impl Instance

async fn execute_opentsdb_write( &self, data_points: Vec<DataPoint>, ctx: QueryContextRef, ) -> ServerResult<usize>

impl Instance

pub(super) async fn ingest_trace_spans( &self, pipeline: &PipelineWay, table_name: String, groups: Vec<TraceSpanGroup>, conventions: &str, ctx: QueryContextRef, ) -> ServerResult<TraceIngestOutcome>

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

fn prepare_trace_aux_write( &self, aux_data: TraceAuxData, pipeline: &PipelineWay, table_name: &str, ctx: &QueryContextRef, metrics: Option<(&IntCounter, &IntCounter)>, ) -> ServerResult<(RowInsertRequests, Vec<TraceAuxCacheKey>)>

async fn insert_trace_aux_requests( &self, requests: RowInsertRequests, keys: Vec<TraceAuxCacheKey>, pipeline: &PipelineWay, ctx: QueryContextRef, ) -> ServerResult<Output>

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_v1_prepared_chunks( &self, ingest_ctx: &TraceChunkIngestContext<'_>, request_schema: TraceRequestSchema, chunks: Vec<TraceChunkRetry>, chunk_states: Vec<TraceChunkSchemaState>, ctx: QueryContextRef, ingest_state: &mut TraceIngestState, ) -> ServerResult<()>

async fn ingest_trace_v1_chunk( &self, request_schema: &TraceRequestSchema, batch_index: usize, retry: TraceChunkRetry, schema_state: TraceChunkSchemaState, ctx: QueryContextRef, ingest_state: &mut TraceIngestState, ) -> ServerResult<()>

async fn ingest_trace_v1_rows_span_by_span( &self, retry: TraceChunkRetry, ctx: QueryContextRef, ingest_state: &mut TraceIngestState, ) -> ServerResult<()>

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

Reconcile and insert one trace request batch.

async fn plan_trace_v1_request_schema( &self, table_name: &str, request_schema: &mut TraceRequestSchema, ctx: &QueryContextRef, ) -> ServerResult<TraceRequestSchemaPlan>

async fn prepare_trace_v1_request_schema( &self, table_name: &str, request_schema: &mut TraceRequestSchema, chunks: &mut Vec<TraceChunkRetry>, chunk_states: &mut Vec<TraceChunkSchemaState>, ctx: &QueryContextRef, ) -> ServerResult<()>

fn prevalidate_trace_v1_fixed_columns( chunks: &[TraceChunkRetry], chunk_states: &[TraceChunkSchemaState], ) -> ServerResult<HashMap<String, HashSet<usize>>>

fn prevalidate_trace_v1_chunk_rewrites( request_schema: &TraceRequestSchema, chunks: &[TraceChunkRetry], chunk_states: &mut [TraceChunkSchemaState], ) -> ServerResult<HashMap<String, HashSet<usize>>>

fn prepare_trace_v1_chunk_rewrite( request_schema: &TraceRequestSchema, batch_index: usize, request: &mut RowInsertRequest, state: &mut TraceChunkSchemaState, ) -> ServerResult<Option<String>>

fn exclude_trace_v1_schema_observations( request_schema: &mut TraceRequestSchema, chunks: &mut Vec<TraceChunkRetry>, chunk_states: &mut Vec<TraceChunkSchemaState>, exclusions: &HashMap<String, HashSet<usize>>, ) -> ServerResult<()>

async fn apply_trace_v1_pre_alter( &self, ctx: &QueryContextRef, table_name: &str, pre_alter: TraceTablePreAlter, ) -> ServerResult<()>

fn classify_trace_prewrite_failure( status: StatusCode, retryable: bool, ) -> ChunkFailureReaction

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

fn trace_failure_cause(err: &Error) -> (String, String)

Returns the full cause of a pre-write failure and its display form.

output_msg masks internal errors and unwraps the root cause. Dedup must key on the full text: two causes can agree on a truncated prefix and differ exactly where the actionable detail is.

fn push_trace_failure_message( messages: &mut TraceFailureMessages, label: &'static str, key: &str, message: String, )

Records one failure, merging repeats of (label, key) into a count.

fn finish_trace_failure_message( accepted_spans: usize, rejected_spans: usize, messages: TraceFailureMessages, ) -> 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

pub async fn handle_otlp_metric_row_inserts( &self, requests: RowInsertRequests, ctx: QueryContextRef, metric_ctx: &OtlpMetricCtx, ) -> ServerResult<Output>

Inserts converted OTLP metrics through the eligible batching or ordinary write path. The caller must check table permissions and admit the complete request first.

impl Instance

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

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

impl Instance

async fn prepare_prom_store_write( &self, request: RowInsertRequests, ctx: QueryContextRef, ) -> ServerResult<(RowInsertRequests, QueryContextRef)>

async fn execute_prom_store_write( &self, request: RowInsertRequests, ctx: QueryContextRef, with_metric_engine: bool, ) -> ServerResult<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>, schema: &str, ctx: &QueryContextRef, ) -> Result<Vec<String>>

Handles metric names query request, returns the names.

pub(crate) async fn handle_query_metric_names_by_labels( &self, matchers: Vec<Matcher>, schema: &str, start: SystemTime, end: SystemTime, ctx: &QueryContextRef, ) -> Result<Vec<String>>

Handles a metric names query constrained by matchers on ordinary labels.

Instance::handle_query_metric_names answers from table metadata, which cannot resolve a label matcher: whether a metric carries pod="abc" is a property of its data. The metric engine multiplexes the logical tables of one physical table into a region holding the union of their label columns alongside __table_id, so a single distinct scan per physical table resolves the matchers for all of its logical tables at once.

Only metric engine tables are covered. Tables on other engines share no column space to scan, and a scan per table does not scale to the table counts this API is expected to answer over.

async fn physical_metric_tables( &self, catalog: &str, schema: &str, ctx: &QueryContextRef, ) -> Result<Vec<TableRef>>

Returns the metric engine physical tables of a schema.

Their data regions carry the union of their logical tables’ label columns, so scanning these covers every metric of the schema.

async fn scan_distinct_column( &self, table: TableRef, matchers: Vec<Matcher>, column: String, start: SystemTime, end: SystemTime, ctx: &QueryContextRef, ) -> Result<Vec<RecordBatch>>

Scans table for the distinct values of column that match matchers within the time range.

Callers look the table up and decode the batches; the plan between is the same whether the projected column is a label or __table_id.

async fn scan_matching_table_ids( &self, physical: TableRef, matchers: &[Matcher], start: SystemTime, end: SystemTime, ctx: &QueryContextRef, ) -> Result<Vec<TableId>>

Returns the __table_ids of physical carrying a row that matches every matcher within the time range.

pub(crate) async fn handle_query_label_values( &self, target: PermissionTableTarget, 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 frontend_peer_addr(&self) -> &str

pub fn catalog_manager(&self) -> &CatalogManagerRef

pub fn query_engine(&self) -> &QueryEngineRef

pub fn plugins(&self) -> &Plugins

fn check_permission( &self, ctx: &QueryContextRef, req: PermissionReq<'_>, ) -> Result<()>

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 event_recorder(&self) -> EventRecorderRef

Returns the event recorder configured for this frontend instance.

pub fn node_manager(&self) -> &NodeManagerRef

pub fn partition_manager(&self) -> &PartitionRuleManagerRef

pub fn table_flownode_set_cache(&self) -> &TableFlownodeSetCacheRef

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

fn statement_slow_query_timer( &self, stmt: &Statement, schema_name: String, ) -> Option<SlowQueryTimer>

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>>, timeout_recorder: Option<SlowQueryRecorder>, ) -> 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>

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

impl Instance

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

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

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

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, plan: LogicalPlan, query_ctx: QueryContextRef, timeout_recorder: Option<SlowQueryRecorder>, ) -> Result<Output>

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

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

fn show_statement_dataframe<'a>( &'a self, stmt: &'a Statement, query_ctx: &'a QueryContextRef, ) -> Pin<Box<dyn Future<Output = Option<Result<DataFrame>>> + Send + 'a>>

Builds the [DataFrame] for an information-schema-backed SHOW statement; None for other statements. The future is boxed to keep do_describe_inner’s state machine small.

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>

impl Instance

fn check_prom_query_privilege(&self, query_ctx: &QueryContextRef) -> Result<()>

fn prom_expr_permission_targets( &self, expr: &Expr, query_ctx: &QueryContextRef, ) -> Result<Option<Vec<PermissionTableTarget>>>

async fn is_physical_query_permission_target( &self, target: &PermissionTableTarget, query_ctx: &QueryContextRef, ) -> Result<bool>

async fn resolve_query_permission_targets( &self, targets: PermissionTableTargets, query_ctx: &QueryContextRef, ) -> Result<PermissionTableTargets>

fn prom_queries_permission_targets( &self, queries: &[ParsedPromQuery], query_ctx: &QueryContextRef, ) -> Result<PermissionTableTargets>

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,

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<MetricsIngestOutcome>> + 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 preflight<'life0, 'life1, 'async_trait>( &'life0 self, data_points: &'life1 [DataPoint], ctx: QueryContextRef, ) -> Pin<Box<dyn Future<Output = ServerResult<()>> + Send + 'async_trait>>

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

Checks all points in one external request before per-point debug execution.

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

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

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

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

Executes an ordinary HTTP put with optional batching. Debug callers retain Self::exec; the frontend clears HTTP batching selection there so diagnostic requests preserve direct, per-point error attribution.

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 insert_all<'life0, 'async_trait>( &'life0 self, inputs: Vec<(QueryContextRef, RowInsertRequests)>, ) -> Pin<Box<dyn Future<Output = ServerResult<Vec<ServerResult<Output>>>> + Send + 'async_trait>>

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

Checks every batch before inserting any of them.

fn check_pipeline_query_permission( &self, query_ctx: &QueryContextRef, ) -> ServerResult<()>

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,

Loads a compiled pipeline for execution.

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_prepared<'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,

Writes one batch after Self::pre_write has succeeded for the entire request.

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 write_all<'life0, 'async_trait>( &'life0 self, requests: Vec<(QueryContextRef, RowInsertRequests)>, with_metric_engine: bool, ) -> Pin<Box<dyn Future<Output = ServerResult<Vec<ServerResult<Output>>>> + Send + 'async_trait>>

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

Checks every batch before writing any of them.

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

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 do_query_parsed<'life0, 'async_trait>( &'life0 self, query: ParsedPromQuery, query_ctx: QueryContextRef, ) -> Pin<Box<dyn Future<Output = Result<Output>> + Send + 'async_trait>>

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

Executes a query whose statement has already been parsed.

fn check_query_permission<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, queries: &'life1 [PromQuery], query_ctx: &'life2 QueryContextRef, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

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

Checks access to every table referenced by a batch of PromQL queries.

fn check_query_permission_parsed<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, queries: &'life1 [ParsedPromQuery], query_ctx: &'life2 QueryContextRef, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

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

Checks access without parsing the queries again.

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

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

Checks access to targets resolved by Prometheus metadata APIs.

fn filter_metadata_metric_names<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, metric_names: Vec<String>, schema: &'life1 str, query_ctx: &'life2 QueryContextRef, ) -> Pin<Box<dyn Future<Output = Result<Vec<String>>> + Send + 'async_trait>>

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

Removes inaccessible metric names from logical-table metadata enumeration results.

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

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

Query metric table names by the __name__ matchers.

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

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

Query metric table names that carry data matching matchers in the time range. matchers must hold only ordinary label matchers: __name__ names a table and the database and field matchers name no column, so the caller resolves all three before calling.

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 supports_metric_batch_ddl(&self) -> bool

Whether this handler implements the bounded logical-table DDL operation.

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

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

Validates and authorizes the whole batch before submitting one procedure.

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_analyze_stream_query<'life0, 'life1, 'async_trait>( &'life0 self, query: &'life1 str, query_ctx: QueryContextRef, ) -> Pin<Box<dyn Future<Output = Result<Output>> + Send + 'async_trait>>

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

Executes an HTTP analyze-stream query.

fn do_exec_plan<'life0, 'async_trait>( &'life0 self, plan: LogicalPlan, stmt: Option<Statement>, 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,