feat: report what the graph derives and fix two duplicate-node bugs (#8936)

* feat(semantic-graph): let the generic container yield to k8s.container A container inside a Kubernetes pod reaches the graph twice: as the k8s.container entity kube-state-metrics describes, identified by [pod uid, container name], and as the generic container the OTel resource attributes describe, identified by container.id. One physical container, two nodes. Kubernetes is the primary scenario, so k8s.container keeps its identity and the generic type stands down where it applies. Conventions gain a row-level condition for that: `suppressed_by` withdraws a declaration on rows where any of the named columns has a value. The test has to be per row, not per table — one descriptor table holds both pod rows and bare-runtime rows. Every branch that turns a declaration into rows now shares one guard (`declaration_predicate`), so the condition cannot apply to entities but not to the edges they carry. Signed-off-by: Dennis Zhuang <killme2008@gmail.com> * feat(semantic-graph): report derived entity declarations in table_semantics information_schema.table_semantics only read table options, so the declarations the built-in conventions derive — for trace tables, for whitelisted Prometheus and OTel descriptor metrics — were invisible. "Why is my table not in the graph?" was answerable only from debug logs, which is not a self-service path. A new `entity_declarations` column reports the entities a table actually contributes: each one's identity, whether it came from an option or from the conventions, and any row-level condition attached to it. An expected entity missing from the list is the answer — the table name is not whitelisted, the source stamp is wrong, an id column is absent. The row filter widens to match: a table that declares nothing by option but derives entities by convention now appears, since it is in the graph and the view has to say so. The provider reaches the derivation through a new metadata-only trait method, keeping catalog below operator. Signed-off-by: Dennis Zhuang <killme2008@gmail.com> * fix(semantic-graph): never yield a container to an entity nothing derives The generic container withdrew on any row carrying a pod UID, on the assumption that k8s.container would cover it. Nothing guaranteed that: k8s.container came only from kube-state-metrics, so an OTLP-only deployment — or one whose KSM data had expired or fell outside the query window — lost the container node and its edges entirely instead of gaining a more specific one. The rule now names the superseding type rather than a trigger column, and withdraws only where that type's full identity is on the row. Both OTel sources declare k8s.container themselves, under the identity kube-state-metrics gives it ([pod uid, container name]), so the two sources name one node; `k8s.container.name` joins the descriptor's projected attributes to carry it. Resolution runs once every declaration for the table is known, so a superseding type that ends up undeclared — its columns are gone, or an explicit declaration of it was skipped — leaves the guard empty and the generic container standing. A container can change type; it cannot disappear. Signed-off-by: Dennis Zhuang <killme2008@gmail.com> * refactor(semantic-graph): build the declaration JSON by serializing a type The column was assembled entry by entry into a serde_json::Map, cloning every value and allocating a String per key. A Serialize struct that consumes the declaration moves the same data instead, and the field order now reads type, origin, identity, then description. The scan path around it was doing the same kind of avoidable work: declarations were derived before the predicate could discard the table, the supersession pass cloned every declaration's identity to look up one, and the option parse claimed the time index for tables that turn out to declare nothing. Signed-off-by: Dennis Zhuang <killme2008@gmail.com> * feat(semantic-graph): keep the structured identity for single-column ids `entity_id_attrs` was NULL whenever the identity came from one column, so a consumer holding `host` = `a3f2...` had no way to tell which column produced it, and no way back to the source table. Most identities are single-column — host, k8s.pod, k8s.node, service — so the common case was the opaque one. Build the JSON object unconditionally. Entity equality still reads `entity_id` alone, so this changes no merging: it records which attributes the id was assembled from, beside an id that deliberately omits them. Also corrects the `entity_id` column doc, which still described the `k=v,k=v` rendering replaced in #8904 by values joined in declared order. Signed-off-by: Dennis Zhuang <killme2008@gmail.com> * fix(semantic-graph): keep what a container carried when it yields Yielding to k8s.container cost the row two things it had as a generic container. The runtime container id, which kube_pod_container_info keeps descriptive precisely because it is the handle back to runtime logs and metrics, went missing entirely: neither an id nor an attribute of any node. And the edge vocabulary knew only the generic type, so a row with the more complete labels ended up with fewer connections than one without — the container layer no longer reached its host. Both OTel sources now keep the runtime id and name descriptive on k8s.container, and the vocabulary gains the two edges that mirror the generic type's. Nothing checks a supersession against the edge vocabulary, so that requirement is written down where the rule is. Also: the conventions-failure path now reports the explicit half as its comment always claimed, entity_declarations reports scope columns, and identifies() is private again now that only declaration_predicate calls it. The two RFCs catch up with entity_id_attrs being unconditional, the view listing convention-derived tables, and supersession existing. Signed-off-by: Dennis Zhuang <killme2008@gmail.com> * feat(semantic-graph): report unmatched clients and the longest request `real_wins` pins a `calls` edge's RED metrics to the observed span pairs: when a window's edge key holds any pair, the unmatched client spans are suppressed so one (window, edge) yields one row. That leaves no way to tell a callee that stopped responding from traffic that stopped arriving — both show up as a lower request_count. Add two columns to `semantic_relationships`: - `unmatched_count` — client spans with no server span, counted outside `real_wins` on the same row, so the suppressed population stays visible without splitting the edge into two rows. NULL for agent calls, whose inner join leaves nothing unmatched, and for declared edges. - `duration_max` — the longest single request. It goes through `real_wins`: a pair is timed by the server span while an unmatched client is timed by its own (network wait included), so mixing them would make the max describe a different population than duration_sum and duration_count. Agent calls compute it from the child spans they already aggregate. The projection contract goes from 16 to 18 columns; every branch projects both. Explicit column queries are unaffected, `SELECT *` and ordinal-based readers see the new shape. Signed-off-by: Dennis Zhuang <killme2008@gmail.com> * test(semantic-graph): drop the duplicated source-gate case The whitelist gate on `source=opentelemetry` is already covered by `otel_implicit_declarations_are_gated` and by the wrong-source case in `table_semantics`; here it only paid for another table create, insert and full graph derivation. Signed-off-by: Dennis Zhuang <killme2008@gmail.com> * style(semantic-graph): trim the comments back to what the code cannot say Several comments restated the code, repeated a rationale already stated at the type or in the RFC, or explained a test in more words than the test body. Signed-off-by: Dennis Zhuang <killme2008@gmail.com> --------- Signed-off-by: Dennis Zhuang <killme2008@gmail.com>
2026-09-05 04:58:59 +00:00 · 2026-08-31 10:39:17 +00:00
parent 6ce5747e02
commit 109257505e
17 changed files with 973 additions and 228 deletions
@@ -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

@@ -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 `<qualifier>/<value>`, 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.

@@ -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<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    id_qualifier: Option<String>,
+    #[serde(skip_serializing_if = "Vec::is_empty")]
+    superseded_by: Vec<String>,
+    #[serde(skip_serializing_if = "Vec::is_empty")]
+    descriptive: Vec<String>,
+    #[serde(skip_serializing_if = "Vec::is_empty")]
+    scope: Vec<String>,
+}
+
+impl From<TableEntityDeclaration> 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<TableEntityDeclaration>) -> Option<String> {
+    if declarations.is_empty() {
+        return None;
+    }
+    let entries: Vec<DeclarationEntry> = 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<TableInfo>,
+        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<RecordBatch> {
@@ -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)
    }
