diff --git a/docs/rfcs/2026-05-28-table-semantic-layer.md b/docs/rfcs/2026-05-28-table-semantic-layer.md index 4c646e5e51..e1846f6b25 100644 --- a/docs/rfcs/2026-05-28-table-semantic-layer.md +++ b/docs/rfcs/2026-05-28-table-semantic-layer.md @@ -44,7 +44,7 @@ The audience is broader than LLM agents. Alert generators need to choose between 1. **`greptime.semantic.*` table options** — table-level identity and lineage. Carried inside the existing `table_options` blob. This is the same slot that today carries `table_data_model = 'greptime_trace_v1'` and `otlp_metric_compat = 'prom'`, so the mechanism is generalising what the OTLP trace auto-create path already does. 2. **Column `COMMENT`** — column-level supplements ("this column is `resource.service.name`"; "this column carries delta values"). Standard SQL. -3. **`information_schema.table_semantics` view** — a denormalised projection of the options, registered through the existing `with_extra_table_factories()` hook. Tables without a `greptime.semantic.*` option do not appear in the view. +3. **`information_schema.table_semantics` view** — a denormalised projection of the options, registered through the existing `with_extra_table_factories()` hook. A table appears in the view when it carries a `greptime.semantic.*` option or when the built-in conventions derive entities from it. ## Vocabulary @@ -102,7 +102,7 @@ SELECT table_catalog, table_schema, table_name, signal_type, source, pipeline FROM information_schema.table_semantics; ``` -returns one row per semantic-tagged table. The view exposes a stable set of core columns (`table_catalog`, `table_schema`, `table_name`, `signal_type`, `source`, `source_version`, `pipeline`) plus a `semantic_options` JSON column carrying the rest of the `greptime.semantic.*` keys verbatim. Future keys appear inside `semantic_options` without forcing a view-schema change; only widely-used keys are ever promoted to first-class columns. +returns one row per semantic-tagged table. The view exposes a stable set of core columns (`table_catalog`, `table_schema`, `table_name`, `signal_type`, `source`, `source_version`, `pipeline`) plus a `semantic_options` JSON column carrying the rest of the `greptime.semantic.*` keys verbatim and an `entity_declarations` JSON column listing the entities the table contributes to the graph, whether declared by option or derived by convention. Future keys appear inside `semantic_options` without forcing a view-schema change; only widely-used keys are ever promoted to first-class columns. # Implementation Plan diff --git a/docs/rfcs/2026-06-25-entity-relationships-and-graph-query.md b/docs/rfcs/2026-06-25-entity-relationships-and-graph-query.md index 1c78e91f25..3de7e4b35d 100644 --- a/docs/rfcs/2026-06-25-entity-relationships-and-graph-query.md +++ b/docs/rfcs/2026-06-25-entity-relationships-and-graph-query.md @@ -139,7 +139,7 @@ The pack's metadata quality depends on Remote Write 2.0's inline per-series meta The graph is exposed as two **computed, read-only tables** under `greptime_private`, not `information_schema`, because scanning them triggers real derivation over telemetry tables, which would break the cheap-metadata expectation of `information_schema`. All DDL/DML against them is rejected on every write path. -`semantic_entities` is the node set: one row per **distinct projected entity observation from a contributing table**, carrying the observation window, the entity's type and id (plus the structured `entity_id_attrs` for composite identities), a descriptive snapshot, and the `source_tables` lineage. An entity declared by three tables yields at least three rows per window. +`semantic_entities` is the node set: one row per **distinct projected entity observation from a contributing table**, carrying the observation window, the entity's type and id (plus the structured `entity_id_attrs`), a descriptive snapshot, and the `source_tables` lineage. An entity declared by three tables yields at least three rows per window. `semantic_relationships` is the edge set: one row per **(window, edge)**, carrying the endpoints, a `rel_type` from the vocabulary below, `provenance` (`trace` | `attribute` | `declared` | `agent`), `confidence`, RED metrics for `calls` edges, and a JSON `attributes` column. @@ -148,7 +148,7 @@ The decisions behind the shape: - **Edges are time-ranged facts.** A row asserts an edge *existed in a window*, with RED metrics for that window — the TSDB-native answer to "edges expire". "The topology now" is `WHERE fresh_until >= now() - INTERVAL '5m'`. - **`provenance` is part of edge identity**, so a declared edge and a derived edge between the same pair coexist, and a user-declared edge survives when derivation is off. - **`confidence` expresses derivation certainty, not statistical completeness**: `1.0` for a successfully paired or declared edge, `< 1.0` for virtual-node or agent-inferred edges. It does not correct for trace sampling. -- **RED metrics describe the observed span-pair population** — the `service_graph` connector semantics. Under sampling, counts understate true traffic, and ratios (error rate, mean duration) are representative only when sampling is unbiased with respect to status and latency — tail sampling that keeps errors and slow traces skews both. +- **RED metrics describe the observed span-pair population** — the `service_graph` connector semantics. Under sampling, counts understate true traffic, and ratios (error rate, mean duration) are representative only when sampling is unbiased with respect to status and latency — tail sampling that keeps errors and slow traces skews both. Clients left out of that population are counted separately on the same row, or a callee that stopped responding reads as a caller that stopped calling. - **Endpoint encoding.** An identity renders as its values in declared order (broad to narrow), joined by `,` with `,` and `\` escaped in each value. The identifying attribute *keys* are deliberately absent: the same identity reaches the graph under different column names per signal — a trace table's `service_name` against a metric table's `job` — and encoding the keys would split one entity into one entity per signal. Where a source carries a qualifier separately (`service.namespace` beside `service.name`) the convention folds it in as `/`, reproducing the identity a source that pre-composed it (Prometheus `job`) already emits. Nodes carry the structured `entity_id_attrs`; edges carry only the string key. - **Hand-declared edges** (`provenance = 'declared'`) live in a physical table of the same shape (plus a business validity window) in `greptime_private` and are unioned into the computed `semantic_relationships`. @@ -197,7 +197,7 @@ GROUP BY 1, src_id, dst_id; Edge endpoints are built from each trace table's `service` entity declaration (the same id expression the registry uses), so edges land on exactly the entity ids the registry emits, self-calls are judged on the full endpoint id, and a table whose declaration is unusable contributes no edges rather than silently switching identity. The join is bounded by a time-proximity window and the results of **all** trace tables combine into one edge per window with merged RED metrics. One assumption to note: a deployment can route traces to multiple tables (`x-greptime-trace-table-name`), so one distributed trace is not guaranteed to live in a single table — pairing must account for that, and how is an implementation concern. Uninstrumented peers become **virtual nodes**: a client span with no matching server span yields an edge to a synthetic node named from `peer.service`/`db.name`/`server.address`. -**The entity registry.** Per declaring table: filter to the window, project (time bin, casts, JSON assembly), `DISTINCT`. The per-table results are unioned without any cross-source merge: no descriptive-conflict or JSON-merging policy lives in the derivation. Rows whose identity columns are NULL identify nothing and are filtered out; ids render as their escaped values joined in declared order, with the `entity_id_attrs` JSON alongside once an id is assembled from more than one column. Consumers deduplicate with `SELECT DISTINCT entity_type, entity_id`; a unique node set is the snapshot relation's job (see *Querying the graph*). +**The entity registry.** Per declaring table: filter to the window, project (time bin, casts, JSON assembly), `DISTINCT`. The per-table results are unioned without any cross-source merge: no descriptive-conflict or JSON-merging policy lives in the derivation. Rows whose identity columns are NULL identify nothing and are filtered out; ids render as their escaped values joined in declared order, with the `entity_id_attrs` JSON alongside naming the attributes they were assembled from. A declaration may name a more specific type that supersedes it: on rows carrying that type's full identity the declaration withdraws, so one container is either the generic or the Kubernetes entity but never both, and never neither. Consumers deduplicate with `SELECT DISTINCT entity_type, entity_id`; a unique node set is the snapshot relation's job (see *Querying the graph*). **Attribute edges.** Shared attributes provide join keys; co-declaration or a relationship template provides the relationship semantics. A shared value alone determines neither direction nor edge type, so the derivation is rule-based, not join-everything. A table declaring both a `service.instance` and a `host` identity on the same rows derives `runs_on` between them: the row itself witnesses the relationship, and the built-in vocabulary fixes the direction. Cross-table edges require a declared relationship template (endpoint mappings, `rel_type`, direction): a derivation *rule* declared on schema, not per-instance edge data. diff --git a/src/catalog/src/system_schema/information_schema/table_semantics.rs b/src/catalog/src/system_schema/information_schema/table_semantics.rs index 62587e80c4..07f1a844f9 100644 --- a/src/catalog/src/system_schema/information_schema/table_semantics.rs +++ b/src/catalog/src/system_schema/information_schema/table_semantics.rs @@ -13,16 +13,21 @@ // limitations under the License. //! `information_schema.table_semantics`: the queryable view over the table -//! semantic layer. One row per table that carries at least one -//! `greptime.semantic.*` option, so a consumer can discover the observability -//! concept a table stands for with a single SQL query instead of parsing every -//! table's `create_options`. +//! semantic layer. One row per table that is part of it, so a consumer can +//! discover the observability concept a table stands for with a single SQL +//! query instead of parsing every table's `create_options`. //! //! The few signal-agnostic keys are promoted to their own columns //! (`signal_type` / `source` / `source_version` / `pipeline` / //! `metadata_quality`); the remaining signal-specific keys are folded into a //! `semantic_options` JSON string, keyed by the option name with the //! `greptime.semantic.` prefix stripped. +//! +//! `entity_declarations` reports the entities the table contributes to the +//! graph, including the ones the conventions derive with no option set — so a +//! table with no semantic option still gets a row. It shows the outcome, not +//! the reasoning: a declaration dropped for naming a missing column is simply +//! absent, and only the log says why. use std::collections::BTreeMap; use std::sync::{Arc, Weak}; @@ -41,6 +46,7 @@ use datatypes::schema::{ColumnSchema, Schema, SchemaRef}; use datatypes::value::Value; use datatypes::vectors::{StringVectorBuilder, UInt32VectorBuilder}; use futures::TryStreamExt; +use serde::Serialize; use snafu::{OptionExt, ResultExt}; use store_api::storage::{ScanRequest, TableId}; use table::metadata::TableInfo; @@ -54,6 +60,8 @@ use crate::error::{ CreateRecordBatchSnafu, InternalSnafu, Result, UpgradeWeakCatalogManagerRefSnafu, }; use crate::system_schema::information_schema::{InformationTable, Predicates, TABLE_SEMANTICS}; +use crate::system_schema::semantic_graph::{EntityGraphProviderRef, TableEntityDeclaration}; +use crate::system_schema::utils; pub const TABLE_CATALOG: &str = "table_catalog"; pub const TABLE_SCHEMA: &str = "table_schema"; @@ -65,6 +73,7 @@ pub const SOURCE_VERSION: &str = "source_version"; pub const PIPELINE: &str = "pipeline"; pub const METADATA_QUALITY: &str = "metadata_quality"; pub const SEMANTIC_OPTIONS: &str = "semantic_options"; +pub const ENTITY_DECLARATIONS: &str = "entity_declarations"; const INIT_CAPACITY: usize = 42; @@ -72,8 +81,49 @@ fn optional_value(v: Option<&str>) -> Value { v.map(Value::from).unwrap_or(Value::Null) } +/// The JSON form of one declaration in the `entity_declarations` column. +#[derive(Serialize)] +struct DeclarationEntry { + entity_type: String, + origin: &'static str, + id: Vec, + #[serde(skip_serializing_if = "Option::is_none")] + id_qualifier: Option, + #[serde(skip_serializing_if = "Vec::is_empty")] + superseded_by: Vec, + #[serde(skip_serializing_if = "Vec::is_empty")] + descriptive: Vec, + #[serde(skip_serializing_if = "Vec::is_empty")] + scope: Vec, +} + +impl From for DeclarationEntry { + fn from(declaration: TableEntityDeclaration) -> Self { + Self { + entity_type: declaration.entity_type, + origin: declaration.origin.as_str(), + id: declaration.id_columns, + id_qualifier: declaration.id_qualifier, + superseded_by: declaration.superseded_by_columns, + descriptive: declaration.descriptive_columns, + scope: declaration.scope_columns, + } + } +} + +fn entity_declarations_json(declarations: Vec) -> Option { + if declarations.is_empty() { + return None; + } + let entries: Vec = declarations.into_iter().map(Into::into).collect(); + // Fold a failure into `None` rather than panicking the query path, as the + // options tail does. + serde_json::to_string(&entries).ok() +} + /// The semantic projection of a single table: the signal-agnostic keys promoted /// to columns, plus a JSON tail for the rest. Borrows from the table's options. +#[derive(Default)] struct SemanticRow<'a> { signal_type: Option<&'a str>, source: Option<&'a str>, @@ -167,6 +217,11 @@ impl InformationSchemaTableSemantics { ColumnSchema::new(PIPELINE, ConcreteDataType::string_datatype(), true), ColumnSchema::new(METADATA_QUALITY, ConcreteDataType::string_datatype(), true), ColumnSchema::new(SEMANTIC_OPTIONS, ConcreteDataType::string_datatype(), true), + ColumnSchema::new( + ENTITY_DECLARATIONS, + ConcreteDataType::string_datatype(), + true, + ), ])) } @@ -228,6 +283,7 @@ struct InformationSchemaSemanticTablesBuilder { pipelines: StringVectorBuilder, metadata_qualities: StringVectorBuilder, semantic_options: StringVectorBuilder, + entity_declarations: StringVectorBuilder, } impl InformationSchemaSemanticTablesBuilder { @@ -250,6 +306,7 @@ impl InformationSchemaSemanticTablesBuilder { pipelines: StringVectorBuilder::with_capacity(INIT_CAPACITY), metadata_qualities: StringVectorBuilder::with_capacity(INIT_CAPACITY), semantic_options: StringVectorBuilder::with_capacity(INIT_CAPACITY), + entity_declarations: StringVectorBuilder::with_capacity(INIT_CAPACITY), } } @@ -260,11 +317,20 @@ impl InformationSchemaSemanticTablesBuilder { .upgrade() .context(UpgradeWeakCatalogManagerRefSnafu)?; let predicates = Predicates::from_scan_request(&request); + // Resolved once for the whole scan: the lookup downcasts the catalog + // manager, while the per-table call behind it is pure metadata work. + let graph_provider = utils::entity_graph_provider(&self.catalog_manager)?; for schema_name in catalog_manager.schema_names(&catalog_name, None).await? { let mut table_stream = catalog_manager.tables(&catalog_name, &schema_name, None); while let Some(table) = table_stream.try_next().await? { - self.add_table(&predicates, &catalog_name, &schema_name, table.table_info()); + self.add_table( + &predicates, + &catalog_name, + &schema_name, + table.table_info(), + graph_provider.as_ref(), + ); } } @@ -277,11 +343,11 @@ impl InformationSchemaSemanticTablesBuilder { catalog_name: &str, schema_name: &str, table_info: Arc, + graph_provider: Option<&EntityGraphProviderRef>, ) { - // A table with no semantic key is not part of the semantic layer. - let Some(row) = SemanticRow::extract(&table_info) else { - return; - }; + let semantic_row = SemanticRow::extract(&table_info); + let carries_options = semantic_row.is_some(); + let row = semantic_row.unwrap_or_default(); let table_name = table_info.name.as_ref(); let catalog_v = Value::from(catalog_name); @@ -306,6 +372,13 @@ impl InformationSchemaSemanticTablesBuilder { return; } + let declarations_json = graph_provider + .map(|provider| provider.table_declarations(&table_info)) + .and_then(entity_declarations_json); + if !carries_options && declarations_json.is_none() { + return; + } + self.catalog_names.push(Some(catalog_name)); self.schema_names.push(Some(schema_name)); self.table_names.push(Some(table_name)); @@ -316,6 +389,7 @@ impl InformationSchemaSemanticTablesBuilder { self.pipelines.push(row.pipeline); self.metadata_qualities.push(row.metadata_quality); self.semantic_options.push(row.options_json.as_deref()); + self.entity_declarations.push(declarations_json.as_deref()); } fn finish(&mut self) -> Result { @@ -330,6 +404,7 @@ impl InformationSchemaSemanticTablesBuilder { Arc::new(self.pipelines.finish()), Arc::new(self.metadata_qualities.finish()), Arc::new(self.semantic_options.finish()), + Arc::new(self.entity_declarations.finish()), ]; RecordBatch::new(self.schema.clone(), columns).context(CreateRecordBatchSnafu) } diff --git a/src/catalog/src/system_schema/semantic_graph.rs b/src/catalog/src/system_schema/semantic_graph.rs index b25dbb0328..977a57e90b 100644 --- a/src/catalog/src/system_schema/semantic_graph.rs +++ b/src/catalog/src/system_schema/semantic_graph.rs @@ -36,13 +36,13 @@ use std::sync::{Arc, LazyLock, Weak}; use common_catalog::consts::{ CONFIDENCE_COLUMN, DEFAULT_PRIVATE_SCHEMA_NAME, DST_ID_COLUMN, DST_TYPE_COLUMN, - DURATION_COUNT_COLUMN, DURATION_SUM_COLUMN, EDGE_ATTRIBUTES_COLUMN, ENTITY_DESCRIPTIVE_COLUMN, - ENTITY_ID_ATTRS_COLUMN, ENTITY_ID_COLUMN, ENTITY_SCOPE_COLUMN, ENTITY_TYPE_COLUMN, - ERROR_COUNT_COLUMN, FRESH_UNTIL_COLUMN, OBSERVED_AT_COLUMN, PROVENANCE_COLUMN, REL_TYPE_COLUMN, - REQUEST_COUNT_COLUMN, SEMANTIC_ENTITIES_TABLE_ID, + DURATION_COUNT_COLUMN, DURATION_MAX_COLUMN, DURATION_SUM_COLUMN, EDGE_ATTRIBUTES_COLUMN, + ENTITY_DESCRIPTIVE_COLUMN, ENTITY_ID_ATTRS_COLUMN, ENTITY_ID_COLUMN, ENTITY_SCOPE_COLUMN, + ENTITY_TYPE_COLUMN, ERROR_COUNT_COLUMN, FRESH_UNTIL_COLUMN, OBSERVED_AT_COLUMN, + PROVENANCE_COLUMN, REL_TYPE_COLUMN, REQUEST_COUNT_COLUMN, SEMANTIC_ENTITIES_TABLE_ID, SEMANTIC_ENTITIES_TABLE_NAME as SEMANTIC_ENTITIES, SEMANTIC_RELATIONSHIPS_TABLE_ID, SEMANTIC_RELATIONSHIPS_TABLE_NAME as SEMANTIC_RELATIONSHIPS, SOURCE_TABLES_COLUMN, - SRC_ID_COLUMN, SRC_TYPE_COLUMN, WINDOW_END_COLUMN, WINDOW_START_COLUMN, + SRC_ID_COLUMN, SRC_TYPE_COLUMN, UNMATCHED_COUNT_COLUMN, WINDOW_END_COLUMN, WINDOW_START_COLUMN, }; use common_error::ext::BoxedError; use common_recordbatch::adapter::AsyncRecordBatchStreamAdapter; @@ -57,6 +57,7 @@ use session::context::QueryContextRef; use snafu::ResultExt; use store_api::storage::{ScanRequest, TableId}; use table::TableRef; +use table::metadata::TableInfo; use crate::CatalogManager; use crate::error::{InternalSnafu, Result}; @@ -64,6 +65,38 @@ use crate::system_schema::{SystemSchemaProviderInner, SystemTable, SystemTableRe pub type EntityGraphProviderRef = Arc; +/// Where a table's entity declaration came from. +pub enum DeclarationOrigin { + /// A `greptime.semantic.entity..*` table option. + Declared, + /// The built-in derivation conventions shipped with the binary. + Convention, +} + +impl DeclarationOrigin { + pub fn as_str(&self) -> &'static str { + match self { + Self::Declared => "declared", + Self::Convention => "convention", + } + } +} + +/// One entity a table declares, as `information_schema.table_semantics` +/// reports it. A catalog-side projection of the derivation's own declaration +/// type, which lives above this crate. +pub struct TableEntityDeclaration { + pub entity_type: String, + pub origin: DeclarationOrigin, + pub id_columns: Vec, + pub id_qualifier: Option, + /// Columns whose presence on a row withdraws this declaration for that row. + /// Reported because the declaration otherwise reads as unconditional. + pub superseded_by_columns: Vec, + pub descriptive_columns: Vec, + pub scope_columns: Vec, +} + /// Produces the rows of the computed entity-graph tables at read time. /// /// Implemented above the query engine (in the frontend) and injected into the @@ -94,6 +127,12 @@ pub trait EntityGraphProvider: Send + Sync { request: ScanRequest, query_ctx: Option, ) -> std::result::Result, BoxedError>; + + /// The entities `table_info` contributes to the graph: its explicit + /// declarations merged with the ones the conventions derive. Metadata-only + /// by contract — it runs per table on the `table_semantics` scan and must + /// not touch the query engine. + fn table_declarations(&self, table_info: &TableInfo) -> Vec; } /// Serves the computed graph tables under `greptime_private`, overlaid on the @@ -185,10 +224,10 @@ fn json() -> ConcreteDataType { /// observed evidence there). /// - `entity_type` — the entity's type, e.g. `service`, `host`, `k8s.pod`, /// `process`, `service.instance` (the OTel-style, possibly dotted, type). -/// - `entity_id` — canonical identifier: the value verbatim for a -/// single-attribute identity, or a sorted `k=v,k=v` rendering for a composite. -/// - `entity_id_attrs` — JSON object of the identifying attributes (the -/// escaping-safe source of truth for composite ids); NULL for single-attribute ids. +/// - `entity_id` — canonical identifier: the identifying values in declared +/// order, escaped and joined. +/// - `entity_id_attrs` — JSON object of the identifying attributes, so a +/// consumer holding an id can tell which columns it came from. /// - `scope` — namespace/environment the id is scoped to; empty when none. /// - `descriptive` — JSON snapshot of the entity's descriptive (non-identifying) /// attributes; NULL when no descriptive columns were declared. @@ -209,7 +248,7 @@ static ENTITIES_SCHEMA: LazyLock = LazyLock::new(|| { }); /// Schema of `semantic_relationships` — the edge set of the graph, one row per -/// edge observed in a time window. This is the 16-column contract every derived +/// edge observed in a time window. This is the 18-column contract every derived /// branch and the declared-edge table must project for the top-level `UNION ALL`. /// /// Columns: @@ -229,10 +268,15 @@ static ENTITIES_SCHEMA: LazyLock = LazyLock::new(|| { /// declared edges, lower for virtual-node or agent-inferred edges. It does /// not correct for trace sampling. /// - `request_count` — RED: number of requests over the window (`calls` edges). +/// - `unmatched_count` — client spans on this edge with no server span. The +/// other RED columns describe the pairs alone, so this is what separates a +/// callee that stopped responding from traffic that stopped arriving. /// - `error_count` — RED: number of errored requests over the window. /// - `duration_sum` — RED: sum of request durations, in seconds, over the window. /// - `duration_count`— RED: number of durations summed (pair with `duration_sum` /// to get an average). +/// - `duration_max` — RED: longest single request, in seconds, over the same +/// population `duration_sum` covers. /// - `attributes` — JSON of edge attributes, e.g. `connection_type`, /// `db.system`, `peer.service`. static RELATIONSHIPS_SCHEMA: LazyLock = LazyLock::new(|| { @@ -257,6 +301,11 @@ static RELATIONSHIPS_SCHEMA: LazyLock = LazyLock::new(|| { ConcreteDataType::int64_datatype(), true, ), + ColumnSchema::new( + UNMATCHED_COUNT_COLUMN, + ConcreteDataType::int64_datatype(), + true, + ), ColumnSchema::new(ERROR_COUNT_COLUMN, ConcreteDataType::int64_datatype(), true), ColumnSchema::new( DURATION_SUM_COLUMN, @@ -268,6 +317,11 @@ static RELATIONSHIPS_SCHEMA: LazyLock = LazyLock::new(|| { ConcreteDataType::int64_datatype(), true, ), + ColumnSchema::new( + DURATION_MAX_COLUMN, + ConcreteDataType::float64_datatype(), + true, + ), ColumnSchema::new(EDGE_ATTRIBUTES_COLUMN, json(), true), ])) }); diff --git a/src/common/catalog/src/consts.rs b/src/common/catalog/src/consts.rs index 63aa103b83..abb8f0007d 100644 --- a/src/common/catalog/src/consts.rs +++ b/src/common/catalog/src/consts.rs @@ -245,9 +245,11 @@ pub const REL_TYPE_COLUMN: &str = "rel_type"; pub const PROVENANCE_COLUMN: &str = "provenance"; pub const CONFIDENCE_COLUMN: &str = "confidence"; pub const REQUEST_COUNT_COLUMN: &str = "request_count"; +pub const UNMATCHED_COUNT_COLUMN: &str = "unmatched_count"; pub const ERROR_COUNT_COLUMN: &str = "error_count"; pub const DURATION_SUM_COLUMN: &str = "duration_sum"; pub const DURATION_COUNT_COLUMN: &str = "duration_count"; +pub const DURATION_MAX_COLUMN: &str = "duration_max"; pub const EDGE_ATTRIBUTES_COLUMN: &str = "attributes"; // Declared-edge table only. pub const VALID_FROM_COLUMN: &str = "valid_from"; diff --git a/src/frontend/src/instance/entity_graph.rs b/src/frontend/src/instance/entity_graph.rs index a0817cf280..eadc6313d0 100644 --- a/src/frontend/src/instance/entity_graph.rs +++ b/src/frontend/src/instance/entity_graph.rs @@ -32,7 +32,9 @@ use auth::{ PermissionTableTargets, SEMANTIC_GRAPH_QUERY, }; use catalog::CatalogManager; -use catalog::system_schema::semantic_graph::EntityGraphProvider; +use catalog::system_schema::semantic_graph::{ + DeclarationOrigin, EntityGraphProvider, TableEntityDeclaration, +}; use common_catalog::consts::{ DEFAULT_PRIVATE_SCHEMA_NAME, DEFAULT_SCHEMA_NAME, INFORMATION_SCHEMA_NAME, OBSERVED_AT_COLUMN, PG_CATALOG_NAME, SEMANTIC_RELATIONSHIPS_DECLARED_TABLE_NAME, @@ -127,15 +129,6 @@ impl EntityGraphProviderImpl { /// Parses `greptime.semantic.entity..{id|descriptive|scope}` options of /// one table into per-type declarations. A type with no `id` columns is skipped. fn parse_declarations(table_info: &TableInfo) -> Vec { - let Some(time_index) = table_info - .meta - .schema - .timestamp_column() - .map(|c| c.name.clone()) - else { - return vec![]; - }; - // entity_type -> (id_columns, descriptive_columns, scope_columns) type RoleColumns = (Vec, Vec, Vec); let mut by_type: HashMap = HashMap::new(); @@ -151,6 +144,17 @@ impl EntityGraphProviderImpl { EntityRole::Scope => entry.2 = cols, } } + if by_type.is_empty() { + return vec![]; + } + let Some(time_index) = table_info + .meta + .schema + .timestamp_column() + .map(|c| c.name.clone()) + else { + return vec![]; + }; by_type .into_iter() @@ -179,6 +183,7 @@ impl EntityGraphProviderImpl { entity_type, id_columns, id_qualifier: None, + superseded_by_columns: vec![], descriptive_columns, scope_columns, }) @@ -200,11 +205,13 @@ impl EntityGraphProviderImpl { conventions: &Conventions, ) -> Vec { let mut declarations = Self::parse_declarations(table_info); + let mut supersessions = Vec::new(); if is_trace_v1_table(table_info) { Self::extend_with_implicit_entities( table_info, &conventions.otlp_trace_entities, &mut declarations, + &mut supersessions, ); } Self::extend_with_info_metric_conventions( @@ -213,6 +220,7 @@ impl EntityGraphProviderImpl { SOURCE_PROMETHEUS, None, &mut declarations, + &mut supersessions, ); Self::extend_with_info_metric_conventions( table_info, @@ -220,10 +228,30 @@ impl EntityGraphProviderImpl { SOURCE_OPENTELEMETRY, Some(servers::semantic::METRIC_TYPE_INFO), &mut declarations, + &mut supersessions, ); + Self::resolve_supersessions(&mut declarations, supersessions); declarations } + /// Binds each `superseded_by` to the identity the superseding type has on + /// this table, once every declaration is known. A type nothing declares + /// here leaves the guard empty, so the superseded entity stands instead of + /// yielding to a node that will never be derived. + fn resolve_supersessions( + declarations: &mut [EntityDeclaration], + supersessions: Vec<(usize, String)>, + ) { + for (index, entity_type) in supersessions { + let identity = declarations + .iter() + .find(|declaration| declaration.entity_type == entity_type) + .map(|declaration| declaration.id_columns.clone()) + .unwrap_or_default(); + declarations[index].superseded_by_columns = identity; + } + } + /// Whether the table carries an explicit `entity..id` option. fn explicitly_declares(table_info: &TableInfo, entity_type: &str) -> bool { table_info.meta.options.extra_options.keys().any(|key| { @@ -245,6 +273,7 @@ impl EntityGraphProviderImpl { expected_source: &str, expected_metric_type: Option<&str>, declarations: &mut Vec, + supersessions: &mut Vec<(usize, String)>, ) { let Some(implicit_entities) = whitelist.get(&table_info.name) else { return; @@ -264,7 +293,12 @@ impl EntityGraphProviderImpl { ); return; } - Self::extend_with_implicit_entities(table_info, implicit_entities, declarations); + Self::extend_with_implicit_entities( + table_info, + implicit_entities, + declarations, + supersessions, + ); } /// Synthesizes the applicable subset of `entities` on `table_info`: @@ -274,6 +308,7 @@ impl EntityGraphProviderImpl { table_info: &TableInfo, entities: &[ImplicitEntity], declarations: &mut Vec, + supersessions: &mut Vec<(usize, String)>, ) { let schema = &table_info.meta.schema; let Some(time_index) = schema.timestamp_column().map(|c| c.name.clone()) else { @@ -323,6 +358,9 @@ impl EntityGraphProviderImpl { .qualified_by .clone() .filter(|c| schema.column_schema_by_name(c).is_some()); + if let Some(entity_type) = &implicit.superseded_by { + supersessions.push((declarations.len(), entity_type.clone())); + } declarations.push(EntityDeclaration { schema: table_info.schema_name.clone(), table: table_info.name.clone(), @@ -330,6 +368,7 @@ impl EntityGraphProviderImpl { entity_type: implicit.entity.clone(), id_columns: implicit.id.clone(), id_qualifier, + superseded_by_columns: vec![], descriptive_columns, scope_columns: vec![], }); @@ -650,6 +689,33 @@ impl EntityGraphProvider for EntityGraphProviderImpl { }; self.execute_plan(catalog, plan, query_ctx).await } + + fn table_declarations(&self, table_info: &TableInfo) -> Vec { + // A broken embedded file still leaves the explicit half reportable; + // the scan paths surface the error itself. + let derived = match conventions() { + Ok(conventions) => Self::declarations_for(table_info, conventions), + Err(_) => Self::parse_declarations(table_info), + }; + let mut declarations = derived + .into_iter() + .map(|declaration| TableEntityDeclaration { + origin: if Self::explicitly_declares(table_info, &declaration.entity_type) { + DeclarationOrigin::Declared + } else { + DeclarationOrigin::Convention + }, + entity_type: declaration.entity_type, + id_columns: declaration.id_columns, + id_qualifier: declaration.id_qualifier, + superseded_by_columns: declaration.superseded_by_columns, + descriptive_columns: declaration.descriptive_columns, + scope_columns: declaration.scope_columns, + }) + .collect::>(); + declarations.sort_by(|a, b| a.entity_type.cmp(&b.entity_type)); + declarations + } } #[cfg(test)] @@ -928,6 +994,43 @@ mod tests { vec!["resource_attributes.host.name"] ); + // A wrong `resource_attributes.` prefix would leave the generic + // container standing beside the k8s one, which the descriptor-table + // case cannot catch. + let pod_container = table_info( + &[ + "service_name", + "resource_attributes.container.id", + "resource_attributes.container.name", + "resource_attributes.k8s.pod.uid", + "resource_attributes.k8s.container.name", + ], + &[(TABLE_DATA_MODEL, TABLE_DATA_MODEL_TRACE_V1)], + ); + let declarations = sorted_declarations(&pod_container); + let types: Vec<&str> = declarations + .iter() + .map(|d| d.entity_type.as_str()) + .collect(); + assert_eq!( + types, + vec!["container", "k8s.container", "k8s.pod", "service"] + ); + assert_eq!( + declarations[0].superseded_by_columns, + vec![ + "resource_attributes.k8s.pod.uid", + "resource_attributes.k8s.container.name" + ] + ); + assert_eq!( + declarations[1].descriptive_columns, + vec![ + "resource_attributes.container.id", + "resource_attributes.container.name" + ] + ); + let names_only = table_info( &[ "service_name", @@ -1008,6 +1111,7 @@ mod tests { "container.name", "k8s.pod.uid", "k8s.pod.name", + "k8s.container.name", "k8s.namespace.name", ], OTEL_STAMPS, @@ -1022,20 +1126,60 @@ mod tests { vec![ "container", "host", + "k8s.container", "k8s.pod", "service", "service.instance" ] ); + // A pod's container is the k8s.container, under the identity + // kube-state-metrics gives it; the generic type yields to it. + assert_eq!( + declarations[2].id_columns, + vec!["k8s.pod.uid", "k8s.container.name"] + ); + assert_eq!( + declarations[0].superseded_by_columns, + vec!["k8s.pod.uid", "k8s.container.name"], + "the generic container must yield to the k8s.container identity itself" + ); assert_eq!(declarations[1].id_columns, vec!["host.id"]); assert_eq!(declarations[1].descriptive_columns, vec!["host.name"]); - assert_eq!(declarations[3].id_columns, vec!["job"]); + assert_eq!(declarations[4].id_columns, vec!["job"]); assert_eq!( - declarations[3].descriptive_columns, + declarations[4].descriptive_columns, vec!["service.name", "service.namespace"] ); - assert_eq!(declarations[4].id_columns, vec!["job", "instance"]); - assert!(declarations[4].descriptive_columns.is_empty()); + assert_eq!(declarations[5].id_columns, vec!["job", "instance"]); + assert!(declarations[5].descriptive_columns.is_empty()); + + // Nothing here can produce a k8s.container, so the generic one must + // stand or the container disappears instead of changing type. + let no_k8s = prom_table_info( + "greptime_otel_resource_info", + &["job", "container.id", "k8s.pod.uid"], + OTEL_STAMPS, + ); + let declarations = sorted_declarations(&no_k8s); + assert_eq!(declarations[0].entity_type, "container"); + assert!(declarations[0].superseded_by_columns.is_empty()); + + // Same rule when a skipped explicit declaration blocks the implicit + // one: nothing declares the type, so nothing may yield to it. + let mut stamps = OTEL_STAMPS.to_vec(); + stamps.push(("greptime.semantic.entity.k8s.container.id", "gone")); + let broken_explicit = prom_table_info( + "greptime_otel_resource_info", + &["job", "container.id", "k8s.pod.uid", "k8s.container.name"], + &stamps, + ); + let declarations = sorted_declarations(&broken_explicit); + let types: Vec<&str> = declarations + .iter() + .map(|d| d.entity_type.as_str()) + .collect(); + assert_eq!(types, vec!["container", "k8s.pod", "service"]); + assert!(declarations[0].superseded_by_columns.is_empty()); let partial = prom_table_info( "greptime_otel_resource_info", diff --git a/src/operator/src/statement/semantic_graph.rs b/src/operator/src/statement/semantic_graph.rs index 86827cf8c7..80d4312e34 100644 --- a/src/operator/src/statement/semantic_graph.rs +++ b/src/operator/src/statement/semantic_graph.rs @@ -61,7 +61,7 @@ use datafusion::dataframe::DataFrame; use datafusion::functions::{core as core_fns, datetime as datetime_fns, string as string_fns}; use datafusion::functions_nested::expr_fn::make_array; use datafusion_common::{Column, Result as DfResult, ScalarValue}; -use datafusion_expr::{Case, Expr, LogicalPlan, ScalarUDF, cast, ident, lit}; +use datafusion_expr::{Case, Expr, LogicalPlan, ScalarUDF, cast, ident, lit, not}; pub use relationships::{ CallsSource, CoDeclaredSource, DeclaredSource, RelationshipSources, build_relationships_plan, }; @@ -292,6 +292,11 @@ pub struct EntityDeclaration { /// carrying the parts separately (a trace table's `service.namespace`) /// yields the same id as one carrying them pre-composed (`job`). pub id_qualifier: Option, + /// Identity columns of a more specific entity type that takes over on the + /// rows carrying all of them (a pod's container is the `k8s.container`, not + /// a generic `container`). Empty for explicit declarations: a user who + /// declares a type means it unconditionally. + pub superseded_by_columns: Vec, /// Descriptive columns snapshotted into the `descriptive` JSON (may be empty). pub descriptive_columns: Vec, /// Scope columns (namespace/environment). One column → scope verbatim; @@ -428,12 +433,31 @@ fn cast_string_or_empty(column: &str) -> Expr { /// unscheduled pod's `node`, an owner-less pod's `owner_*`), and an empty /// string is never a meaningful entity id. The qualifier is optional by /// construction and so is not guarded. -pub(crate) fn identifies(column: &str) -> Expr { +fn identifies(column: &str) -> Expr { ident(column) .is_not_null() .and(cast(ident(column), DataType::Utf8).not_eq(lit(""))) } +/// The row-level guard a declaration carries: every identity component present, +/// and the superseding type's identity not complete. Every branch that turns a +/// declaration into rows applies this, so the guard cannot drift between them. +pub(crate) fn declaration_predicate(declaration: &EntityDeclaration) -> Expr { + let mut predicate = lit(true); + for column in &declaration.id_columns { + predicate = predicate.and(identifies(column)); + } + if let Some(superseding) = declaration + .superseded_by_columns + .iter() + .map(|column| identifies(column)) + .reduce(Expr::and) + { + predicate = predicate.and(not(superseding)); + } + predicate +} + const ID_SEPARATOR: &str = ","; const ID_ESCAPE: &str = "\\"; /// Left unescaped: `/` is how Prometheus renders `job`. @@ -461,8 +485,8 @@ fn escaped_id_value(value: Expr) -> Expr { /// The identifying *column names* are deliberately absent: the same identity /// reaches us under different names per source — a trace table's /// `service_name` against a metric table's `job` — and encoding them would -/// split one entity into one per signal. `entity_id_attrs` keeps the -/// structured form. +/// split one entity into one per signal. `entity_id_attrs` carries the names +/// beside the id, where they document its origin without dividing it. /// /// `col` constructs the column reference (unqualified for registry branches, /// join-side-qualified for the calls derivation). @@ -652,9 +676,7 @@ fn registry_source( let mut rows = Vec::with_capacity(1 + rest.len()); for decl in std::iter::once(first).chain(rest) { // CAST even a single-column id: id columns need not be strings, and - // the computed table declares entity_id STRING. An id assembled from - // more than one column additionally carries its parts as a JSON - // object in entity_id_attrs, the structured form of the id string. + // the computed table declares entity_id STRING. let entity_id = entity_id_expr(&decl.id_columns, decl.id_qualifier.as_deref(), &|c| { ident(c) }); @@ -664,11 +686,9 @@ fn registry_source( .chain(&decl.id_columns) .cloned() .collect::>(); - let entity_id_attrs = if id_parts.len() == 1 { - null_json() - } else { - json_object_expr(&id_parts) - }; + // Carried for single-column ids too. Entity equality reads entity_id + // alone, so this is not part of the identity. + let entity_id_attrs = json_object_expr(&id_parts); let scope = match decl.scope_columns.as_slice() { [] => lit(""), @@ -694,10 +714,7 @@ fn registry_source( // Keep this predicate per declaration so an absent identity for one // entity does not remove other entities on the row. - let valid = decl - .id_columns - .iter() - .fold(lit(true), |predicate, id| predicate.and(identifies(id))); + let valid = declaration_predicate(decl); rows.push(vec![ valid, @@ -889,6 +906,7 @@ mod tests { entity_type: entity_type.to_string(), id_columns: id_columns.iter().map(|s| s.to_string()).collect(), id_qualifier: None, + superseded_by_columns: vec![], descriptive_columns: vec![], scope_columns: vec![], } @@ -999,8 +1017,11 @@ mod tests { let batch = &batches[0]; assert_eq!(strings(batch, 4), vec!["service"; batch.num_rows()]); assert_eq!(strings(batch, 5), vec!["cart"; batch.num_rows()]); - // Single-column id -> entity_id_attrs and descriptive are typed-JSON NULLs. - assert!(json_texts(batch, 6).iter().all(Option::is_none)); + // A single-column id still names its attribute. + assert_eq!( + json_texts(batch, 6), + vec![Some(r#"{"service_name":"cart"}"#.to_string()); batch.num_rows()] + ); assert!(json_texts(batch, 8).iter().all(Option::is_none)); assert_eq!( json_texts(batch, 9), @@ -1154,6 +1175,95 @@ mod tests { assert_eq!(ids, vec!["h2"]); } + #[tokio::test] + async fn registry_superseding_is_per_row_and_never_drops_an_entity() { + // One table holds pod rows and bare-runtime rows, so the rule is per + // row: each must end up with exactly one container node, the specific + // type where its identity is complete and the generic one elsewhere. + let schema = Arc::new(Schema::new(vec![ + Field::new( + "ts", + DataType::Timestamp(TimeUnit::Millisecond, None), + false, + ), + Field::new("container_id", DataType::Utf8, false), + Field::new("pod_uid", DataType::Utf8, true), + Field::new("container_name", DataType::Utf8, true), + ])); + let batch = RecordBatch::try_new( + schema.clone(), + vec![ + Arc::new(TimestampMillisecondArray::from(vec![ + 1_000, 2_000, 3_000, 4_000, + ])) as ArrayRef, + Arc::new(StringArray::from(vec![ + "c-pod", + "c-docker", + "c-empty", + "c-partial", + ])), + Arc::new(StringArray::from(vec![ + Some("uid-1"), + None, + Some(""), + Some("uid-2"), + ])), + Arc::new(StringArray::from(vec![ + Some("api"), + Some("api"), + Some("api"), + None, + ])), + ], + ) + .unwrap(); + let ctx = SessionContext::new(); + ctx.register_table( + "descriptors", + Arc::new(MemTable::try_new(schema, vec![vec![batch]]).unwrap()), + ) + .unwrap(); + + let mut generic = decl("container", &["container_id"]); + generic.table = "descriptors".to_string(); + generic.superseded_by_columns = vec!["pod_uid".to_string(), "container_name".to_string()]; + let mut specific = decl("k8s.container", &["pod_uid", "container_name"]); + specific.table = "descriptors".to_string(); + let plan = build_registry_plan( + vec![RegistrySource { + declarations: vec![generic, specific], + scan: ctx.table("descriptors").await.unwrap(), + }], + &test_window(), + ) + .unwrap() + .unwrap(); + + let mut rows: Vec<(String, String)> = collect(&ctx, plan) + .await + .iter() + .flat_map(|b| { + strings(b, 4) + .into_iter() + .zip(strings(b, 5)) + .collect::>() + }) + .collect(); + rows.sort(); + assert_eq!( + rows, + vec![ + // an empty uid is no uid: it supersedes nothing + ("container".to_string(), "c-docker".to_string()), + ("container".to_string(), "c-empty".to_string()), + // the pod row with no container name cannot produce the + // specific entity, so the generic one has to stand + ("container".to_string(), "c-partial".to_string()), + ("k8s.container".to_string(), "uid-1,api".to_string()), + ] + ); + } + #[tokio::test] async fn registry_expands_declarations_with_one_source_scan() { let ctx = metric_table_ctx(); diff --git a/src/operator/src/statement/semantic_graph/conventions.rs b/src/operator/src/statement/semantic_graph/conventions.rs index 9ec6f52e15..572140c206 100644 --- a/src/operator/src/statement/semantic_graph/conventions.rs +++ b/src/operator/src/statement/semantic_graph/conventions.rs @@ -57,6 +57,12 @@ pub struct ImplicitEntity { /// Skipped when the column is absent or empty. #[serde(default)] pub qualified_by: Option, + /// A more specific entity type declared next to this one, which takes over + /// on the rows carrying its full identity. Naming the type rather than a + /// trigger column is what keeps the entity replaceable but never + /// droppable: where the specific type is not derivable, this one stands. + #[serde(default)] + pub superseded_by: Option, /// Descriptive label columns, filtered to those present (kube-state-metrics /// label sets vary across versions). #[serde(default)] @@ -243,6 +249,27 @@ fn validate(conventions: &Conventions) -> Result<(), String> { implicit.entity )); } + if let Some(superseding) = &implicit.superseded_by { + let superseding = entities + .iter() + .find(|other| &other.entity == superseding) + .ok_or_else(|| { + format!( + "entity `{}` of info metric `{table}` is superseded by `{superseding}`, \ + which `{table}` does not declare", + implicit.entity + ) + })?; + // A chain would let the middle entity withdraw while its own + // replacement is absent. + if superseding.superseded_by.is_some() { + return Err(format!( + "entity `{}` of info metric `{table}` is superseded by `{}`, which is \ + itself superseded", + implicit.entity, superseding.entity + )); + } + } if implicit.qualified_by.as_ref().is_some_and(String::is_empty) { return Err(format!( "entity `{}` of info metric `{table}` has an empty qualified_by", @@ -317,6 +344,16 @@ mod tests { ) .contains("descriptive_rest") ); + // Superseding a type the same table does not declare leaves the rule + // dead and the duplicate node back in the graph. + assert!( + err( + "", + "t: [{entity: container, id: [x], superseded_by: k8s.container}]", + "" + ) + .contains("does not declare") + ); // the otel map runs through the same per-table validation assert!(err("", "", "t: [{entity: hosts, id: [x]}]").contains("unknown entity type")); // ids of a different arity for one type can never match each other diff --git a/src/operator/src/statement/semantic_graph/conventions.yaml b/src/operator/src/statement/semantic_graph/conventions.yaml index 808fe0f265..6a2da4de40 100644 --- a/src/operator/src/statement/semantic_graph/conventions.yaml +++ b/src/operator/src/statement/semantic_graph/conventions.yaml @@ -17,6 +17,8 @@ co_declared_edges: - { src: service.instance, dst: k8s.pod, rel: runs_on } - { src: service.instance, dst: container, rel: runs_on } - { src: container, dst: host, rel: runs_on } + - { src: service.instance, dst: k8s.container, rel: runs_on } + - { src: k8s.container, dst: host, rel: runs_on } # Applied only to trace sources; the endpoints are Greptime entity types # derived from GenAI semantic-convention attributes. @@ -51,8 +53,21 @@ otlp_trace_entities: id: [resource_attributes.host.id] descriptive: - resource_attributes.host.name + # A container inside a pod is the k8s.container entity, under the identity + # kube-state-metrics gives it, so the two sources name one node. The generic + # container yields to it only on rows that produce it, or a container would + # vanish instead of merging, and keeps its runtime id descriptive. + # + # Nothing checks a supersession against the edge vocabulary: whatever edges + # the superseded type takes part in, the superseding type needs its own. + - entity: k8s.container + id: [resource_attributes.k8s.pod.uid, resource_attributes.k8s.container.name] + descriptive: + - resource_attributes.container.id + - resource_attributes.container.name - entity: container id: [resource_attributes.container.id] + superseded_by: k8s.container descriptive: - resource_attributes.container.name - entity: k8s.pod @@ -152,8 +167,16 @@ otel_info_metrics: id: [host.id] descriptive: - host.name + - entity: k8s.container + id: [k8s.pod.uid, k8s.container.name] + descriptive: + - container.id + - container.name + # Yields to k8s.container, as on the trace side above. One descriptor table + # holds both pod rows and bare-runtime rows, so the test is per row. - entity: container id: [container.id] + superseded_by: k8s.container descriptive: - container.name - entity: k8s.pod diff --git a/src/operator/src/statement/semantic_graph/relationships.rs b/src/operator/src/statement/semantic_graph/relationships.rs index 257c844a9a..c9b194b110 100644 --- a/src/operator/src/statement/semantic_graph/relationships.rs +++ b/src/operator/src/statement/semantic_graph/relationships.rs @@ -19,18 +19,19 @@ //! expression helpers and the entity registry live in the parent module. use common_catalog::consts::{ - CONFIDENCE_COLUMN, DST_ID_COLUMN, DST_TYPE_COLUMN, DURATION_COUNT_COLUMN, DURATION_NANO_COLUMN, - DURATION_SUM_COLUMN, EDGE_ATTRIBUTES_COLUMN, ENTITY_SCOPE_COLUMN, ERROR_COUNT_COLUMN, - FRESH_UNTIL_COLUMN, GENERATION_ID_COLUMN, OBSERVED_AT_COLUMN, PARENT_SPAN_ID_COLUMN, - PROVENANCE_COLUMN, REL_TYPE_COLUMN, REQUEST_COUNT_COLUMN, SPAN_ID_COLUMN, SPAN_KIND_CLIENT, - SPAN_KIND_COLUMN, SPAN_KIND_SERVER, SPAN_STATUS_CODE_COLUMN, SPAN_STATUS_ERROR, SRC_ID_COLUMN, - SRC_TYPE_COLUMN, TRACE_ID_COLUMN, TRACE_TIMESTAMP_COLUMN, VALID_FROM_COLUMN, - VALID_UNTIL_COLUMN, WINDOW_END_COLUMN, WINDOW_START_COLUMN, + CONFIDENCE_COLUMN, DST_ID_COLUMN, DST_TYPE_COLUMN, DURATION_COUNT_COLUMN, DURATION_MAX_COLUMN, + DURATION_NANO_COLUMN, DURATION_SUM_COLUMN, EDGE_ATTRIBUTES_COLUMN, ENTITY_SCOPE_COLUMN, + ERROR_COUNT_COLUMN, FRESH_UNTIL_COLUMN, GENERATION_ID_COLUMN, OBSERVED_AT_COLUMN, + PARENT_SPAN_ID_COLUMN, PROVENANCE_COLUMN, REL_TYPE_COLUMN, REQUEST_COUNT_COLUMN, + SPAN_ID_COLUMN, SPAN_KIND_CLIENT, SPAN_KIND_COLUMN, SPAN_KIND_SERVER, SPAN_STATUS_CODE_COLUMN, + SPAN_STATUS_ERROR, SRC_ID_COLUMN, SRC_TYPE_COLUMN, TRACE_ID_COLUMN, TRACE_TIMESTAMP_COLUMN, + UNMATCHED_COUNT_COLUMN, VALID_FROM_COLUMN, VALID_UNTIL_COLUMN, WINDOW_END_COLUMN, + WINDOW_START_COLUMN, }; use datafusion::arrow::datatypes::DataType; use datafusion::dataframe::DataFrame; use datafusion::functions::core as core_fns; -use datafusion::functions_aggregate::expr_fn::{bool_or, count, min, sum}; +use datafusion::functions_aggregate::expr_fn::{bool_or, count, max, min, sum}; use datafusion::functions_window::expr_fn::row_number; use datafusion_common::{Result as DfResult, ScalarValue}; use datafusion_expr::{Expr, ExprFunctionExt, JoinType, LogicalPlan, cast, ident, lit, when}; @@ -41,8 +42,8 @@ use crate::statement::semantic_graph::conventions::{ }; use crate::statement::semantic_graph::{ DECLARED_EDGE_IDENTITY_COLUMNS, EntityDeclaration, GraphQueryWindow, bin_interval, bin_ms, - conventions, entity_id_expr, identifies, interval, null_json, parse_json_expr, qcol, union_all, - unnest_rows, + conventions, declaration_predicate, entity_id_expr, interval, null_json, parse_json_expr, qcol, + union_all, unnest_rows, }; /// The embedded conventions, with a broken file surfaced as a plan error. @@ -56,7 +57,7 @@ fn builtin() -> DfResult<&'static Conventions> { /// them over the union to enforce the contract. (The physical declared table /// additionally stores `valid_from`/`valid_until`, which feed the validity /// filter and the projected window columns.) -const RELATIONSHIP_COLUMNS: [&str; 16] = [ +const RELATIONSHIP_COLUMNS: [&str; 18] = [ OBSERVED_AT_COLUMN, WINDOW_START_COLUMN, WINDOW_END_COLUMN, @@ -69,9 +70,11 @@ const RELATIONSHIP_COLUMNS: [&str; 16] = [ PROVENANCE_COLUMN, CONFIDENCE_COLUMN, REQUEST_COUNT_COLUMN, + UNMATCHED_COUNT_COLUMN, ERROR_COUNT_COLUMN, DURATION_SUM_COLUMN, DURATION_COUNT_COLUMN, + DURATION_MAX_COLUMN, EDGE_ATTRIBUTES_COLUMN, ]; @@ -117,7 +120,7 @@ pub struct RelationshipSources { /// Builds the `semantic_relationships` plan: the service-calls, agent-calls, /// co-declared, and declared-edge branches unioned and re-projected to the -/// 16-column contract. Returns `None` when no source can contribute edges, so +/// 18-column contract. Returns `None` when no source can contribute edges, so /// the computed table streams empty. pub fn build_relationships_plan( sources: RelationshipSources, @@ -188,11 +191,7 @@ fn co_declared_branch( .map(|(src, dst, rel_type, provenance)| { // Both endpoints must identify something on the row for the // row to witness the edge. - let valid = src - .id_columns - .iter() - .chain(&dst.id_columns) - .fold(lit(true), |predicate, id| predicate.and(identifies(id))); + let valid = declaration_predicate(src).and(declaration_predicate(dst)); vec![ valid, bin.clone(), @@ -208,8 +207,10 @@ fn co_declared_branch( lit(1.0_f64), null_i64(), null_i64(), + null_i64(), lit(ScalarValue::Float64(None)), null_i64(), + lit(ScalarValue::Float64(None)), null_json(), ] }) @@ -300,9 +301,13 @@ fn declared_edges(scan: DataFrame, window: &GraphQueryWindow) -> DfResult DfResult> { let mut clients: Option = None; let mut servers: Option = None; @@ -406,10 +412,14 @@ fn calls_branch(traces: &[CallsSource], window: &GraphQueryWindow) -> DfResult DfResult DfResult DfResult DfResult { /// side's timestamp and cannot prune this side's scan on their own). fn span_predicate(service: &EntityDeclaration, window: &GraphQueryWindow, strict: bool) -> Expr { let ts = ident(TRACE_TIMESTAMP_COLUMN); - let mut predicate = if strict { + let window_predicate = if strict { ts.clone() .gt_eq(window.source_start()) .and(ts.lt(window.source_end())) @@ -494,10 +524,7 @@ fn span_predicate(service: &EntityDeclaration, window: &GraphQueryWindow, strict .and(ts.lt(window.source_end() + interval(CHILD_SPAN_LATE_NANOS))) }; // An absent identity component identifies nothing, on either endpoint. - for id in &service.id_columns { - predicate = predicate.and(identifies(id)); - } - predicate + window_predicate.and(declaration_predicate(service)) } /// One trace table's client spans, normalized to @@ -685,6 +712,7 @@ fn agent_calls_branch( .build()? .alias(ERROR_COUNT_COLUMN), sum(ident(DURATION_NANO_COLUMN)).alias("duration_nano_sum"), + max(ident(DURATION_NANO_COLUMN)).alias("duration_nano_max"), ], )? .select(vec![ @@ -700,10 +728,15 @@ fn agent_calls_branch( lit(PROVENANCE_TRACE).alias(PROVENANCE_COLUMN), lit(1.0_f64).alias(CONFIDENCE_COLUMN), ident(REQUEST_COUNT_COLUMN), + // The join is inner: an unanswered delegation leaves no row, so + // there is no unmatched population here. + lit(ScalarValue::Int64(None)).alias(UNMATCHED_COUNT_COLUMN), ident(ERROR_COUNT_COLUMN), (cast(ident("duration_nano_sum"), DataType::Float64) / lit(1e9_f64)) .alias(DURATION_SUM_COLUMN), ident(REQUEST_COUNT_COLUMN).alias(DURATION_COUNT_COLUMN), + (cast(ident("duration_nano_max"), DataType::Float64) / lit(1e9_f64)) + .alias(DURATION_MAX_COLUMN), null_json().alias(EDGE_ATTRIBUTES_COLUMN), ])?; Ok(Some(df)) @@ -878,6 +911,7 @@ mod tests { entity_type: "service".to_string(), id_columns: id_columns.iter().map(|s| s.to_string()).collect(), id_qualifier: None, + superseded_by_columns: vec![], descriptive_columns: vec![], scope_columns: vec![], } @@ -917,32 +951,11 @@ mod tests { assert_eq!(strings(batch, 8), vec!["calls"]); assert_eq!(strings(batch, 9), vec!["trace"]); - let request_count = batch - .column(11) - .as_any() - .downcast_ref::() - .unwrap(); - assert_eq!(request_count.value(0), 2); - let error_count = batch - .column(12) - .as_any() - .downcast_ref::() - .unwrap(); - assert_eq!(error_count.value(0), 1); - let duration_sum = batch - .column(13) - .as_any() - .downcast_ref::() - .unwrap(); - assert!((duration_sum.value(0) - 2.0).abs() < 1e-9); - let duration_count = batch - .column(14) - .as_any() - .downcast_ref::() - .unwrap(); - assert_eq!(duration_count.value(0), 2); + assert_eq!(red_metrics(batch, 0), (2, 1, 2.0, 2)); + // The slower of the two paired server spans. + assert_eq!(duration_max(batch, 0), 1.5); // Derived calls edges carry no attributes: a typed-JSON NULL. - assert!(json_texts(batch, 15)[0].is_none()); + assert!(json_texts(batch, 17)[0].is_none()); } /// RED columns of one output row: `(request_count, error_count, @@ -957,12 +970,30 @@ mod tests { .value(row) }; let duration_sum = batch - .column(13) + .column(14) .as_any() .downcast_ref::() .unwrap() .value(row); - (i64_at(11), i64_at(12), duration_sum, i64_at(14)) + (i64_at(11), i64_at(13), duration_sum, i64_at(15)) + } + + fn unmatched_count(batch: &RecordBatch, row: usize) -> i64 { + batch + .column(12) + .as_any() + .downcast_ref::() + .unwrap() + .value(row) + } + + fn duration_max(batch: &RecordBatch, row: usize) -> f64 { + batch + .column(16) + .as_any() + .downcast_ref::() + .unwrap() + .value(row) } fn confidence(batch: &RecordBatch, row: usize) -> f64 { @@ -1124,8 +1155,11 @@ mod tests { assert_eq!(strings(batch, 9), vec!["trace"]); assert_eq!(confidence(batch, 0), VIRTUAL_NODE_CONFIDENCE); assert_eq!(red_metrics(batch, 0), (1, 1, 0.25, 1)); + // With no pair to prefer, both columns come from the client spans. + assert_eq!(duration_max(batch, 0), 0.25); + assert_eq!(unmatched_count(batch, 0), 1); assert_eq!( - json_texts(batch, 15), + json_texts(batch, 17), vec![Some(r#"{"connection_type":"virtual_node"}"#.to_string())] ); } @@ -1153,7 +1187,7 @@ mod tests { let batch = &batches[0]; assert_eq!(strings(batch, 7), vec!["mysql"]); assert_eq!( - json_texts(batch, 15), + json_texts(batch, 17), vec![Some(r#"{"connection_type":"database"}"#.to_string())] ); } @@ -1231,6 +1265,9 @@ mod tests { assert_eq!(strings(batch, 9), vec!["trace"]); assert_eq!(confidence(batch, 0), 1.0); assert_eq!(red_metrics(batch, 0), (2, 1, 3.0, 2)); + // The child spans carry the durations, so the max is real here. + assert_eq!(duration_max(batch, 0), 2.0); + assert!(batch.column(12).is_null(0)); } #[tokio::test] @@ -1331,6 +1368,7 @@ mod tests { entity_type: entity_type.to_string(), id_columns: id_columns.iter().map(|s| s.to_string()).collect(), id_qualifier: None, + superseded_by_columns: vec![], descriptive_columns: vec![], scope_columns: vec![], } @@ -1377,7 +1415,7 @@ mod tests { )); assert_eq!(confidence(batch, i), 1.0); // Co-declared edges carry no RED metrics or attributes. - for column in 11..=15 { + for column in 11..=17 { assert!(batch.column(column).is_null(i)); } } @@ -1527,8 +1565,17 @@ mod tests { )], &[ // A sampled-out server: the client names the same peer an - // actual pair witnesses in the same window. - (1_000, "t1", "c1", None, CLIENT, ERROR, "frontend", 999), + // actual pair witnesses in the same window, and outlasts it. + ( + 1_000, + "t1", + "c1", + None, + CLIENT, + ERROR, + "frontend", + 9_000_000_000, + ), ( 2_010, "t2", @@ -1556,7 +1603,11 @@ mod tests { assert_eq!(strings(batch, 7), vec!["cart"]); assert_eq!(confidence(batch, 0), 1.0); assert_eq!(red_metrics(batch, 0), (1, 0, 0.5, 1)); - assert!(json_texts(batch, 15)[0].is_none()); + // The longer, client-timed span is excluded from the max but still + // counted. + assert_eq!(duration_max(batch, 0), 0.5); + assert_eq!(unmatched_count(batch, 0), 1); + assert!(json_texts(batch, 17)[0].is_none()); } #[tokio::test] @@ -1785,7 +1836,7 @@ mod tests { let fresh_until = ts_values(batch, 3); let src = strings(batch, 5); let provenance = strings(batch, 9); - let attributes = json_texts(batch, 15); + let attributes = json_texts(batch, 17); for i in 0..batch.num_rows() { rows.push(( src[i].clone(), diff --git a/src/servers/src/otlp/metrics/resource_info.rs b/src/servers/src/otlp/metrics/resource_info.rs index cd3427f716..2100ce7313 100644 --- a/src/servers/src/otlp/metrics/resource_info.rs +++ b/src/servers/src/otlp/metrics/resource_info.rs @@ -36,8 +36,9 @@ use crate::otlp::metrics::{ exponential_histogram_value, scalar_value_string, service_identity, }; use crate::otlp::trace::{ - KEY_CONTAINER_ID, KEY_CONTAINER_NAME, KEY_HOST_ID, KEY_HOST_NAME, KEY_K8S_NAMESPACE_NAME, - KEY_K8S_POD_NAME, KEY_K8S_POD_UID, KEY_SERVICE_NAME, KEY_SERVICE_NAMESPACE, + KEY_CONTAINER_ID, KEY_CONTAINER_NAME, KEY_HOST_ID, KEY_HOST_NAME, KEY_K8S_CONTAINER_NAME, + KEY_K8S_NAMESPACE_NAME, KEY_K8S_POD_NAME, KEY_K8S_POD_UID, KEY_SERVICE_NAME, + KEY_SERVICE_NAMESPACE, }; use crate::row_writer::{self, MultiTableData}; @@ -61,13 +62,14 @@ fn is_projected_attr(key: &str) -> bool { | KEY_CONTAINER_NAME | KEY_K8S_POD_UID | KEY_K8S_POD_NAME + | KEY_K8S_CONTAINER_NAME | KEY_K8S_NAMESPACE_NAME ) } /// Upper bound of [`is_projected_attr`] plus the derived `job`/`instance`, /// used to size the per-row buffers. -const MAX_PROJECTED_TAGS: usize = 11; +const MAX_PROJECTED_TAGS: usize = 12; /// Projected attributes (sorted `(name, value)` pairs) -> graph window -> /// the newest data-point time seen in that window, which is what the row for diff --git a/src/servers/src/otlp/trace.rs b/src/servers/src/otlp/trace.rs index 9ee16507bd..d2e9b1131c 100644 --- a/src/servers/src/otlp/trace.rs +++ b/src/servers/src/otlp/trace.rs @@ -54,6 +54,7 @@ pub const KEY_CONTAINER_ID: &str = "container.id"; pub const KEY_CONTAINER_NAME: &str = "container.name"; pub const KEY_K8S_POD_UID: &str = "k8s.pod.uid"; pub const KEY_K8S_POD_NAME: &str = "k8s.pod.name"; +pub const KEY_K8S_CONTAINER_NAME: &str = "k8s.container.name"; pub const KEY_K8S_NAMESPACE_NAME: &str = "k8s.namespace.name"; pub const KEY_SPAN_KIND: &str = "span.kind"; diff --git a/tests/cases/standalone/common/information_schema/table_semantics.result b/tests/cases/standalone/common/information_schema/table_semantics.result index b57087b3f5..ee66ea3019 100644 --- a/tests/cases/standalone/common/information_schema/table_semantics.result +++ b/tests/cases/standalone/common/information_schema/table_semantics.result @@ -1,19 +1,20 @@ DESC TABLE information_schema.table_semantics; -+------------------+--------+-----+------+---------+---------------+ -| Column | Type | Key | Null | Default | Semantic Type | -+------------------+--------+-----+------+---------+---------------+ -| table_catalog | String | | NO | | FIELD | -| table_schema | String | | NO | | FIELD | -| table_name | String | | NO | | FIELD | -| table_id | UInt32 | | NO | | FIELD | -| signal_type | String | | YES | | FIELD | -| source | String | | YES | | FIELD | -| source_version | String | | YES | | FIELD | -| pipeline | String | | YES | | FIELD | -| metadata_quality | String | | YES | | FIELD | -| semantic_options | String | | YES | | FIELD | -+------------------+--------+-----+------+---------+---------------+ ++---------------------+--------+-----+------+---------+---------------+ +| Column | Type | Key | Null | Default | Semantic Type | ++---------------------+--------+-----+------+---------+---------------+ +| table_catalog | String | | NO | | FIELD | +| table_schema | String | | NO | | FIELD | +| table_name | String | | NO | | FIELD | +| table_id | UInt32 | | NO | | FIELD | +| signal_type | String | | YES | | FIELD | +| source | String | | YES | | FIELD | +| source_version | String | | YES | | FIELD | +| pipeline | String | | YES | | FIELD | +| metadata_quality | String | | YES | | FIELD | +| semantic_options | String | | YES | | FIELD | +| entity_declarations | String | | YES | | FIELD | ++---------------------+--------+-----+------+---------+---------------+ CREATE TABLE metrics_tagged ( ts TIMESTAMP TIME INDEX, @@ -204,3 +205,148 @@ DROP TABLE phy_sem; Affected Rows: 0 +-- entity_declarations reports what a table actually contributes to the graph, +-- including the declarations the built-in conventions derive. A whitelisted +-- OTel descriptor table declares its entities without any entity option, and +-- the generic container records the identity it yields to. +CREATE TABLE greptime_otel_resource_info ( + greptime_timestamp TIMESTAMP TIME INDEX, + greptime_value DOUBLE, + "job" STRING, + "instance" STRING, + "service.name" STRING, + "host.id" STRING, + "container.id" STRING, + "container.name" STRING, + "k8s.pod.uid" STRING, + "k8s.pod.name" STRING, + "k8s.container.name" STRING, + PRIMARY KEY ("job", "instance", "service.name", "host.id", "container.id", "container.name", "k8s.pod.uid", "k8s.pod.name", "k8s.container.name") +) +WITH ( + 'greptime.semantic.signal_type' = 'metric', + 'greptime.semantic.source' = 'opentelemetry', + 'greptime.semantic.metric.type' = 'info' +); + +Affected Rows: 0 + +SELECT table_name, entity_declarations +FROM information_schema.table_semantics +WHERE table_name = 'greptime_otel_resource_info'; + ++-----------------------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ +| table_name | entity_declarations | ++-----------------------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ +| greptime_otel_resource_info | [{"entity_type":"container","origin":"convention","id":["container.id"],"superseded_by":["k8s.pod.uid","k8s.container.name"],"descriptive":["container.name"]},{"entity_type":"host","origin":"convention","id":["host.id"]},{"entity_type":"k8s.container","origin":"convention","id":["k8s.pod.uid","k8s.container.name"],"descriptive":["container.id","container.name"]},{"entity_type":"k8s.pod","origin":"convention","id":["k8s.pod.uid"],"descriptive":["k8s.pod.name"]},{"entity_type":"service","origin":"convention","id":["job"],"descriptive":["service.name"]},{"entity_type":"service.instance","origin":"convention","id":["job","instance"]}] | ++-----------------------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ + +-- The same table under the wrong source stamp derives nothing: the column is +-- empty and `source` names the reason. +CREATE TABLE kube_pod_info ( + greptime_timestamp TIMESTAMP TIME INDEX, + greptime_value DOUBLE, + "namespace" STRING, + "pod" STRING, + "uid" STRING, + "node" STRING, + PRIMARY KEY ("namespace", "pod", "uid", "node") +) +WITH ( + 'greptime.semantic.signal_type' = 'metric', + 'greptime.semantic.source' = 'influxdb' +); + +Affected Rows: 0 + +SELECT table_name, source, entity_declarations +FROM information_schema.table_semantics +WHERE table_name = 'kube_pod_info'; + ++---------------+----------+---------------------+ +| table_name | source | entity_declarations | ++---------------+----------+---------------------+ +| kube_pod_info | influxdb | | ++---------------+----------+---------------------+ + +ALTER TABLE kube_pod_info SET 'greptime.semantic.source' = 'prometheus'; + +Affected Rows: 0 + +SELECT table_name, source, entity_declarations +FROM information_schema.table_semantics +WHERE table_name = 'kube_pod_info'; + ++---------------+------------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ +| table_name | source | entity_declarations | ++---------------+------------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ +| kube_pod_info | prometheus | [{"entity_type":"k8s.node","origin":"convention","id":["node"]},{"entity_type":"k8s.pod","origin":"convention","id":["uid"],"descriptive":["namespace","pod","node"]}] | ++---------------+------------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ + +DROP TABLE greptime_otel_resource_info; + +Affected Rows: 0 + +DROP TABLE kube_pod_info; + +Affected Rows: 0 + +-- A missing id column drops that entity alone: without `uid` there is no +-- k8s.pod, and the k8s.node next to it still lands. +CREATE TABLE kube_pod_info ( + greptime_timestamp TIMESTAMP TIME INDEX, + greptime_value DOUBLE, + "namespace" STRING, + "pod" STRING, + "node" STRING, + PRIMARY KEY ("namespace", "pod", "node") +) +WITH ( + 'greptime.semantic.signal_type' = 'metric', + 'greptime.semantic.source' = 'prometheus' +); + +Affected Rows: 0 + +SELECT table_name, entity_declarations +FROM information_schema.table_semantics +WHERE table_name = 'kube_pod_info'; + ++---------------+------------------------------------------------------------------+ +| table_name | entity_declarations | ++---------------+------------------------------------------------------------------+ +| kube_pod_info | [{"entity_type":"k8s.node","origin":"convention","id":["node"]}] | ++---------------+------------------------------------------------------------------+ + +DROP TABLE kube_pod_info; + +Affected Rows: 0 + +-- A trace table carries no semantic option at all, yet the conventions derive +-- its entities: it must still be visible here, or the view cannot answer why +-- it is in the graph. +CREATE TABLE traces_untagged ( + "timestamp" TIMESTAMP(9) TIME INDEX, + trace_id STRING, + span_id STRING, + service_name STRING, + "resource_attributes.host.id" STRING, + PRIMARY KEY (service_name) +) WITH ('table_data_model' = 'greptime_trace_v1', 'append_mode' = 'true'); + +Affected Rows: 0 + +SELECT table_name, signal_type, semantic_options, entity_declarations +FROM information_schema.table_semantics +WHERE table_name = 'traces_untagged'; + ++-----------------+-------------+------------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------+ +| table_name | signal_type | semantic_options | entity_declarations | ++-----------------+-------------+------------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------+ +| traces_untagged | | | [{"entity_type":"host","origin":"convention","id":["resource_attributes.host.id"]},{"entity_type":"service","origin":"convention","id":["service_name"]}] | ++-----------------+-------------+------------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------+ + +DROP TABLE traces_untagged; + +Affected Rows: 0 + diff --git a/tests/cases/standalone/common/information_schema/table_semantics.sql b/tests/cases/standalone/common/information_schema/table_semantics.sql index 4acb15294d..577c1edb69 100644 --- a/tests/cases/standalone/common/information_schema/table_semantics.sql +++ b/tests/cases/standalone/common/information_schema/table_semantics.sql @@ -107,3 +107,100 @@ WHERE table_name = 'logical_sem'; DROP TABLE logical_sem; DROP TABLE phy_sem; + +-- entity_declarations reports what a table actually contributes to the graph, +-- including the declarations the built-in conventions derive. A whitelisted +-- OTel descriptor table declares its entities without any entity option, and +-- the generic container records the identity it yields to. +CREATE TABLE greptime_otel_resource_info ( + greptime_timestamp TIMESTAMP TIME INDEX, + greptime_value DOUBLE, + "job" STRING, + "instance" STRING, + "service.name" STRING, + "host.id" STRING, + "container.id" STRING, + "container.name" STRING, + "k8s.pod.uid" STRING, + "k8s.pod.name" STRING, + "k8s.container.name" STRING, + PRIMARY KEY ("job", "instance", "service.name", "host.id", "container.id", "container.name", "k8s.pod.uid", "k8s.pod.name", "k8s.container.name") +) +WITH ( + 'greptime.semantic.signal_type' = 'metric', + 'greptime.semantic.source' = 'opentelemetry', + 'greptime.semantic.metric.type' = 'info' +); + +SELECT table_name, entity_declarations +FROM information_schema.table_semantics +WHERE table_name = 'greptime_otel_resource_info'; + +-- The same table under the wrong source stamp derives nothing: the column is +-- empty and `source` names the reason. +CREATE TABLE kube_pod_info ( + greptime_timestamp TIMESTAMP TIME INDEX, + greptime_value DOUBLE, + "namespace" STRING, + "pod" STRING, + "uid" STRING, + "node" STRING, + PRIMARY KEY ("namespace", "pod", "uid", "node") +) +WITH ( + 'greptime.semantic.signal_type' = 'metric', + 'greptime.semantic.source' = 'influxdb' +); + +SELECT table_name, source, entity_declarations +FROM information_schema.table_semantics +WHERE table_name = 'kube_pod_info'; + +ALTER TABLE kube_pod_info SET 'greptime.semantic.source' = 'prometheus'; + +SELECT table_name, source, entity_declarations +FROM information_schema.table_semantics +WHERE table_name = 'kube_pod_info'; + +DROP TABLE greptime_otel_resource_info; + +DROP TABLE kube_pod_info; + +-- A missing id column drops that entity alone: without `uid` there is no +-- k8s.pod, and the k8s.node next to it still lands. +CREATE TABLE kube_pod_info ( + greptime_timestamp TIMESTAMP TIME INDEX, + greptime_value DOUBLE, + "namespace" STRING, + "pod" STRING, + "node" STRING, + PRIMARY KEY ("namespace", "pod", "node") +) +WITH ( + 'greptime.semantic.signal_type' = 'metric', + 'greptime.semantic.source' = 'prometheus' +); + +SELECT table_name, entity_declarations +FROM information_schema.table_semantics +WHERE table_name = 'kube_pod_info'; + +DROP TABLE kube_pod_info; + +-- A trace table carries no semantic option at all, yet the conventions derive +-- its entities: it must still be visible here, or the view cannot answer why +-- it is in the graph. +CREATE TABLE traces_untagged ( + "timestamp" TIMESTAMP(9) TIME INDEX, + trace_id STRING, + span_id STRING, + service_name STRING, + "resource_attributes.host.id" STRING, + PRIMARY KEY (service_name) +) WITH ('table_data_model' = 'greptime_trace_v1', 'append_mode' = 'true'); + +SELECT table_name, signal_type, semantic_options, entity_declarations +FROM information_schema.table_semantics +WHERE table_name = 'traces_untagged'; + +DROP TABLE traces_untagged; diff --git a/tests/cases/standalone/common/system/information_schema.result b/tests/cases/standalone/common/system/information_schema.result index 17e1f0ef80..2546bb34d5 100644 --- a/tests/cases/standalone/common/system/information_schema.result +++ b/tests/cases/standalone/common/system/information_schema.result @@ -491,6 +491,7 @@ order by table_schema, table_name, column_name; | greptime | information_schema | table_privileges | table_catalog | 2 | 2147483647 | 2147483647 | | | | utf8 | utf8_bin | | | select,insert | | String | string | FIELD | | NO | string | | | | greptime | information_schema | table_privileges | table_name | 4 | 2147483647 | 2147483647 | | | | utf8 | utf8_bin | | | select,insert | | String | string | FIELD | | NO | string | | | | greptime | information_schema | table_privileges | table_schema | 3 | 2147483647 | 2147483647 | | | | utf8 | utf8_bin | | | select,insert | | String | string | FIELD | | NO | string | | | +| greptime | information_schema | table_semantics | entity_declarations | 11 | 2147483647 | 2147483647 | | | | utf8 | utf8_bin | | | select,insert | | String | string | FIELD | | YES | string | | | | greptime | information_schema | table_semantics | metadata_quality | 9 | 2147483647 | 2147483647 | | | | utf8 | utf8_bin | | | select,insert | | String | string | FIELD | | YES | string | | | | greptime | information_schema | table_semantics | pipeline | 8 | 2147483647 | 2147483647 | | | | utf8 | utf8_bin | | | select,insert | | String | string | FIELD | | YES | string | | | | greptime | information_schema | table_semantics | semantic_options | 10 | 2147483647 | 2147483647 | | | | utf8 | utf8_bin | | | select,insert | | String | string | FIELD | | YES | string | | | diff --git a/tests/cases/standalone/common/system/semantic_graph.result b/tests/cases/standalone/common/system/semantic_graph.result index f889431984..e1728c4ae0 100644 --- a/tests/cases/standalone/common/system/semantic_graph.result +++ b/tests/cases/standalone/common/system/semantic_graph.result @@ -97,10 +97,10 @@ order by entity_type, entity_id; +------------------+-----------+-------------------------------------+---------+-------------------+------------------------------+ | entity_type | entity_id | entity_id_attrs | scope | descriptive | source_tables | +------------------+-----------+-------------------------------------+---------+-------------------+------------------------------+ -| host | h1 | | | | ["public.graph_app_metrics"] | +| host | h1 | {"host":"h1"} | | | ["public.graph_app_metrics"] | | process | cart,h1 | {"host":"h1","service_name":"cart"} | | {"env":"us-east"} | ["public.graph_app_metrics"] | -| service | cart | | us-east | | ["public.graph_app_metrics"] | -| service.instance | cart-0 | | | | ["public.graph_app_metrics"] | +| service | cart | {"service_name":"cart"} | us-east | | ["public.graph_app_metrics"] | +| service.instance | cart-0 | {"instance":"cart-0"} | | | ["public.graph_app_metrics"] | +------------------+-----------+-------------------------------------+---------+-------------------+------------------------------+ select src_type, src_id, dst_type, dst_id, rel_type, provenance, confidence @@ -241,29 +241,30 @@ Affected Rows: 1 -- SQLNESS PROTOCOL MYSQL select src_id, dst_id, rel_type, provenance, confidence, - request_count, error_count, duration_sum, duration_count, attributes + request_count, unmatched_count, error_count, + duration_sum, duration_count, duration_max, attributes from greptime_private.semantic_relationships order by dst_id; -+----------+-----------+----------+------------+------------+---------------+-------------+--------------+----------------+------------------------------------+ -| src_id | dst_id | rel_type | provenance | confidence | request_count | error_count | duration_sum | duration_count | attributes | -+----------+-----------+----------+------------+------------+---------------+-------------+--------------+----------------+------------------------------------+ -| frontend | cart | calls | trace | 1 | 2 | 1 | 2 | 2 | | -| frontend | orders-db | calls | trace | 0.5 | 1 | 0 | 0.1 | 1 | {"connection_type":"database"} | -| frontend | redis | calls | trace | 0.5 | 1 | 0 | 0.25 | 1 | {"connection_type":"virtual_node"} | -+----------+-----------+----------+------------+------------+---------------+-------------+--------------+----------------+------------------------------------+ ++----------+-----------+----------+------------+------------+---------------+-----------------+-------------+--------------+----------------+--------------+------------------------------------+ +| src_id | dst_id | rel_type | provenance | confidence | request_count | unmatched_count | error_count | duration_sum | duration_count | duration_max | attributes | ++----------+-----------+----------+------------+------------+---------------+-----------------+-------------+--------------+----------------+--------------+------------------------------------+ +| frontend | cart | calls | trace | 1 | 2 | 0 | 1 | 2 | 2 | 1.5 | | +| frontend | orders-db | calls | trace | 0.5 | 1 | 1 | 0 | 0.1 | 1 | 0.1 | {"connection_type":"database"} | +| frontend | redis | calls | trace | 0.5 | 1 | 1 | 0 | 0.25 | 1 | 0.25 | {"connection_type":"virtual_node"} | ++----------+-----------+----------+------------+------------+---------------+-----------------+-------------+--------------+----------------+--------------+------------------------------------+ -- SQLNESS PROTOCOL MYSQL select entity_type, entity_id, entity_id_attrs, scope, source_tables from greptime_private.semantic_entities order by entity_id; -+-------------+-----------+-----------------+-------+---------------------------+ -| entity_type | entity_id | entity_id_attrs | scope | source_tables | -+-------------+-----------+-----------------+-------+---------------------------+ -| service | cart | | | ["public.graph_traces_b"] | -| service | frontend | | | ["public.graph_traces_a"] | -+-------------+-----------+-----------------+-------+---------------------------+ ++-------------+-----------+-----------------------------+-------+---------------------------+ +| entity_type | entity_id | entity_id_attrs | scope | source_tables | ++-------------+-----------+-----------------------------+-------+---------------------------+ +| service | cart | {"service_name":"cart"} | | ["public.graph_traces_b"] | +| service | frontend | {"service_name":"frontend"} | | ["public.graph_traces_a"] | ++-------------+-----------+-----------------------------+-------+---------------------------+ drop table graph_traces_a; @@ -633,7 +634,10 @@ Affected Rows: 0 -- the stable ids only, so rows with an empty host.id / container.id link no -- infrastructure, and a row without an instance value declares no -- service.instance. The same table name stamped source=prometheus is not --- whitelisted. +-- whitelisted. Container rows show the superseding rule: a pod's container is +-- the k8s.container entity where the row names it, and stays a generic +-- container where it does not — one node per container either way, with no +-- kube-state-metrics deployed. create table greptime_otel_resource_info ( greptime_timestamp timestamp(3) time index, job string, @@ -644,9 +648,13 @@ create table greptime_otel_resource_info ( "host.name" string, "container.id" string, "container.name" string, + "k8s.pod.uid" string, + "k8s.pod.name" string, + "k8s.container.name" string, greptime_value double, primary key (job, instance, "service.name", "service.namespace", - "host.id", "host.name", "container.id", "container.name") + "host.id", "host.name", "container.id", "container.name", + "k8s.pod.uid", "k8s.pod.name", "k8s.container.name") ) with ( 'greptime.semantic.signal_type' = 'metric', 'greptime.semantic.source' = 'opentelemetry', @@ -656,11 +664,13 @@ create table greptime_otel_resource_info ( Affected Rows: 0 insert into greptime_otel_resource_info values - (now(), 'shop/api', 'inst-1', 'api', 'shop', 'h-1', 'node-a', 'c-1', 'api-ctr', 1), - (now(), 'shop/api', 'inst-2', 'api', 'shop', '', 'laptop', '', '', 1), - (now(), 'worker', '', 'worker', '', 'h-1', 'node-a', '', '', 1); + (now(), 'shop/api', 'inst-1', 'api', 'shop', 'h-1', 'node-a', 'c-1', 'api-ctr', '', '', '', 1), + (now(), 'shop/api', 'inst-2', 'api', 'shop', '', 'laptop', '', '', '', '', '', 1), + (now(), 'worker', '', 'worker', '', 'h-1', 'node-a', '', '', '', '', '', 1), + (now(), 'shop/api', 'inst-3', 'api', 'shop', 'h-2', 'node-b', 'c-2', 'api-ctr', 'uid-1', 'api-pod', 'api', 1), + (now(), 'shop/api', 'inst-4', 'api', 'shop', 'h-2', 'node-b', 'c-3', 'api-ctr', 'uid-2', 'api-pod-2', '', 1); -Affected Rows: 3 +Affected Rows: 5 -- SQLNESS PROTOCOL MYSQL select entity_type, entity_id, source_tables @@ -671,59 +681,58 @@ order by entity_type, entity_id, source_tables; | entity_type | entity_id | source_tables | +------------------+-----------------+----------------------------------------+ | container | c-1 | ["public.greptime_otel_resource_info"] | +| container | c-3 | ["public.greptime_otel_resource_info"] | | host | h-1 | ["public.greptime_otel_resource_info"] | +| host | h-2 | ["public.greptime_otel_resource_info"] | +| k8s.container | uid-1,api | ["public.greptime_otel_resource_info"] | +| k8s.pod | uid-1 | ["public.greptime_otel_resource_info"] | +| k8s.pod | uid-2 | ["public.greptime_otel_resource_info"] | | service | shop/api | ["public.greptime_otel_resource_info"] | | service | worker | ["public.greptime_otel_resource_info"] | | service.instance | shop/api,inst-1 | ["public.greptime_otel_resource_info"] | | service.instance | shop/api,inst-2 | ["public.greptime_otel_resource_info"] | +| service.instance | shop/api,inst-3 | ["public.greptime_otel_resource_info"] | +| service.instance | shop/api,inst-4 | ["public.greptime_otel_resource_info"] | +------------------+-----------------+----------------------------------------+ +-- Superseding must not cost the row an attribute it carried: the runtime +-- container id stays reachable as a descriptive attribute of k8s.container. +-- SQLNESS PROTOCOL MYSQL +select entity_id, descriptive +from greptime_private.semantic_entities +where entity_type = 'k8s.container'; + ++-----------+---------------------------------------------------+ +| entity_id | descriptive | ++-----------+---------------------------------------------------+ +| uid-1,api | {"container.id":"c-2","container.name":"api-ctr"} | ++-----------+---------------------------------------------------+ + -- SQLNESS PROTOCOL MYSQL select src_type, src_id, dst_type, dst_id, rel_type, provenance from greptime_private.semantic_relationships order by rel_type, src_id, dst_id; -+------------------+-----------------+-----------+----------+----------+------------+ -| src_type | src_id | dst_type | dst_id | rel_type | provenance | -+------------------+-----------------+-----------+----------+----------+------------+ -| service.instance | shop/api,inst-1 | service | shop/api | part_of | attribute | -| service.instance | shop/api,inst-2 | service | shop/api | part_of | attribute | -| container | c-1 | host | h-1 | runs_on | attribute | -| service.instance | shop/api,inst-1 | container | c-1 | runs_on | attribute | -| service.instance | shop/api,inst-1 | host | h-1 | runs_on | attribute | -+------------------+-----------------+-----------+----------+----------+------------+ - -drop table greptime_otel_resource_info; - -Affected Rows: 0 - --- The descriptor whitelist is gated on source=opentelemetry: the same table --- shape stamped as a prometheus source contributes nothing. -create table greptime_otel_resource_info ( - greptime_timestamp timestamp(3) time index, - job string, - instance string, - "host.id" string, - greptime_value double, - primary key (job, instance, "host.id") -) with ( - 'greptime.semantic.signal_type' = 'metric', - 'greptime.semantic.source' = 'prometheus' -); - -Affected Rows: 0 - -insert into greptime_otel_resource_info values - (now(), 'shop/api', 'inst-1', 'h-1', 1); - -Affected Rows: 1 - --- SQLNESS PROTOCOL MYSQL -select entity_type, entity_id -from greptime_private.semantic_entities -order by entity_type, entity_id; - -affected_rows: 0 ++------------------+-----------------+---------------+-----------+----------+------------+ +| src_type | src_id | dst_type | dst_id | rel_type | provenance | ++------------------+-----------------+---------------+-----------+----------+------------+ +| k8s.pod | uid-1 | k8s.container | uid-1,api | contains | attribute | +| service.instance | shop/api,inst-1 | service | shop/api | part_of | attribute | +| service.instance | shop/api,inst-2 | service | shop/api | part_of | attribute | +| service.instance | shop/api,inst-3 | service | shop/api | part_of | attribute | +| service.instance | shop/api,inst-4 | service | shop/api | part_of | attribute | +| container | c-1 | host | h-1 | runs_on | attribute | +| container | c-3 | host | h-2 | runs_on | attribute | +| service.instance | shop/api,inst-1 | container | c-1 | runs_on | attribute | +| service.instance | shop/api,inst-1 | host | h-1 | runs_on | attribute | +| service.instance | shop/api,inst-3 | host | h-2 | runs_on | attribute | +| service.instance | shop/api,inst-3 | k8s.pod | uid-1 | runs_on | attribute | +| service.instance | shop/api,inst-3 | k8s.container | uid-1,api | runs_on | attribute | +| service.instance | shop/api,inst-4 | container | c-3 | runs_on | attribute | +| service.instance | shop/api,inst-4 | host | h-2 | runs_on | attribute | +| service.instance | shop/api,inst-4 | k8s.pod | uid-2 | runs_on | attribute | +| k8s.container | uid-1,api | host | h-2 | runs_on | attribute | ++------------------+-----------------+---------------+-----------+----------+------------+ drop table greptime_otel_resource_info; diff --git a/tests/cases/standalone/common/system/semantic_graph.sql b/tests/cases/standalone/common/system/semantic_graph.sql index 60d95630d9..d259a3ba00 100644 --- a/tests/cases/standalone/common/system/semantic_graph.sql +++ b/tests/cases/standalone/common/system/semantic_graph.sql @@ -150,7 +150,8 @@ insert into graph_traces_malformed values (now(), 'not a trace'); -- SQLNESS PROTOCOL MYSQL select src_id, dst_id, rel_type, provenance, confidence, - request_count, error_count, duration_sum, duration_count, attributes + request_count, unmatched_count, error_count, + duration_sum, duration_count, duration_max, attributes from greptime_private.semantic_relationships order by dst_id; @@ -394,7 +395,10 @@ drop table http_requests_total; -- the stable ids only, so rows with an empty host.id / container.id link no -- infrastructure, and a row without an instance value declares no -- service.instance. The same table name stamped source=prometheus is not --- whitelisted. +-- whitelisted. Container rows show the superseding rule: a pod's container is +-- the k8s.container entity where the row names it, and stays a generic +-- container where it does not — one node per container either way, with no +-- kube-state-metrics deployed. create table greptime_otel_resource_info ( greptime_timestamp timestamp(3) time index, job string, @@ -405,9 +409,13 @@ create table greptime_otel_resource_info ( "host.name" string, "container.id" string, "container.name" string, + "k8s.pod.uid" string, + "k8s.pod.name" string, + "k8s.container.name" string, greptime_value double, primary key (job, instance, "service.name", "service.namespace", - "host.id", "host.name", "container.id", "container.name") + "host.id", "host.name", "container.id", "container.name", + "k8s.pod.uid", "k8s.pod.name", "k8s.container.name") ) with ( 'greptime.semantic.signal_type' = 'metric', 'greptime.semantic.source' = 'opentelemetry', @@ -415,15 +423,24 @@ create table greptime_otel_resource_info ( ); insert into greptime_otel_resource_info values - (now(), 'shop/api', 'inst-1', 'api', 'shop', 'h-1', 'node-a', 'c-1', 'api-ctr', 1), - (now(), 'shop/api', 'inst-2', 'api', 'shop', '', 'laptop', '', '', 1), - (now(), 'worker', '', 'worker', '', 'h-1', 'node-a', '', '', 1); + (now(), 'shop/api', 'inst-1', 'api', 'shop', 'h-1', 'node-a', 'c-1', 'api-ctr', '', '', '', 1), + (now(), 'shop/api', 'inst-2', 'api', 'shop', '', 'laptop', '', '', '', '', '', 1), + (now(), 'worker', '', 'worker', '', 'h-1', 'node-a', '', '', '', '', '', 1), + (now(), 'shop/api', 'inst-3', 'api', 'shop', 'h-2', 'node-b', 'c-2', 'api-ctr', 'uid-1', 'api-pod', 'api', 1), + (now(), 'shop/api', 'inst-4', 'api', 'shop', 'h-2', 'node-b', 'c-3', 'api-ctr', 'uid-2', 'api-pod-2', '', 1); -- SQLNESS PROTOCOL MYSQL select entity_type, entity_id, source_tables from greptime_private.semantic_entities order by entity_type, entity_id, source_tables; +-- Superseding must not cost the row an attribute it carried: the runtime +-- container id stays reachable as a descriptive attribute of k8s.container. +-- SQLNESS PROTOCOL MYSQL +select entity_id, descriptive +from greptime_private.semantic_entities +where entity_type = 'k8s.container'; + -- SQLNESS PROTOCOL MYSQL select src_type, src_id, dst_type, dst_id, rel_type, provenance from greptime_private.semantic_relationships @@ -431,30 +448,6 @@ order by rel_type, src_id, dst_id; drop table greptime_otel_resource_info; --- The descriptor whitelist is gated on source=opentelemetry: the same table --- shape stamped as a prometheus source contributes nothing. -create table greptime_otel_resource_info ( - greptime_timestamp timestamp(3) time index, - job string, - instance string, - "host.id" string, - greptime_value double, - primary key (job, instance, "host.id") -) with ( - 'greptime.semantic.signal_type' = 'metric', - 'greptime.semantic.source' = 'prometheus' -); - -insert into greptime_otel_resource_info values - (now(), 'shop/api', 'inst-1', 'h-1', 1); - --- SQLNESS PROTOCOL MYSQL -select entity_type, entity_id -from greptime_private.semantic_entities -order by entity_type, entity_id; - -drop table greptime_otel_resource_info; - -- Cross-signal identity: the same service reaches the graph as trace spans and -- as an ingestion-synthesized descriptor. The two sources name their identity -- columns differently and only the metric side pre-composes the namespace into