catalog/system_schema/
semantic_graph.rs

1// Copyright 2023 Greptime Team
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15//! The computed entity-graph tables `greptime_private.semantic_entities` and
16//! `greptime_private.semantic_relationships`.
17//!
18//! They live in `greptime_private`, not `information_schema`: scanning them
19//! triggers read-time derivation over telemetry tables (trace self-joins, ...),
20//! which breaks the "cheap, metadata-only" expectation users have of
21//! `information_schema`. `greptime_private` also hosts the physical
22//! declared-edge table (`semantic_relationships_declared`), whose rows are
23//! unioned into the computed `semantic_relationships`.
24//!
25//! These are thin forwarders: their rows are derived at read time by the injected
26//! [`EntityGraphProvider`], which enumerates the `table_semantics` declarations,
27//! builds typed DataFusion derivation plans, and executes them via the query
28//! engine. When no provider is injected (e.g. before the engine is up, or on a
29//! non-frontend node) they stream empty. The fixed schemas here must match the
30//! columns the provider's plans project; JSON columns are `json` (JSONB), whose
31//! Arrow storage type is `Binary` — the derived batches are rebuilt against the
32//! declared Arrow schema (which carries the `json` extension metadata) in
33//! [`SystemTable::to_stream`].
34
35use std::sync::{Arc, LazyLock, Weak};
36
37use common_catalog::consts::{
38    CONFIDENCE_COLUMN, DEFAULT_PRIVATE_SCHEMA_NAME, DST_ID_COLUMN, DST_TYPE_COLUMN,
39    DURATION_COUNT_COLUMN, DURATION_MAX_COLUMN, DURATION_SUM_COLUMN, EDGE_ATTRIBUTES_COLUMN,
40    ENTITY_DESCRIPTIVE_COLUMN, ENTITY_ID_ATTRS_COLUMN, ENTITY_ID_COLUMN, ENTITY_SCOPE_COLUMN,
41    ENTITY_TYPE_COLUMN, ERROR_COUNT_COLUMN, FRESH_UNTIL_COLUMN, OBSERVED_AT_COLUMN,
42    PROVENANCE_COLUMN, REL_TYPE_COLUMN, REQUEST_COUNT_COLUMN, SEMANTIC_ENTITIES_TABLE_ID,
43    SEMANTIC_ENTITIES_TABLE_NAME as SEMANTIC_ENTITIES, SEMANTIC_RELATIONSHIPS_TABLE_ID,
44    SEMANTIC_RELATIONSHIPS_TABLE_NAME as SEMANTIC_RELATIONSHIPS, SOURCE_TABLES_COLUMN,
45    SRC_ID_COLUMN, SRC_TYPE_COLUMN, UNMATCHED_COUNT_COLUMN, WINDOW_END_COLUMN, WINDOW_START_COLUMN,
46};
47use common_error::ext::BoxedError;
48use common_recordbatch::adapter::AsyncRecordBatchStreamAdapter;
49use common_recordbatch::{
50    DfRecordBatch, EmptyRecordBatchStream, RecordBatch, RecordBatchStreamWrapper,
51    SendableRecordBatchStream,
52};
53use datatypes::prelude::ConcreteDataType;
54use datatypes::schema::{ColumnSchema, Schema, SchemaRef};
55use futures::StreamExt;
56use session::context::QueryContextRef;
57use snafu::ResultExt;
58use store_api::storage::{ScanRequest, TableId};
59use table::TableRef;
60use table::metadata::TableInfo;
61
62use crate::CatalogManager;
63use crate::error::{InternalSnafu, Result};
64use crate::system_schema::{SystemSchemaProviderInner, SystemTable, SystemTableRef, utils};
65
66pub type EntityGraphProviderRef = Arc<dyn EntityGraphProvider>;
67
68/// Where a table's entity declaration came from.
69pub enum DeclarationOrigin {
70    /// A `greptime.semantic.entity.<type>.*` table option.
71    Declared,
72    /// The built-in derivation conventions shipped with the binary.
73    Convention,
74}
75
76impl DeclarationOrigin {
77    pub fn as_str(&self) -> &'static str {
78        match self {
79            Self::Declared => "declared",
80            Self::Convention => "convention",
81        }
82    }
83}
84
85/// One entity a table declares, as `information_schema.table_semantics`
86/// reports it. A catalog-side projection of the derivation's own declaration
87/// type, which lives above this crate.
88pub struct TableEntityDeclaration {
89    pub entity_type: String,
90    pub origin: DeclarationOrigin,
91    pub id_columns: Vec<String>,
92    pub id_qualifier: Option<String>,
93    /// Columns whose presence on a row withdraws this declaration for that row.
94    /// Reported because the declaration otherwise reads as unconditional.
95    pub superseded_by_columns: Vec<String>,
96    pub descriptive_columns: Vec<String>,
97    pub scope_columns: Vec<String>,
98}
99
100/// Produces the rows of the computed entity-graph tables at read time.
101///
102/// Implemented above the query engine (in the frontend) and injected into the
103/// catalog manager *after* construction — the provider needs the engine, which
104/// needs the catalog manager — so this late binding breaks the
105/// `catalog -> query` dependency cycle.
106///
107/// `query_ctx` is the outer query's context, captured when the computed table
108/// is resolved: the derivation must read only sources that context may read
109/// and execute under it, inheriting the caller's permissions, cancellation and
110/// deadline (the RFC's derivation contract). `None` (context-less internal
111/// resolution) keeps the provider's default behaviour.
112#[async_trait::async_trait]
113pub trait EntityGraphProvider: Send + Sync {
114    /// Produces the entity registry (`semantic_entities`) rows for `catalog`.
115    /// `None` means no source table declared an entity.
116    async fn scan_entities(
117        &self,
118        catalog: &str,
119        request: ScanRequest,
120        query_ctx: Option<QueryContextRef>,
121    ) -> std::result::Result<Option<SendableRecordBatchStream>, BoxedError>;
122
123    /// Produces the relationship set (`semantic_relationships`) rows for `catalog`.
124    async fn scan_relationships(
125        &self,
126        catalog: &str,
127        request: ScanRequest,
128        query_ctx: Option<QueryContextRef>,
129    ) -> std::result::Result<Option<SendableRecordBatchStream>, BoxedError>;
130
131    /// The entities `table_info` contributes to the graph: its explicit
132    /// declarations merged with the ones the conventions derive. Metadata-only
133    /// by contract — it runs per table on the `table_semantics` scan and must
134    /// not touch the query engine.
135    fn table_declarations(&self, table_info: &TableInfo) -> Vec<TableEntityDeclaration>;
136}
137
138/// Serves the computed graph tables under `greptime_private`, overlaid on the
139/// schema's physical tables the same way the `numbers` table overlays `public`
140/// (the system catalog is consulted before physical table resolution).
141pub(crate) struct SemanticGraphTableProvider {
142    catalog_name: String,
143    catalog_manager: Weak<dyn CatalogManager>,
144    /// The resolving query's context; captured per resolution (the provider is
145    /// built on demand, never cached), so it cannot leak across sessions.
146    query_ctx: Option<QueryContextRef>,
147}
148
149impl SemanticGraphTableProvider {
150    pub(crate) fn new(
151        catalog_name: String,
152        catalog_manager: Weak<dyn CatalogManager>,
153        query_ctx: Option<QueryContextRef>,
154    ) -> Self {
155        Self {
156            catalog_name,
157            catalog_manager,
158            query_ctx,
159        }
160    }
161
162    pub(crate) fn table_names() -> Vec<String> {
163        vec![
164            SEMANTIC_ENTITIES.to_string(),
165            SEMANTIC_RELATIONSHIPS.to_string(),
166        ]
167    }
168
169    pub(crate) fn table_exists(name: &str) -> bool {
170        name == SEMANTIC_ENTITIES || name == SEMANTIC_RELATIONSHIPS
171    }
172
173    pub(crate) fn table(&self, name: &str) -> Option<TableRef> {
174        self.build_table(name)
175    }
176}
177
178impl SystemSchemaProviderInner for SemanticGraphTableProvider {
179    fn catalog_name(&self) -> &str {
180        &self.catalog_name
181    }
182
183    fn schema_name() -> &'static str {
184        DEFAULT_PRIVATE_SCHEMA_NAME
185    }
186
187    fn system_table(&self, name: &str) -> Option<SystemTableRef> {
188        let kind = match name {
189            SEMANTIC_ENTITIES => GraphTableKind::Entities,
190            SEMANTIC_RELATIONSHIPS => GraphTableKind::Relationships,
191            _ => return None,
192        };
193        Some(Arc::new(SemanticGraphTable::new(
194            kind,
195            self.catalog_name.clone(),
196            self.catalog_manager.clone(),
197            self.query_ctx.clone(),
198        )) as _)
199    }
200}
201
202fn ts() -> ConcreteDataType {
203    ConcreteDataType::timestamp_millisecond_datatype()
204}
205
206fn string() -> ConcreteDataType {
207    ConcreteDataType::string_datatype()
208}
209
210fn json() -> ConcreteDataType {
211    ConcreteDataType::json_datatype()
212}
213
214/// Schema of `semantic_entities` — the node set of the graph, one row per entity
215/// observed in a time window. Must match the registry derivation projection.
216///
217/// Columns:
218/// - `observed_at`   — TIME INDEX; the 60s time bucket the entity was observed in.
219/// - `window_start`  — start of that observation window.
220/// - `window_end`    — end of the window (`window_start` + 60s).
221/// - `fresh_until`   — time up to which the entity is considered present; equals
222///   `window_end` for derived rows (the graph is a sliding window, not a
223///   current-state table: an entity exists in a query window only if it has
224///   observed evidence there).
225/// - `entity_type`   — the entity's type, e.g. `service`, `host`, `k8s.pod`,
226///   `process`, `service.instance` (the OTel-style, possibly dotted, type).
227/// - `entity_id`     — canonical identifier: the identifying values in declared
228///   order, escaped and joined.
229/// - `entity_id_attrs` — JSON object of the identifying attributes, so a
230///   consumer holding an id can tell which columns it came from.
231/// - `scope`         — namespace/environment the id is scoped to; empty when none.
232/// - `descriptive`   — JSON snapshot of the entity's descriptive (non-identifying)
233///   attributes; NULL when no descriptive columns were declared.
234/// - `source_tables` — JSON array of the telemetry tables that contributed this entity.
235static ENTITIES_SCHEMA: LazyLock<SchemaRef> = LazyLock::new(|| {
236    Arc::new(Schema::new(vec![
237        ColumnSchema::new(OBSERVED_AT_COLUMN, ts(), false).with_time_index(true),
238        ColumnSchema::new(WINDOW_START_COLUMN, ts(), true),
239        ColumnSchema::new(WINDOW_END_COLUMN, ts(), true),
240        ColumnSchema::new(FRESH_UNTIL_COLUMN, ts(), true),
241        ColumnSchema::new(ENTITY_TYPE_COLUMN, string(), false),
242        ColumnSchema::new(ENTITY_ID_COLUMN, string(), false),
243        ColumnSchema::new(ENTITY_ID_ATTRS_COLUMN, json(), true),
244        ColumnSchema::new(ENTITY_SCOPE_COLUMN, string(), true),
245        ColumnSchema::new(ENTITY_DESCRIPTIVE_COLUMN, json(), true),
246        ColumnSchema::new(SOURCE_TABLES_COLUMN, json(), true),
247    ]))
248});
249
250/// Schema of `semantic_relationships` — the edge set of the graph, one row per
251/// edge observed in a time window. This is the 18-column contract every derived
252/// branch and the declared-edge table must project for the top-level `UNION ALL`.
253///
254/// Columns:
255/// - `observed_at`   — TIME INDEX; the 60s time bucket the edge was observed in.
256/// - `window_start` / `window_end` — the observation window (`window_start` + 60s).
257/// - `fresh_until`   — time up to which the edge is considered live; equals
258///   `window_end` for derived edges (from `valid_until` for declared edges).
259/// - `src_type` / `src_id` — type and canonical id of the source endpoint.
260/// - `dst_type` / `dst_id` — type and canonical id of the destination endpoint.
261/// - `rel_type`      — relationship kind, e.g. `calls`, `runs_on`, `contains`,
262///   `part_of`, `depends_on` (direction is src → dst; the inverse is a query concern).
263/// - `provenance`    — how the edge was obtained: `trace` (derived from spans),
264///   `attribute` (shared-identity join), `declared` (hand-inserted), or `agent`
265///   (agent-inferred). Part of the edge identity, so edges of different provenance
266///   for the same pair coexist.
267/// - `confidence`    — derivation certainty in `[0, 1]`: `1.0` for paired or
268///   declared edges, lower for virtual-node or agent-inferred edges. It does
269///   not correct for trace sampling.
270/// - `request_count` — RED: number of requests over the window (`calls` edges).
271/// - `unmatched_count` — client spans on this edge with no server span. The
272///   other RED columns describe the pairs alone, so this is what separates a
273///   callee that stopped responding from traffic that stopped arriving.
274/// - `error_count`   — RED: number of errored requests over the window.
275/// - `duration_sum`  — RED: sum of request durations, in seconds, over the window.
276/// - `duration_count`— RED: number of durations summed (pair with `duration_sum`
277///   to get an average).
278/// - `duration_max`  — RED: longest single request, in seconds, over the same
279///   population `duration_sum` covers.
280/// - `attributes`    — JSON of edge attributes, e.g. `connection_type`,
281///   `db.system`, `peer.service`.
282static RELATIONSHIPS_SCHEMA: LazyLock<SchemaRef> = LazyLock::new(|| {
283    Arc::new(Schema::new(vec![
284        ColumnSchema::new(OBSERVED_AT_COLUMN, ts(), false).with_time_index(true),
285        ColumnSchema::new(WINDOW_START_COLUMN, ts(), true),
286        ColumnSchema::new(WINDOW_END_COLUMN, ts(), true),
287        ColumnSchema::new(FRESH_UNTIL_COLUMN, ts(), true),
288        ColumnSchema::new(SRC_TYPE_COLUMN, string(), false),
289        ColumnSchema::new(SRC_ID_COLUMN, string(), false),
290        ColumnSchema::new(DST_TYPE_COLUMN, string(), false),
291        ColumnSchema::new(DST_ID_COLUMN, string(), false),
292        ColumnSchema::new(REL_TYPE_COLUMN, string(), false),
293        ColumnSchema::new(PROVENANCE_COLUMN, string(), false),
294        ColumnSchema::new(
295            CONFIDENCE_COLUMN,
296            ConcreteDataType::float64_datatype(),
297            true,
298        ),
299        ColumnSchema::new(
300            REQUEST_COUNT_COLUMN,
301            ConcreteDataType::int64_datatype(),
302            true,
303        ),
304        ColumnSchema::new(
305            UNMATCHED_COUNT_COLUMN,
306            ConcreteDataType::int64_datatype(),
307            true,
308        ),
309        ColumnSchema::new(ERROR_COUNT_COLUMN, ConcreteDataType::int64_datatype(), true),
310        ColumnSchema::new(
311            DURATION_SUM_COLUMN,
312            ConcreteDataType::float64_datatype(),
313            true,
314        ),
315        ColumnSchema::new(
316            DURATION_COUNT_COLUMN,
317            ConcreteDataType::int64_datatype(),
318            true,
319        ),
320        ColumnSchema::new(
321            DURATION_MAX_COLUMN,
322            ConcreteDataType::float64_datatype(),
323            true,
324        ),
325        ColumnSchema::new(EDGE_ATTRIBUTES_COLUMN, json(), true),
326    ]))
327});
328
329/// Which computed table this shell represents, so the two share one forwarder.
330#[derive(Clone, Copy)]
331enum GraphTableKind {
332    Entities,
333    Relationships,
334}
335
336/// Forwarder for a computed entity-graph table.
337struct SemanticGraphTable {
338    kind: GraphTableKind,
339    schema: SchemaRef,
340    catalog_name: String,
341    catalog_manager: Weak<dyn CatalogManager>,
342    query_ctx: Option<QueryContextRef>,
343}
344
345impl SemanticGraphTable {
346    fn new(
347        kind: GraphTableKind,
348        catalog_name: String,
349        catalog_manager: Weak<dyn CatalogManager>,
350        query_ctx: Option<QueryContextRef>,
351    ) -> Self {
352        let schema = match kind {
353            GraphTableKind::Entities => ENTITIES_SCHEMA.clone(),
354            GraphTableKind::Relationships => RELATIONSHIPS_SCHEMA.clone(),
355        };
356        Self {
357            kind,
358            schema,
359            catalog_name,
360            catalog_manager,
361            query_ctx,
362        }
363    }
364
365    async fn derive(
366        kind: GraphTableKind,
367        catalog: String,
368        catalog_manager: Weak<dyn CatalogManager>,
369        request: ScanRequest,
370        query_ctx: Option<QueryContextRef>,
371    ) -> Result<Option<SendableRecordBatchStream>> {
372        let provider = utils::entity_graph_provider(&catalog_manager)?;
373        // No provider (engine not up / non-frontend node): stream empty.
374        let Some(provider) = provider else {
375            return Ok(None);
376        };
377        match kind {
378            GraphTableKind::Entities => provider.scan_entities(&catalog, request, query_ctx).await,
379            GraphTableKind::Relationships => {
380                provider
381                    .scan_relationships(&catalog, request, query_ctx)
382                    .await
383            }
384        }
385        .context(InternalSnafu)
386    }
387
388    fn align_schema(
389        stream: SendableRecordBatchStream,
390        schema: SchemaRef,
391    ) -> SendableRecordBatchStream {
392        let batch_schema = schema.clone();
393        let arrow_schema = schema.arrow_schema().clone();
394        let batches = stream.map(move |batch| {
395            let batch = batch?;
396            // The derivation output is structurally identical, but its JSON
397            // fields are plain Binary without the declared extension metadata.
398            let batch = DfRecordBatch::try_new(
399                arrow_schema.clone(),
400                batch.into_df_record_batch().columns().to_vec(),
401            )
402            .context(common_recordbatch::error::NewDfRecordBatchSnafu)?;
403            Ok(RecordBatch::from_df_record_batch(
404                batch_schema.clone(),
405                batch,
406            ))
407        });
408        Box::pin(RecordBatchStreamWrapper::new(schema, Box::pin(batches)))
409    }
410}
411
412impl SystemTable for SemanticGraphTable {
413    fn table_id(&self) -> TableId {
414        match self.kind {
415            GraphTableKind::Entities => SEMANTIC_ENTITIES_TABLE_ID,
416            GraphTableKind::Relationships => SEMANTIC_RELATIONSHIPS_TABLE_ID,
417        }
418    }
419
420    fn table_name(&self) -> &'static str {
421        match self.kind {
422            GraphTableKind::Entities => SEMANTIC_ENTITIES,
423            GraphTableKind::Relationships => SEMANTIC_RELATIONSHIPS,
424        }
425    }
426
427    fn schema(&self) -> SchemaRef {
428        self.schema.clone()
429    }
430
431    fn to_stream(&self, request: ScanRequest) -> Result<SendableRecordBatchStream> {
432        let schema = self.schema.clone();
433        let kind = self.kind;
434        let catalog = self.catalog_name.clone();
435        let catalog_manager = self.catalog_manager.clone();
436        let query_ctx = self.query_ctx.clone();
437
438        let stream_schema = schema.clone();
439        let stream = async move {
440            let stream = Self::derive(kind, catalog, catalog_manager, request, query_ctx)
441                .await
442                .map_err(BoxedError::new)
443                .context(common_recordbatch::error::ExternalSnafu)?;
444            Ok(match stream {
445                Some(stream) => Self::align_schema(stream, stream_schema.clone()),
446                None => Box::pin(EmptyRecordBatchStream::new(stream_schema.clone())),
447            })
448        };
449
450        Ok(Box::pin(AsyncRecordBatchStreamAdapter::new(
451            schema,
452            Box::pin(stream),
453        )))
454    }
455}
456
457#[cfg(test)]
458mod tests {
459    use super::*;
460
461    #[test]
462    fn graph_tables_use_observed_at_as_time_index() {
463        for schema in [&*ENTITIES_SCHEMA, &*RELATIONSHIPS_SCHEMA] {
464            assert_eq!(
465                schema.timestamp_column().map(|column| column.name.as_str()),
466                Some("observed_at")
467            );
468        }
469    }
470
471    #[test]
472    fn relationship_schema_does_not_expose_generation_id() {
473        assert!(
474            RELATIONSHIPS_SCHEMA
475                .column_schema_by_name("generation_id")
476                .is_none()
477        );
478    }
479}
catalog/system_schema/semantic_graph.rs

catalog/system_schema/
semantic_graph.rs