@@ -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<dyn EntityGraphProvider>;

+/// Where a table's entity declaration came from.
+pub enum DeclarationOrigin {
+    /// A `greptime.semantic.entity.<type>.*` 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<String>,
+    pub id_qualifier: Option<String>,
+    /// 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<String>,
+    pub descriptive_columns: Vec<String>,
+    pub scope_columns: Vec<String>,
+}
+
 /// 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<QueryContextRef>,
    ) -> std::result::Result<Option<SendableRecordBatchStream>, 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<TableEntityDeclaration>;
 }

 /// 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<SchemaRef> = 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<SchemaRef> = 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<SchemaRef> = LazyLock::new(|| {
@@ -257,6 +301,11 @@ static RELATIONSHIPS_SCHEMA: LazyLock<SchemaRef> = 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<SchemaRef> = LazyLock::new(|| {
            ConcreteDataType::int64_datatype(),
            true,
        ),
+        ColumnSchema::new(
+            DURATION_MAX_COLUMN,
+            ConcreteDataType::float64_datatype(),
+            true,
+        ),
        ColumnSchema::new(EDGE_ATTRIBUTES_COLUMN, json(), true),
    ]))
 });
@@ -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";
@@ -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.<type>.{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<EntityDeclaration> {
-        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<String>, Vec<String>, Vec<String>);
        let mut by_type: HashMap<String, RoleColumns> = 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<EntityDeclaration> {
        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.<type>.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<EntityDeclaration>,
+        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<EntityDeclaration>,
+        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<TableEntityDeclaration> {
+        // 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::<Vec<_>>();
+        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",
@@ -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<String>,
+    /// 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<String>,
    /// Descriptive columns snapshotted into the `descriptive` JSON (may be empty).
    pub descriptive_columns: Vec<String>,
    /// 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: `<namespace>/<name>` 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::<Vec<_>>();
-        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::<Vec<_>>()
+            })
+            .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();
@@ -57,6 +57,12 @@ pub struct ImplicitEntity {
    /// Skipped when the column is absent or empty.
    #[serde(default)]
    pub qualified_by: Option<String>,
+    /// 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<String>,
    /// 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
@@ -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
@@ -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<DataFr
            tag_utf8(PROVENANCE_COLUMN),
            ident(CONFIDENCE_COLUMN),
            ident(REQUEST_COUNT_COLUMN),
+            // A hand-declared edge asserts a dependency, not an observation:
+            // it has no span population to leave unmatched or to time.
+            lit(ScalarValue::Int64(None)).alias(UNMATCHED_COUNT_COLUMN),
            ident(ERROR_COUNT_COLUMN),
            ident(DURATION_SUM_COLUMN),
            ident(DURATION_COUNT_COLUMN),
+            lit(ScalarValue::Float64(None)).alias(DURATION_MAX_COLUMN),
            ident(EDGE_ATTRIBUTES_COLUMN),
        ])
 }
@@ -327,7 +332,8 @@ const VIRTUAL_NODE_CONFIDENCE: f64 = 0.5;
 /// and `confidence < 1.0`. Real pairs win: when a window's edge key holds any
 /// pair, the edge reports only the pair population (the RFC pins RED metrics
 /// to observed span pairs) and the unmatched clients are suppressed, so one
-/// `(window, edge)` never yields two rows.
+/// `(window, edge)` never yields two rows. `unmatched_count` stays outside the
+/// rule and counts them on the same row.
 fn calls_branch(traces: &[CallsSource], window: &GraphQueryWindow) -> DfResult<Option<DataFrame>> {
    let mut clients: Option<DataFrame> = None;
    let mut servers: Option<DataFrame> = None;
@@ -406,10 +412,14 @@ fn calls_branch(traces: &[CallsSource], window: &GraphQueryWindow) -> DfResult<O
                    .filter(paired.clone())
                    .build()?
                    .alias("pair_duration_nano"),
+                max(ident("server_duration_nano"))
+                    .filter(paired.clone())
+                    .build()?
+                    .alias("pair_duration_max_nano"),
                count(lit(1))
                    .filter(!paired.clone())
                    .build()?
-                    .alias("unmatched_count"),
+                    .alias(UNMATCHED_COUNT_COLUMN),
                count(lit(1))
                    .filter(
                        (!paired.clone()).and(ident("client_status").eq(lit(SPAN_STATUS_ERROR))),
@@ -420,6 +430,10 @@ fn calls_branch(traces: &[CallsSource], window: &GraphQueryWindow) -> DfResult<O
                    .filter(!paired.clone())
                    .build()?
                    .alias("unmatched_duration_nano"),
+                max(ident("client_duration_nano"))
+                    .filter(!paired.clone())
+                    .build()?
+                    .alias("unmatched_duration_max_nano"),
                bool_or(paired.clone()).alias("has_pair"),
                // `min` makes a mixed-provenance virtual edge deterministic
                // (`database` sorts before `virtual_node`).
@@ -441,7 +455,11 @@ fn calls_branch(traces: &[CallsSource], window: &GraphQueryWindow) -> DfResult<O
            lit(REL_TYPE_CALLS).alias(REL_TYPE_COLUMN),
            lit(PROVENANCE_TRACE).alias(PROVENANCE_COLUMN),
            real_wins(lit(1.0_f64), lit(VIRTUAL_NODE_CONFIDENCE))?.alias(CONFIDENCE_COLUMN),
-            real_wins(ident("pair_count"), ident("unmatched_count"))?.alias(REQUEST_COUNT_COLUMN),
+            real_wins(ident("pair_count"), ident(UNMATCHED_COUNT_COLUMN))?
+                .alias(REQUEST_COUNT_COLUMN),
+            // Outside `real_wins`: reporting what the pair population
+            // swallowed is the point of the column.
+            ident(UNMATCHED_COUNT_COLUMN),
            real_wins(ident("pair_errors"), ident("unmatched_errors"))?.alias(ERROR_COUNT_COLUMN),
            // duration sums in nanoseconds; the contract column is seconds.
            (cast(
@@ -452,7 +470,19 @@ fn calls_branch(traces: &[CallsSource], window: &GraphQueryWindow) -> DfResult<O
                DataType::Float64,
            ) / lit(1e9_f64))
            .alias(DURATION_SUM_COLUMN),
-            real_wins(ident("pair_count"), ident("unmatched_count"))?.alias(DURATION_COUNT_COLUMN),
+            real_wins(ident("pair_count"), ident(UNMATCHED_COUNT_COLUMN))?
+                .alias(DURATION_COUNT_COLUMN),
+            // A pair is timed by the server span, an unmatched client by its
+            // own (network wait included), so mixing them would describe a
+            // different population than duration_sum/duration_count.
+            (cast(
+                real_wins(
+                    ident("pair_duration_max_nano"),
+                    ident("unmatched_duration_max_nano"),
+                )?,
+                DataType::Float64,
+            ) / lit(1e9_f64))
+            .alias(DURATION_MAX_COLUMN),
            real_wins(null_json(), virtual_attrs_expr()?)?.alias(EDGE_ATTRIBUTES_COLUMN),
        ])?;
    Ok(Some(df))
@@ -484,7 +514,7 @@ fn virtual_attrs_expr() -> DfResult<Expr> {
 /// 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::<Int64Array>()
-            .unwrap();
-        assert_eq!(request_count.value(0), 2);
-        let error_count = batch
-            .column(12)
-            .as_any()
-            .downcast_ref::<Int64Array>()
-            .unwrap();
-        assert_eq!(error_count.value(0), 1);
-        let duration_sum = batch
-            .column(13)
-            .as_any()
-            .downcast_ref::<Float64Array>()
-            .unwrap();
-        assert!((duration_sum.value(0) - 2.0).abs() < 1e-9);
-        let duration_count = batch
-            .column(14)
-            .as_any()
-            .downcast_ref::<Int64Array>()
-            .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::<Float64Array>()
            .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::<Int64Array>()
+            .unwrap()
+            .value(row)
+    }
+
+    fn duration_max(batch: &RecordBatch, row: usize) -> f64 {
+        batch
+            .column(16)
+            .as_any()
+            .downcast_ref::<Float64Array>()
+            .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(),
@@ -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
@@ -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";

@@ -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
+
@@ -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;
@@ -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              |                |        |
@@ -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;

@@ -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