feat: synthesize OTLP resource descriptor for the semantic entity graph (#8904)

* fix(servers): compose OTLP metrics job from service.namespace/service.name

The OTel Prometheus compatibility spec defines job as
"<service.namespace>/<service.name>" when the namespace is present.
The OTLP metrics path only used the bare service.name, so the job tag
diverged from target_info produced by Prometheus-side exporters for the
same resource. Compose the namespace form, and keep not fabricating a
job when service.name is absent.

Behavior change: resources carrying service.namespace now get
"namespace/name" as their job tag value.

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* feat(otlp): synthesize otel_resource_info at OTLP metrics ingestion

Ordinary OTLP metrics scatter filtered resource attributes as tags over
every logical metric table, so metrics-only services contribute nothing
to the semantic entity graph. Each request now also projects its
distinct resources into one info-metric-shaped mito table,
otel_resource_info: a fixed allowlist of identity-relevant attributes
under their raw OTel keys (independent of the label translation
strategy and the promote/ignore headers) plus derived job/instance
compatibility columns, value 1.0, and the newest data-point timestamp.

The descriptor is written after the main insert is committed; a failure
there (conflicting pre-existing table, auto-create disabled) degrades
to an OTLP partial_success warning with rejected_data_points = 0
instead of failing the request and triggering client retries of
already-accepted data. A request writing a metric named
otel_resource_info suppresses synthesis. Legacy mode is unchanged.

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* feat(operator): otel info-metric conventions with host/container entities

Whitelist the ingestion-synthesized otel_resource_info descriptor via a
new otel_info_metrics conventions map, gated on source=opentelemetry
(the existing gate hardcoded source=prometheus). Its declarations use
explicit descriptive lists instead of descriptive_rest so identifying
attributes of other entities do not leak into service.instance.

Conventions tightened per the Astronomy Shop findings: host identity is
host.id with host.name descriptive only (host.name is not stable across
SDKs and resource detectors), a generic container entity (new entity
type) is declared only when container.id is present, and trace-v1
tables now synthesize host/container from their flattened resource
attributes too. New co-declared edges: service.instance runs_on
container, container runs_on host.

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* test(otlp): cover the resource descriptor in integration tests

Covers the descriptor's raw-key columns and info-metric options through
the HTTP path, the namespace/name job composition end-to-end, column
names being independent of the translation strategy, the allowlist
excluding unlisted resource attributes, auto-create after a drop, the
metric-name collision suppressing synthesis, and the partial-success
warning (rejected_data_points = 0) when a pre-existing incompatible
table fails the descriptor write while metric data is accepted.

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* chore: cargo fmt

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* fix(frontend): degrade descriptor permission denial to a warning

A table-level permission policy denying otel_resource_info would have
failed the whole OTLP metrics request because the descriptor's
permission check ran before the main insert. The descriptor is derived
enrichment: check its permission in the degrade path so a denial skips
the write and surfaces as the partial-success warning, like any other
descriptor write failure.

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* fix(otlp): guard descriptor writes with semantic ownership markers

A pre-existing schema-compatible table named otel_resource_info would
silently receive descriptor rows while its missing semantic stamps kept
it out of the entity graph. The descriptor write now requires the
auto-created table's ownership markers (mito engine + signal_type +
source + metric.type=info + metadata_quality=declared) and otherwise
degrades to the partial-success warning; the entity-graph gate for the
otel whitelist likewise requires metric.type=info, so a user table
stamped with only signal/source no longer picks up implicit
declarations.

Also fold the descriptor write cost into the response and surface the
degrade warning through the otel-arrow BatchStatus status_message.
Integration tests pin the full marker set on auto-create and that an
existing owned descriptor keeps accepting writes without degrading —
a missing marker would otherwise silently stop every descriptor write
after the first request.

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* perf(otlp): build descriptor rows without the per-resource BTreeMap

Projecting a resource allocated a BTreeMap and then collected it into the
row key, and every attribute was matched against the allowlist by linear
scan. Collect the tags into a Vec and sort once, and match the allowlist
instead of scanning it. Measured on the conversion path: descriptor work
drops 16-18%, from 10.6% to 8.9% of conversion CPU on the worst shape
(1000 resources with 4 data points each), where the cost tracks resource
count rather than data-point count.

Also trims the comments and tests added with the descriptor to what
carries information.

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* test(otlp): pin the descriptor permission-denial degrade path

A policy denying the descriptor table must not fail the metrics request,
which the fix in 2401b3dd9c does but nothing covered. Verified as a
regression guard by mutation: moving the permission check back before
the main write makes this test fail.

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* test(otlp): keep legacy mode covered after trimming the unit tests

Trimming the descriptor tests dropped the only assertion that legacy
mode skips the job/instance remap and the promote filter. Both alter
the columns of tables already in use, so fold the check into the legacy
conversion test rather than leaving it uncovered.

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* fix(semantic-graph): stop encoding column names into composite entity ids

A composite entity id rendered the identifying columns as sorted
`col=value` pairs, so the same identity split into one entity per
signal: a trace table names its columns service_name and
resource_attributes.service.instance.id where a metric table names them
job and instance. One service instance became two nodes with two
parallel edge sets, breaking the walk from a trace to that instance's
metrics.

Render an id as its values in declared order instead, escaping the
separator so components stay distinguishable, which is what single-column
ids already did by keeping only the value. entity_id_attrs still carries
the structured form.

Values alone are not enough for a namespaced service: the metric side
folds service.namespace into job while traces keep the bare name. Add
qualified_by to the conventions so the trace declarations compose the
namespace the same way, per the OTel rule that job is
<service.namespace>/<service.name> or the bare name when the namespace
is empty. A table without the namespace column keeps the unqualified
identity rather than losing the declaration.

Conventions validation now rejects one entity type declared with a
different number of id columns by two sources, which would silently
produce ids that can never match.

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* style(otlp): import the parent module by crate path

check-super-imports.py, part of the CI format gate, rejects a
file-level `use super::`.

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* feat(otlp): gate the resource descriptor, and fix what review found

Synthesizing greptime_otel_resource_info creates and writes a table the
user never sent, so it is now off unless
otlp.experimental_enable_resource_info says otherwise. With it off the
request costs exactly what it did before the descriptor existed: nothing
is projected, no table is created, no write and no permission check
happen. Tests run with it on. StandaloneOptions carried no otlp field,
so the whole [otlp] section was silently dropped in standalone mode; map
it through, or the new option (and trace_ingest_chunk_size before it)
would do nothing there.

Renamed from otel_resource_info: the greptime_ prefix marks the table as
engine-managed and makes a collision with a user metric unlikely, which
is what the pre-existing-table ownership check and its per-request
catalog lookup were defending against. Both are gone.

A request may carry data for several graph windows, but the descriptor
folded every data-point time into one row at the newest of them, leaving
the earlier windows with metric rows and no entities. Key the rows by
window as well, and take the times from the data points the encoder
actually writes: it drops exponential histograms, and a resource
carrying nothing else was being described as an entity with no
measurements.

Projecting a resource cloned its attributes once per data point. Nest
the windows under the attributes instead, so they are moved once per
resource, and walk the data-point times through a visitor rather than
collecting a Vec per metric.

Also documents what the two maps key and hold, and lifts the projected
attribute names to constants beside KEY_SERVICE_NAME.

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* fix(otlp): skip the descriptor's work entirely when it is disabled

The collision scan over the request's output tables ran even with the
feature off. Short-circuit on the option instead, and update the config
snapshot the new [otlp] section changed.

Also drops the doc comment orphaned by the deleted ownership check: it
had attached itself to the trait impl and described a check that no
longer exists.

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* refactor(semantic-graph): drop the expect and name the service identity

The CASE is built through Case directly rather than the fallible
when().otherwise() builder, so the non-test path no longer carries an
expect (architecture-invariants $4).

service_identity returned two same-typed Options that both call sites
destructured positionally; a named struct makes a swap fail to compile.

Also records that id-column order is part of the identity, where the
option docs and the conventions authors will read it.

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* chore(semantic-graph): drop comments that narrate the code

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* fix(semantic-graph): cast duration_nano before the trace-table union

Trace tables written before the signed-integer ingest change hold
duration_nano as UInt64 and later ones as Int64. The calls derivation
unions the per-table selects, and the two have no common integer type,
so a deployment holding both shapes could not build the plan. The
cross-table test now spans both.

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* refactor(semantic-graph): drop the redundant duration_nano casts

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* fix(otlp): decide exponential histogram acceptance in one place

The resource descriptor mirrored only the experimental gate, so with both
experimental flags on a resource whose only metric is a delta exponential
histogram was described as an entity with no measurements. The encoder's
whole-metric rules move into exponential_histogram_gate, which both call,
and the descriptor takes its timestamps through exponential_histogram_value
so per-point rejections drop out too.

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

---------

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

config/config.md

@@ -76,6 +76,7 @@
 | `otlp.enable` | Bool | `true` | Whether to enable OpenTelemetry protocol in HTTP API. |
 | `otlp.experimental_enable_exponential_histogram` | Bool | `false` | Experimental: enable cumulative OTLP exponential histogram ingestion. |
 | `otlp.trace_ingest_chunk_size` | Integer | `512` | Maximum spans per trace ingest chunk. Set to 0 to disable splitting. |
+| `otlp.experimental_enable_resource_info` | Bool | `false` | Whether to synthesize the `greptime_otel_resource_info` table from OTLP metric<br/>resource attributes, so metrics-only services reach the semantic graph. |
 | `prom_store` | -- | -- | Prometheus remote storage options |
 | `prom_store.enable` | Bool | `true` | Whether to enable Prometheus remote write and read in HTTP API. |
 | `prom_store.with_metric_engine` | Bool | `true` | Whether to store the data from Prometheus remote write in metric engine. |
@@ -319,6 +320,7 @@
 | `otlp.enable` | Bool | `true` | Whether to enable OpenTelemetry protocol in HTTP API. |
 | `otlp.experimental_enable_exponential_histogram` | Bool | `false` | Experimental: enable cumulative OTLP exponential histogram ingestion. |
 | `otlp.trace_ingest_chunk_size` | Integer | `512` | Maximum spans per trace ingest chunk. Set to 0 to disable splitting. |
+| `otlp.experimental_enable_resource_info` | Bool | `false` | Whether to synthesize the `greptime_otel_resource_info` table from OTLP metric<br/>resource attributes, so metrics-only services reach the semantic graph. |
 | `prom_store` | -- | -- | Prometheus remote storage options |
 | `prom_store.enable` | Bool | `true` | Whether to enable Prometheus remote write and read in HTTP API. |
 | `prom_store.with_metric_engine` | Bool | `true` | Whether to store the data from Prometheus remote write in metric engine. |

config/frontend.example.toml

@@ -241,6 +241,9 @@ enable = true
 experimental_enable_exponential_histogram = false
 ## Maximum spans per trace ingest chunk. Set to 0 to disable splitting.
 trace_ingest_chunk_size = 512
+## Whether to synthesize the `greptime_otel_resource_info` table from OTLP metric
+## resource attributes, so metrics-only services reach the semantic graph.
+experimental_enable_resource_info = false
 
 ## Prometheus remote storage options
 [prom_store]

config/standalone.example.toml

@@ -208,6 +208,9 @@ enable = true
 experimental_enable_exponential_histogram = false
 ## Maximum spans per trace ingest chunk. Set to 0 to disable splitting.
 trace_ingest_chunk_size = 512
+## Whether to synthesize the `greptime_otel_resource_info` table from OTLP metric
+## resource attributes, so metrics-only services reach the semantic graph.
+experimental_enable_resource_info = false
 
 ## Prometheus remote storage options
 [prom_store]

docs/rfcs/2026-06-25-entity-relationships-and-graph-query.md

@@ -149,7 +149,7 @@ The decisions behind the shape:
 - **`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.
-- **Endpoint encoding (v1).** A single-attribute identity is the value verbatim; a composite identity is the attributes sorted by key, rendered `k1=v1,k2=v2` — a human-readable traversal key, not a collision-free canonical encoding. Nodes carry the structured `entity_id_attrs`; edges carry only the string key. Known v1 limitation: composite components containing unescaped `,` or `=` are not guaranteed to be distinguished.
+- **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`.
 
 The relationship vocabulary is small, typed, and inverse-paired; stored direction is `src -> dst`, the inverse is a query concern:
@@ -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; single-column ids are used verbatim, composite ids render the sorted `k=v` form plus the `entity_id_attrs` JSON. 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 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*).
 
 **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.
 
@@ -284,7 +284,7 @@ Drawbacks:
 
 - Read-time derivation costs a self-join over trace tables on every scan of the computed tables. The default one-hour window bounds it, but very large deployments will want materialisation (Future Work).
 - `semantic_entities` exposes per-observation rows; deduplication falls on the consumer (`DISTINCT`) until the snapshot relation lands in M2.
-- The v1 `k=v` id encoding can collide on hostile values (unescaped `,`/`=`); hardening is an open question.
+- The id encoding is readable rather than canonical: escaping keeps components distinguishable, but a value containing the qualifier separator (`/`) is still ambiguous against a qualified name.
 - The graph is only as connected as the identity columns tables actually share: the `job`/`service_name` misalignment above is the canonical example, and enrichment discipline gates connectivity.
 
 # Implementation Plan
@@ -306,7 +306,7 @@ Drawbacks:
 
 # Open Questions
 
-1. **Endpoint encoding hardening.** The v1 `k=v` rendering (with `entity_id_attrs` JSON as the source of truth) is readable but collides when values contain `,`/`=`. Open: a fixed-width `entity_id_hash` join key (New-Relic-style); typed value encoding; the criteria for introducing the `semantic_key=column` mapping form; `schema_url`-gated entity merge. Related and unresolved: whether the identifying attribute *key* should participate in entity equality — a single-attribute id keeps only the value, so `job=api` and `service_name=api` currently name the same entity even when the keys differ semantically.
+1. **Endpoint encoding hardening.** The rendering (with `entity_id_attrs` JSON as the source of truth) is readable rather than canonical. Open: a fixed-width `entity_id_hash` join key (New-Relic-style); typed value encoding; the criteria for introducing the `semantic_key=column` mapping form; `schema_url`-gated entity merge. Settled: identifying attribute *keys* do not participate in entity equality, so `job=api` and `service_name=api` name the same entity — the property that lets one service reach the graph from several signals.
 2. **Identity reconciliation across signals.** Trace edges key on the trace table's service declaration; host edges key on `host.id`/`k8s.pod.uid`. Linking them relies on tables carrying both columns; how aggressively to enrich (a `k8sattributes`-equivalent) gates how connected the graph is.
 3. **`GRAPH_TABLE` surface scope.** How much of SQL/PGQ to implement vs. a minimal single-hop subset. Start minimal and grow with demand.
 4. **Visibility beyond table grants.** The derivation contract filters by table-read permission; whether topology needs finer policies (row-level, tenant-scoped visibility of cross-schema edges) is open.

src/common/catalog/src/consts.rs

@@ -220,6 +220,11 @@ pub const SEMANTIC_RELATIONSHIPS_TABLE_NAME: &str = "semantic_relationships";
 /// `semantic_relationships`; see [`is_ddl_reserved_table`] for its lifecycle.
 pub const SEMANTIC_RELATIONSHIPS_DECLARED_TABLE_NAME: &str = "semantic_relationships_declared";
 
+/// Width of the window the graph derivation bins observations into. Sources
+/// synthesizing observations must land a row in every window they describe,
+/// so this is shared rather than restated per crate.
+pub const SEMANTIC_GRAPH_WINDOW_NANOS: i64 = 60 * 1_000_000_000;
+
 // Column names of the graph tables: `catalog` exposes these schemas and the
 // read-time plans in `operator` must project exactly them.
 pub const OBSERVED_AT_COLUMN: &str = "observed_at";

src/frontend/src/instance.rs

@@ -134,6 +134,7 @@ pub struct Instance {
     slow_query_options: SlowQueryOptions,
     influxdb_default_merge_mode: InfluxdbMergeMode,
     trace_ingest_chunk_size: usize,
+    otlp_resource_info: bool,
     suspend: Arc<AtomicBool>,
 
     // cache for otlp metrics

src/frontend/src/instance/builder.rs

@@ -367,6 +367,7 @@ impl FrontendBuilder {
             slow_query_options: self.options.slow_query.clone(),
             influxdb_default_merge_mode: self.options.influxdb.default_merge_mode,
             trace_ingest_chunk_size: self.options.otlp.trace_ingest_chunk_size,
+            otlp_resource_info: self.options.otlp.experimental_enable_resource_info,
             suspend: Arc::new(AtomicBool::new(false)),
         })
     }

src/frontend/src/instance/entity_graph.rs

@@ -23,7 +23,7 @@
 //! the query engine. Injected into the catalog manager after the engine is built,
 //! breaking the `catalog -> query` cycle.
 
-use std::collections::HashMap;
+use std::collections::{BTreeMap, HashMap};
 use std::sync::Weak;
 
 use async_trait::async_trait;
@@ -60,8 +60,9 @@ use table::TableRef;
 use table::metadata::TableInfo;
 use table::predicate::{TimeRangeExtraction, extract_time_range_strict};
 use table::requests::{
-    EntityRole, SEMANTIC_SIGNAL_TYPE, SEMANTIC_SOURCE, SIGNAL_TYPE_METRIC, SOURCE_PROMETHEUS,
-    is_trace_v1_table, parse_entity_columns, parse_entity_option_key,
+    EntityRole, SEMANTIC_METRIC_TYPE, SEMANTIC_SIGNAL_TYPE, SEMANTIC_SOURCE, SIGNAL_TYPE_METRIC,
+    SOURCE_OPENTELEMETRY, SOURCE_PROMETHEUS, is_trace_v1_table, parse_entity_columns,
+    parse_entity_option_key,
 };
 
 use crate::error;
@@ -177,6 +178,7 @@ impl EntityGraphProviderImpl {
                         time_index: time_index.clone(),
                         entity_type,
                         id_columns,
+                        id_qualifier: None,
                         descriptive_columns,
                         scope_columns,
                     })
@@ -188,10 +190,11 @@ impl EntityGraphProviderImpl {
     /// All entity declarations of one table: the explicit options plus the
     /// zero-configuration conventions (`otlp_trace_entities` for trace-v1
     /// tables — including the `service` identity of tables created before the
-    /// ingest-side auto-stamp — and the Prometheus descriptor whitelist). An
-    /// explicit declaration of a type always suppresses the implicit one, even
-    /// when the explicit declaration is invalid and skipped: silently falling
-    /// back would change entity identity behind the user's back.
+    /// ingest-side auto-stamp — and the Prometheus/OTel descriptor
+    /// whitelists). An explicit declaration of a type always suppresses the
+    /// implicit one, even when the explicit declaration is invalid and
+    /// skipped: silently falling back would change entity identity behind the
+    /// user's back.
     fn declarations_for(
         table_info: &TableInfo,
         conventions: &Conventions,
@@ -204,7 +207,20 @@ impl EntityGraphProviderImpl {
                 &mut declarations,
             );
         }
-        Self::extend_with_prometheus_conventions(table_info, conventions, &mut declarations);
+        Self::extend_with_info_metric_conventions(
+            table_info,
+            &conventions.prometheus_info_metrics,
+            SOURCE_PROMETHEUS,
+            None,
+            &mut declarations,
+        );
+        Self::extend_with_info_metric_conventions(
+            table_info,
+            &conventions.otel_info_metrics,
+            SOURCE_OPENTELEMETRY,
+            Some(servers::semantic::METRIC_TYPE_INFO),
+            &mut declarations,
+        );
         declarations
     }
 
@@ -216,29 +232,34 @@ impl EntityGraphProviderImpl {
         })
     }
 
-    /// Implicit declarations of the well-known Prometheus entity-descriptor
-    /// metrics (the `prometheus_info_metrics` whitelist of `conventions.yaml`),
-    /// gated on the ingest-stamped `signal_type=metric` + `source=prometheus`
-    /// options and keyed by table name. The metric engine's physical table
+    /// Implicit declarations of the well-known entity-descriptor metrics
+    /// (the `prometheus_info_metrics` / `otel_info_metrics` whitelists of
+    /// `conventions.yaml`), gated on the ingest-stamped `signal_type=metric`
+    /// option plus the whitelist's expected `source`; OTel descriptors also
+    /// require `metric.type=info`. The metric engine's physical table
     /// aggregates every logical table's columns and must not contribute a
     /// duplicate source.
-    fn extend_with_prometheus_conventions(
+    fn extend_with_info_metric_conventions(
         table_info: &TableInfo,
-        conventions: &Conventions,
+        whitelist: &BTreeMap<String, Vec<ImplicitEntity>>,
+        expected_source: &str,
+        expected_metric_type: Option<&str>,
         declarations: &mut Vec<EntityDeclaration>,
     ) {
-        let Some(implicit_entities) = conventions.prometheus_info_metrics.get(&table_info.name)
-        else {
+        let Some(implicit_entities) = whitelist.get(&table_info.name) else {
             return;
         };
         let options = &table_info.meta.options.extra_options;
         if options.get(SEMANTIC_SIGNAL_TYPE).map(String::as_str) != Some(SIGNAL_TYPE_METRIC)
-            || options.get(SEMANTIC_SOURCE).map(String::as_str) != Some(SOURCE_PROMETHEUS)
+            || options.get(SEMANTIC_SOURCE).map(String::as_str) != Some(expected_source)
+            || expected_metric_type.is_some_and(|expected| {
+                options.get(SEMANTIC_METRIC_TYPE).map(String::as_str) != Some(expected)
+            })
             || table_info.is_physical_table()
         {
             debug!(
-                "Table `{}` matches the info-metric whitelist but is not a prometheus logical \
-                 metric table; skipping its implicit declarations",
+                "Table `{}` matches the info-metric whitelist but is not an eligible \
+                 `{expected_source}` info-metric source; skipping its implicit declarations",
                 table_info.name
             );
             return;
@@ -296,12 +317,19 @@ impl EntityGraphProviderImpl {
                     .cloned()
                     .collect()
             };
+            // A table predating the qualifier column keeps the unqualified
+            // identity rather than losing the declaration.
+            let id_qualifier = implicit
+                .qualified_by
+                .clone()
+                .filter(|c| schema.column_schema_by_name(c).is_some());
             declarations.push(EntityDeclaration {
                 schema: table_info.schema_name.clone(),
                 table: table_info.name.clone(),
                 time_index: time_index.clone(),
                 entity_type: implicit.entity.clone(),
                 id_columns: implicit.id.clone(),
+                id_qualifier,
                 descriptive_columns,
                 scope_columns: vec![],
             });
@@ -832,6 +860,24 @@ mod tests {
             declarations[2].id_columns,
             vec!["service_name", "resource_attributes.service.instance.id"]
         );
+        assert_eq!(declarations[1].id_qualifier, None);
+
+        let namespaced = table_info(
+            &[
+                "service_name",
+                "resource_attributes.service.namespace",
+                "resource_attributes.service.instance.id",
+            ],
+            &[(TABLE_DATA_MODEL, TABLE_DATA_MODEL_TRACE_V1)],
+        );
+        for declaration in sorted_declarations(&namespaced) {
+            assert_eq!(
+                declaration.id_qualifier.as_deref(),
+                Some("resource_attributes.service.namespace"),
+                "{} must qualify its identity like the metric side's job",
+                declaration.entity_type
+            );
+        }
 
         // Missing uid column: no pod entity synthesized, no name-based guess.
         let no_uid = table_info(
@@ -856,6 +902,47 @@ mod tests {
         assert!(sorted_declarations(&invalid_explicit).is_empty());
     }
 
+    #[test]
+    fn trace_table_host_and_container_require_stable_ids() {
+        let with_ids = table_info(
+            &[
+                "service_name",
+                "resource_attributes.host.id",
+                "resource_attributes.host.name",
+                "resource_attributes.container.id",
+            ],
+            &[(TABLE_DATA_MODEL, TABLE_DATA_MODEL_TRACE_V1)],
+        );
+        let declarations = sorted_declarations(&with_ids);
+        let types: Vec<&str> = declarations
+            .iter()
+            .map(|d| d.entity_type.as_str())
+            .collect();
+        assert_eq!(types, vec!["container", "host", "service"]);
+        assert_eq!(
+            declarations[1].id_columns,
+            vec!["resource_attributes.host.id"]
+        );
+        assert_eq!(
+            declarations[1].descriptive_columns,
+            vec!["resource_attributes.host.name"]
+        );
+
+        let names_only = table_info(
+            &[
+                "service_name",
+                "resource_attributes.host.name",
+                "resource_attributes.container.name",
+            ],
+            &[(TABLE_DATA_MODEL, TABLE_DATA_MODEL_TRACE_V1)],
+        );
+        let types: Vec<String> = sorted_declarations(&names_only)
+            .into_iter()
+            .map(|d| d.entity_type)
+            .collect();
+        assert_eq!(types, vec!["service"]);
+    }
+
     #[test]
     fn prometheus_implicit_declarations_are_gated() {
         let labels: &[&str] = &["namespace", "pod", "node"];
@@ -900,6 +987,111 @@ mod tests {
         assert_eq!(declarations[1].id_columns, vec!["pod"]);
     }
 
+    const OTEL_STAMPS: &[(&str, &str)] = &[
+        (SEMANTIC_SIGNAL_TYPE, SIGNAL_TYPE_METRIC),
+        (SEMANTIC_SOURCE, SOURCE_OPENTELEMETRY),
+        (SEMANTIC_METRIC_TYPE, servers::semantic::METRIC_TYPE_INFO),
+    ];
+
+    #[test]
+    fn greptime_otel_resource_info_gets_implicit_declarations() {
+        let info = prom_table_info(
+            "greptime_otel_resource_info",
+            &[
+                "job",
+                "instance",
+                "service.name",
+                "service.namespace",
+                "host.id",
+                "host.name",
+                "container.id",
+                "container.name",
+                "k8s.pod.uid",
+                "k8s.pod.name",
+                "k8s.namespace.name",
+            ],
+            OTEL_STAMPS,
+        );
+        let declarations = sorted_declarations(&info);
+        let types: Vec<&str> = declarations
+            .iter()
+            .map(|d| d.entity_type.as_str())
+            .collect();
+        assert_eq!(
+            types,
+            vec![
+                "container",
+                "host",
+                "k8s.pod",
+                "service",
+                "service.instance"
+            ]
+        );
+        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[3].descriptive_columns,
+            vec!["service.name", "service.namespace"]
+        );
+        assert_eq!(declarations[4].id_columns, vec!["job", "instance"]);
+        assert!(declarations[4].descriptive_columns.is_empty());
+
+        let partial = prom_table_info(
+            "greptime_otel_resource_info",
+            &["job", "service.name", "host.id"],
+            OTEL_STAMPS,
+        );
+        let types: Vec<String> = sorted_declarations(&partial)
+            .into_iter()
+            .map(|d| d.entity_type)
+            .collect();
+        assert_eq!(types, vec!["host", "service"]);
+    }
+
+    #[test]
+    fn otel_implicit_declarations_are_gated() {
+        let labels: &[&str] = &["job", "instance", "host.id"];
+        assert!(
+            sorted_declarations(&prom_table_info(
+                "greptime_otel_resource_info",
+                labels,
+                PROM_STAMPS
+            ))
+            .is_empty()
+        );
+        let mut stamps = OTEL_STAMPS.to_vec();
+        stamps.push((PHYSICAL_TABLE_METADATA_KEY, "true"));
+        assert!(
+            sorted_declarations(&prom_table_info(
+                "greptime_otel_resource_info",
+                labels,
+                &stamps
+            ))
+            .is_empty()
+        );
+        assert!(
+            conventions()
+                .unwrap()
+                .otel_info_metrics
+                .contains_key(servers::otlp::metrics::OTEL_RESOURCE_INFO_TABLE_NAME)
+        );
+
+        let wrong_type = [
+            (SEMANTIC_SIGNAL_TYPE, SIGNAL_TYPE_METRIC),
+            (SEMANTIC_SOURCE, SOURCE_OPENTELEMETRY),
+            (SEMANTIC_METRIC_TYPE, servers::semantic::METRIC_TYPE_GAUGE),
+        ];
+        assert!(
+            sorted_declarations(&prom_table_info(
+                "greptime_otel_resource_info",
+                labels,
+                &wrong_type
+            ))
+            .is_empty()
+        );
+    }
+
     #[test]
     fn target_info_descriptive_rest_covers_remaining_tags() {
         let info = prom_table_info(

src/frontend/src/instance/otlp.rs

@@ -50,7 +50,7 @@ use table::requests::{
 
 use self::trace_ingest::trace_conventions;
 use crate::instance::Instance;
-use crate::metrics::{OTLP_LOGS_ROWS, OTLP_METRICS_ROWS};
+use crate::metrics::{OTLP_LOGS_ROWS, OTLP_METRICS_ROWS, OTLP_RESOURCE_INFO_WRITE_ERRORS};
 
 fn trace_permission_targets(
     table_name: &str,
@@ -119,9 +119,15 @@ impl OpenTelemetryProtocolHandler for Instance {
             .cloned()
             .unwrap_or_default();
         metric_ctx.is_legacy = is_legacy;
+        metric_ctx.resource_info = self.otlp_resource_info;
 
-        let (requests, rows, semantic_index, mut outcome) =
-            otlp::metrics::to_grpc_insert_requests(request, &mut metric_ctx)?;
+        let otlp::metrics::MetricsConversion {
+            requests,
+            rows,
+            semantic_index,
+            resource_info,
+            mut outcome,
+        } = otlp::metrics::to_grpc_insert_requests(request, &mut metric_ctx)?;
         if outcome.rejected_data_points > 0 {
             warn!(
                 "Rejected {} OTLP exponential histogram data points: {}",
@@ -156,7 +162,7 @@ impl OpenTelemetryProtocolHandler for Instance {
 
         // OTLP tables have one sample field in both the legacy and physical paths.
         let output = if metric_ctx.is_legacy || !metric_ctx.with_metric_engine {
-            self.handle_row_inserts(requests, ctx, false, true)
+            self.handle_row_inserts(requests, ctx.clone(), false, true)
                 .await
                 .map_err(BoxedError::new)
                 .context(error::ExecuteGrpcQuerySnafu)
@@ -165,12 +171,43 @@ impl OpenTelemetryProtocolHandler for Instance {
                 .extension(PHYSICAL_TABLE_PARAM)
                 .unwrap_or(GREPTIME_PHYSICAL_TABLE)
                 .to_string();
-            self.handle_metric_row_inserts(requests, ctx, physical_table)
+            self.handle_metric_row_inserts(requests, ctx.clone(), physical_table)
                 .await
                 .map_err(BoxedError::new)
                 .context(error::ExecuteGrpcQuerySnafu)
         }?;
         outcome.write_cost = output.meta.cost;
+
+        // Derived enrichment, written after the metric data is committed:
+        // failing here would make the client retry data the server already
+        // accepted, so every failure degrades to a warning instead.
+        if let Some(resource_info) = resource_info {
+            let written = match self.check_row_insert_permission(
+                &resource_info,
+                &ctx,
+                PermissionReq::Action(OTLP_WRITE),
+            ) {
+                Ok(_) => self
+                    .handle_row_inserts(resource_info, ctx, false, false)
+                    .await
+                    .map_err(BoxedError::new)
+                    .map_err(|e| e.to_string()),
+                Err(e) => Err(e.to_string()),
+            };
+            match written {
+                Ok(descriptor_output) => outcome.write_cost += descriptor_output.meta.cost,
+                Err(e) => {
+                    OTLP_RESOURCE_INFO_WRITE_ERRORS.inc();
+                    warn!("Failed to write the OTLP resource descriptor table: {e}");
+                    outcome.error_message.get_or_insert(format!(
+                        "metric data was accepted, but writing the resource \
+                         descriptor table `{}` failed: {e}",
+                        otlp::metrics::OTEL_RESOURCE_INFO_TABLE_NAME
+                    ));
+                }
+            }
+        }
+
         Ok(outcome)
     }
 

src/frontend/src/metrics.rs

@@ -45,6 +45,14 @@ lazy_static! {
     )
     .unwrap();
 
+    /// Failed writes of the synthesized OTLP resource descriptor table; these
+    /// surface as OTLP partial-success warnings, not request failures.
+    pub static ref OTLP_RESOURCE_INFO_WRITE_ERRORS: IntCounter = register_int_counter!(
+        "greptime_frontend_otlp_resource_info_write_errors",
+        "frontend otlp resource descriptor write errors"
+    )
+    .unwrap();
+
     /// The number of OpenTelemetry traces send by frontend node.
     pub static ref OTLP_TRACES_ROWS: IntCounter = register_int_counter!(
         "greptime_frontend_otlp_traces_rows",

src/frontend/src/service_config/otlp.rs

@@ -23,6 +23,11 @@ pub struct OtlpOptions {
     pub experimental_enable_exponential_histogram: bool,
     /// Maximum spans per trace ingest chunk. Set to 0 to disable splitting.
     pub trace_ingest_chunk_size: usize,
+    /// Whether to synthesize the `greptime_otel_resource_info` descriptor
+    /// table from the resource attributes of OTLP metrics, so metrics-only
+    /// services reach the semantic graph. Off by default: it creates and
+    /// writes a table the user did not send.
+    pub experimental_enable_resource_info: bool,
 }
 
 impl Default for OtlpOptions {
@@ -31,6 +36,7 @@ impl Default for OtlpOptions {
             enable: true,
             experimental_enable_exponential_histogram: false,
             trace_ingest_chunk_size: DEFAULT_TRACE_INGEST_CHUNK_SIZE,
+            experimental_enable_resource_info: false,
         }
     }
 }
@@ -45,6 +51,7 @@ mod tests {
         assert!(default.enable);
         assert!(!default.experimental_enable_exponential_histogram);
         assert_eq!(default.trace_ingest_chunk_size, 512);
+        assert!(!default.experimental_enable_resource_info);
 
         let options: OtlpOptions = toml::from_str("enable = false").unwrap();
         assert!(!options.enable);

src/operator/src/statement/semantic_graph.rs

@@ -41,8 +41,9 @@ use common_catalog::consts::{
     ENTITY_DESCRIPTIVE_COLUMN, ENTITY_ID_ATTRS_COLUMN, ENTITY_ID_COLUMN, ENTITY_SCOPE_COLUMN,
     ENTITY_TYPE_COLUMN, ERROR_COUNT_COLUMN, FRESH_UNTIL_COLUMN, GENERATION_ID_COLUMN,
     OBSERVED_AT_COLUMN, PROVENANCE_COLUMN, REL_TYPE_COLUMN, REQUEST_COUNT_COLUMN,
-    SEMANTIC_RELATIONSHIPS_DECLARED_TABLE_NAME, SOURCE_TABLES_COLUMN, SRC_ID_COLUMN,
-    SRC_TYPE_COLUMN, VALID_FROM_COLUMN, VALID_UNTIL_COLUMN, WINDOW_END_COLUMN, WINDOW_START_COLUMN,
+    SEMANTIC_GRAPH_WINDOW_NANOS, SEMANTIC_RELATIONSHIPS_DECLARED_TABLE_NAME, SOURCE_TABLES_COLUMN,
+    SRC_ID_COLUMN, SRC_TYPE_COLUMN, VALID_FROM_COLUMN, VALID_UNTIL_COLUMN, WINDOW_END_COLUMN,
+    WINDOW_START_COLUMN,
 };
 use common_function::function::FunctionContext;
 use common_function::function_registry::FUNCTION_REGISTRY;
@@ -60,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::{Expr, LogicalPlan, ScalarUDF, cast, ident, lit};
+use datafusion_expr::{Case, Expr, LogicalPlan, ScalarUDF, cast, ident, lit};
 pub use relationships::{
     CallsSource, CoDeclaredSource, DeclaredSource, RelationshipSources, build_relationships_plan,
 };
@@ -122,9 +123,10 @@ pub fn declared_relationships_schema_matches(table_info: &table::metadata::Table
         })
 }
 
-/// Bin width for the temporal window of derived rows: 60s buckets, matching the
-/// service-graph convention.
-const BIN_NANOS: i64 = 60 * 1_000_000_000;
+/// Bin width for the temporal window of derived rows, matching the
+/// service-graph convention. Shared so ingestion-synthesized observations
+/// land in the same buckets.
+const BIN_NANOS: i64 = SEMANTIC_GRAPH_WINDOW_NANOS;
 
 /// Default retention for the declared-edge table; expiry slides the topology window.
 const DEFAULT_DECLARED_RELATIONSHIPS_TTL: &str = "90d";
@@ -284,8 +286,12 @@ pub struct EntityDeclaration {
     /// The table's time index column, used for the temporal window filter.
     pub time_index: String,
     pub entity_type: String,
-    /// Identifying columns (>= 1). One column → id verbatim; several → composite.
+    /// Identifying columns (>= 1), ordered broad to narrow.
     pub id_columns: Vec<String>,
+    /// Optional column qualifying the first identity component, so a source
+    /// 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>,
     /// Descriptive columns snapshotted into the `descriptive` JSON (may be empty).
     pub descriptive_columns: Vec<String>,
     /// Scope columns (namespace/environment). One column → scope verbatim;
@@ -417,28 +423,91 @@ fn cast_string_or_empty(column: &str) -> Expr {
     core_fns::coalesce().call(vec![cast(ident(column), DataType::Utf8), lit("")])
 }
 
-/// The canonical entity-id expression for `id_columns`: the value verbatim for
-/// a single column, the sorted `k=v,k=v` rendering for a composite. `col`
-/// constructs the column reference (unqualified for registry branches,
-/// join-side-qualified for the calls derivation).
 /// A row identifies an entity only when every identity component is present
 /// and non-empty: kube-state-metrics descriptors emit empty-string labels (an
 /// unscheduled pod's `node`, an owner-less pod's `owner_*`), and an empty
-/// string is never a meaningful entity id.
+/// 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 {
     ident(column)
         .is_not_null()
         .and(cast(ident(column), DataType::Utf8).not_eq(lit("")))
 }
 
-fn entity_id_expr(id_columns: &[String], col: &dyn Fn(&str) -> Expr) -> Expr {
-    if let [id] = id_columns {
-        cast(col(id), DataType::Utf8)
-    } else {
-        let mut cols = id_columns.to_vec();
-        cols.sort();
-        sorted_kv_expr_with(&cols, false, col)
+const ID_SEPARATOR: &str = ",";
+const ID_ESCAPE: &str = "\\";
+/// Left unescaped: `<namespace>/<name>` is how Prometheus renders `job`.
+const ID_QUALIFIER_SEPARATOR: &str = "/";
+
+/// Escapes so a composite id decodes back to its components. The escape
+/// character goes first, or it would double the escapes the separator pass
+/// introduces.
+fn escaped_id_value(value: Expr) -> Expr {
+    let escape_escape = string_fns::replace().call(vec![
+        value,
+        lit(ID_ESCAPE),
+        lit(format!("{ID_ESCAPE}{ID_ESCAPE}")),
+    ]);
+    string_fns::replace().call(vec![
+        escape_escape,
+        lit(ID_SEPARATOR),
+        lit(format!("{ID_ESCAPE}{ID_SEPARATOR}")),
+    ])
+}
+
+/// The identity values in declared order (broad to narrow), escaped and
+/// joined.
+///
+/// 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.
+///
+/// `col` constructs the column reference (unqualified for registry branches,
+/// join-side-qualified for the calls derivation).
+fn entity_id_expr(
+    id_columns: &[String],
+    qualifier: Option<&str>,
+    col: &dyn Fn(&str) -> Expr,
+) -> Expr {
+    let mut parts = Vec::with_capacity(id_columns.len() * 2);
+    for (index, column) in id_columns.iter().enumerate() {
+        if index > 0 {
+            parts.push(lit(ID_SEPARATOR));
+        }
+        let value = escaped_id_value(cast(col(column), DataType::Utf8));
+        parts.push(match qualifier {
+            Some(qualifier) if index == 0 => qualified_id_expr(qualifier, value, col),
+            _ => value,
+        });
     }
+    if let [single] = parts.as_slice() {
+        single.clone()
+    } else {
+        concat_expr(parts)
+    }
+}
+
+/// `<qualifier>/<value>`, or bare `value` when the qualifier is empty on this
+/// row — the spec's rule composing `job` from `service.namespace` and
+/// `service.name`.
+fn qualified_id_expr(qualifier: &str, value: Expr, col: &dyn Fn(&str) -> Expr) -> Expr {
+    let qualifier = escaped_id_value(
+        core_fns::coalesce().call(vec![cast(col(qualifier), DataType::Utf8), lit("")]),
+    );
+    Expr::Case(Case::new(
+        None,
+        vec![(
+            Box::new(qualifier.clone().eq(lit(""))),
+            Box::new(value.clone()),
+        )],
+        Some(Box::new(concat_expr(vec![
+            qualifier,
+            lit(ID_QUALIFIER_SEPARATOR),
+            value,
+        ]))),
+    ))
 }
 
 /// The `parse_json` UDF, shared by all derivation plans. Resolved from the
@@ -583,14 +652,22 @@ 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. Composite ids
-        // additionally carry a JSON object of the id columns in
-        // entity_id_attrs.
-        let entity_id = entity_id_expr(&decl.id_columns, &|c| ident(c));
-        let entity_id_attrs = if decl.id_columns.len() == 1 {
+        // 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.
+        let entity_id = entity_id_expr(&decl.id_columns, decl.id_qualifier.as_deref(), &|c| {
+            ident(c)
+        });
+        let id_parts = decl
+            .id_qualifier
+            .iter()
+            .chain(&decl.id_columns)
+            .cloned()
+            .collect::<Vec<_>>();
+        let entity_id_attrs = if id_parts.len() == 1 {
             null_json()
         } else {
-            json_object_expr(&decl.id_columns)
+            json_object_expr(&id_parts)
         };
 
         let scope = match decl.scope_columns.as_slice() {
@@ -811,11 +888,74 @@ mod tests {
             time_index: "ts".to_string(),
             entity_type: entity_type.to_string(),
             id_columns: id_columns.iter().map(|s| s.to_string()).collect(),
+            id_qualifier: None,
             descriptive_columns: vec![],
             scope_columns: vec![],
         }
     }
 
+    /// The id string must decode back to its components, and a qualifier must
+    /// reproduce the identity a source that pre-composed it already emits.
+    #[tokio::test]
+    async fn registry_id_escaping_and_qualifier() {
+        let schema = Arc::new(Schema::new(vec![
+            Field::new(
+                "ts",
+                DataType::Timestamp(TimeUnit::Millisecond, None),
+                false,
+            ),
+            Field::new("service_name", DataType::Utf8, false),
+            Field::new("instance", DataType::Utf8, false),
+            Field::new("namespace", DataType::Utf8, true),
+        ]));
+        let batch = RecordBatch::try_new(
+            schema.clone(),
+            vec![
+                Arc::new(TimestampMillisecondArray::from(vec![1_000, 2_000, 3_000])) as ArrayRef,
+                Arc::new(StringArray::from(vec!["cart", "a,b", "we\\ird"])),
+                Arc::new(StringArray::from(vec!["i-1", "c", "i-3"])),
+                Arc::new(StringArray::from(vec![Some("shop"), None, Some("")])),
+            ],
+        )
+        .unwrap();
+        let ctx = SessionContext::new();
+        ctx.register_table(
+            "svc",
+            Arc::new(MemTable::try_new(schema, vec![vec![batch]]).unwrap()),
+        )
+        .unwrap();
+
+        let mut declaration = decl("service.instance", &["service_name", "instance"]);
+        declaration.id_qualifier = Some("namespace".to_string());
+        declaration.time_index = "ts".to_string();
+        let plan = build_registry_plan(
+            vec![RegistrySource {
+                declarations: vec![declaration],
+                scan: ctx.table("svc").await.unwrap(),
+            }],
+            &test_window(),
+        )
+        .unwrap()
+        .unwrap();
+
+        let mut ids: Vec<String> = collect(&ctx, plan)
+            .await
+            .iter()
+            .flat_map(|batch| strings(batch, 5))
+            .collect();
+        ids.sort();
+        assert_eq!(
+            ids,
+            vec![
+                // an absent qualifier leaves the identity bare, and a value
+                // holding the separator stays distinguishable from two values
+                "a\\,b,c".to_string(),
+                "shop/cart,i-1".to_string(),
+                "we\\\\ird,i-3".to_string(),
+            ]
+        );
+    }
+
     #[tokio::test]
     async fn registry_single_column_identity() {
         let ctx = metric_table_ctx();
@@ -900,24 +1040,24 @@ mod tests {
             .collect();
         rows.sort();
 
-        // Composite id -> sorted `k=v,k=v` plus a JSON object of the id columns;
-        // descriptive JSON keeps `\`, `"` and control characters intact in
-        // runtime values, NULL -> "".
+        // Composite id -> values in declared order plus a JSON object of the
+        // id columns; descriptive JSON keeps `\`, `"` and control characters
+        // intact in runtime values, NULL -> "".
         assert_eq!(
             rows,
             vec![
                 (
-                    "pid=42,service_name=cart".to_string(),
+                    "cart,42".to_string(),
                     Some(r#"{"pid":"42","service_name":"cart"}"#.to_string()),
                     Some(r#"{"host":""}"#.to_string()),
                 ),
                 (
-                    "pid=42,service_name=cart".to_string(),
+                    "cart,42".to_string(),
                     Some(r#"{"pid":"42","service_name":"cart"}"#.to_string()),
                     Some(r#"{"host":"h2"}"#.to_string()),
                 ),
                 (
-                    "pid=42,service_name=cart".to_string(),
+                    "cart,42".to_string(),
                     Some(r#"{"pid":"42","service_name":"cart"}"#.to_string()),
                     Some(r#"{"host":"we\"ird\\\nhost"}"#.to_string()),
                 ),

src/operator/src/statement/semantic_graph/conventions.rs

@@ -19,7 +19,7 @@
 //! with the binary, not an operator-editable configuration surface; explicit
 //! `greptime.semantic.entity.*` declarations always override it.
 
-use std::collections::{BTreeMap, HashSet};
+use std::collections::{BTreeMap, HashMap, HashSet};
 use std::sync::LazyLock;
 
 use serde::Deserialize;
@@ -43,15 +43,20 @@ pub struct VirtualDstCandidate {
     pub connection_type: String,
 }
 
-/// One implicit entity declaration: of a whitelisted Prometheus info metric,
-/// or of a trace-v1 table's flattened resource attributes.
+/// One implicit entity declaration: of a whitelisted Prometheus or OTel info
+/// metric, or of a trace-v1 table's flattened resource attributes.
 #[derive(Debug, Deserialize)]
 #[serde(deny_unknown_fields)]
 pub struct ImplicitEntity {
     pub entity: String,
-    /// Identifying label columns; every one must exist on the table for the
-    /// declaration to apply.
+    /// Identifying label columns, ordered broad to narrow; every one must
+    /// exist on the table for the declaration to apply.
     pub id: Vec<String>,
+    /// Column qualifying `id[0]` as `<qualifier>/<id[0]>`, so a source
+    /// carrying the parts separately matches one carrying them pre-composed.
+    /// Skipped when the column is absent or empty.
+    #[serde(default)]
+    pub qualified_by: Option<String>,
     /// Descriptive label columns, filtered to those present (kube-state-metrics
     /// label sets vary across versions).
     #[serde(default)]
@@ -70,7 +75,12 @@ pub struct Conventions {
     pub trace_co_declared_edges: Vec<EdgeRule>,
     pub virtual_dst_candidates: Vec<VirtualDstCandidate>,
     pub otlp_trace_entities: Vec<ImplicitEntity>,
+    /// Table name -> the entities that table declares, for Prometheus-sourced
+    /// descriptor metrics (`source = prometheus`).
     pub prometheus_info_metrics: BTreeMap<String, Vec<ImplicitEntity>>,
+    /// Table name -> the entities that table declares, for OTLP-sourced
+    /// descriptor tables (`source = opentelemetry`).
+    pub otel_info_metrics: BTreeMap<String, Vec<ImplicitEntity>>,
 }
 
 /// The built-in entity-type vocabulary. User-declared types are open-ended;
@@ -78,6 +88,7 @@ pub struct Conventions {
 pub const ENTITY_TYPE_SERVICE: &str = "service";
 pub const ENTITY_TYPE_SERVICE_INSTANCE: &str = "service.instance";
 pub const ENTITY_TYPE_HOST: &str = "host";
+pub const ENTITY_TYPE_CONTAINER: &str = "container";
 pub const ENTITY_TYPE_PROCESS: &str = "process";
 pub const ENTITY_TYPE_K8S_POD: &str = "k8s.pod";
 pub const ENTITY_TYPE_K8S_NODE: &str = "k8s.node";
@@ -88,10 +99,11 @@ pub const ENTITY_TYPE_GEN_AI_AGENT: &str = "gen_ai.agent";
 pub const ENTITY_TYPE_GEN_AI_MODEL: &str = "gen_ai.model";
 pub const ENTITY_TYPE_GEN_AI_TOOL: &str = "gen_ai.tool";
 
-const ENTITY_TYPES: [&str; 12] = [
+const ENTITY_TYPES: [&str; 13] = [
     ENTITY_TYPE_SERVICE,
     ENTITY_TYPE_SERVICE_INSTANCE,
     ENTITY_TYPE_HOST,
+    ENTITY_TYPE_CONTAINER,
     ENTITY_TYPE_PROCESS,
     ENTITY_TYPE_K8S_POD,
     ENTITY_TYPE_K8S_NODE,
@@ -200,11 +212,16 @@ fn validate(conventions: &Conventions) -> Result<(), String> {
     let per_table = conventions
         .prometheus_info_metrics
         .iter()
+        .chain(&conventions.otel_info_metrics)
         .map(|(table, entities)| (table.as_str(), entities))
         .chain(std::iter::once((
             "otlp traces",
             &conventions.otlp_trace_entities,
         )));
+    // An entity type declared with a different number of id columns by two
+    // sources yields ids that can never match, silently splitting one entity
+    // in two.
+    let mut arity = HashMap::new();
     for (table, entities) in per_table {
         let mut seen_types = HashSet::new();
         for implicit in entities {
@@ -226,6 +243,12 @@ fn validate(conventions: &Conventions) -> Result<(), String> {
                     implicit.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",
+                    implicit.entity
+                ));
+            }
             if implicit.descriptive_rest && !implicit.descriptive.is_empty() {
                 return Err(format!(
                     "entity `{}` of info metric `{table}` sets both descriptive and \
@@ -233,6 +256,17 @@ fn validate(conventions: &Conventions) -> Result<(), String> {
                     implicit.entity
                 ));
             }
+            match arity.insert(implicit.entity.as_str(), implicit.id.len()) {
+                Some(previous) if previous != implicit.id.len() => {
+                    return Err(format!(
+                        "entity `{}` is declared with {previous} id columns elsewhere but \
+                         {} for `{table}`",
+                        implicit.entity,
+                        implicit.id.len()
+                    ));
+                }
+                _ => {}
+            }
         }
     }
     Ok(())
@@ -249,25 +283,28 @@ mod tests {
 
     /// Each case must fail on exactly the rule it names, so the shared
     /// boilerplate is valid and the mutated part is inside the vocabulary.
-    fn broken(edges: &str, info_metrics: &str) -> String {
+    fn broken(edges: &str, info_metrics: &str, otel_metrics: &str) -> String {
         format!(
             "co_declared_edges: [{edges}]\ntrace_co_declared_edges: []\n\
              virtual_dst_candidates: []\notlp_trace_entities: []\n\
-             prometheus_info_metrics: {{{info_metrics}}}"
+             prometheus_info_metrics: {{{info_metrics}}}\n\
+             otel_info_metrics: {{{otel_metrics}}}"
         )
     }
 
     #[test]
     fn validation_rejects_broken_conventions() {
-        let err = |edges, info| parse(&broken(edges, info)).unwrap_err();
+        let err = |edges, info, otel| parse(&broken(edges, info, otel)).unwrap_err();
 
-        assert!(err("{src: host, dst: service, rel: pets}", "").contains("unknown rel_type"));
+        assert!(err("{src: host, dst: service, rel: pets}", "", "").contains("unknown rel_type"));
         assert!(
-            err("{src: k8s.pods, dst: k8s.node, rel: runs_on}", "").contains("unknown entity type")
+            err("{src: k8s.pods, dst: k8s.node, rel: runs_on}", "", "")
+                .contains("unknown entity type")
         );
         assert!(
             err(
                 "{src: host, dst: service, rel: uses}, {src: host, dst: service, rel: uses}",
+                "",
                 ""
             )
             .contains("duplicate edge rule")
@@ -275,14 +312,27 @@ mod tests {
         assert!(
             err(
                 "",
-                "t: [{entity: host, id: [x], descriptive: [y], descriptive_rest: true}]"
+                "t: [{entity: host, id: [x], descriptive: [y], descriptive_rest: true}]",
+                ""
             )
             .contains("descriptive_rest")
         );
+        // 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
+        assert!(
+            err(
+                "",
+                "a: [{entity: host, id: [x]}]",
+                "b: [{entity: host, id: [x, y]}]"
+            )
+            .contains("id columns")
+        );
         // Unknown YAML keys are rejected, catching typos in the embedded file.
         assert!(
             parse(&broken(
                 "{src: host, dst: service, rel: uses, direction: down}",
+                "",
                 ""
             ))
             .is_err()

src/operator/src/statement/semantic_graph/conventions.yaml

@@ -2,6 +2,10 @@
 # versioned with the binary (include_str!), not an operator-editable surface;
 # non-standard deployments override per table with explicit
 # greptime.semantic.entity.* declarations, which always win.
+#
+# Every `id` list is ordered broad to narrow, and that order is part of the
+# identity: the id renders as the values joined in this order, so all
+# declarations of one entity type must agree on it or they name two entities.
 
 co_declared_edges:
   - { src: service.instance, dst: host, rel: runs_on }
@@ -11,6 +15,8 @@ co_declared_edges:
   - { src: service.instance, dst: service, rel: part_of }
   - { src: k8s.pod, dst: k8s.workload, rel: part_of }
   - { src: service.instance, dst: k8s.pod, rel: runs_on }
+  - { src: service.instance, dst: container, rel: runs_on }
+  - { src: container, dst: host, rel: runs_on }
 
 # Applied only to trace sources; the endpoints are Greptime entity types
 # derived from GenAI semantic-convention attributes.
@@ -30,12 +36,25 @@ virtual_dst_candidates:
   - { column: span_attributes.server.address, connection_type: virtual_node }
 
 # Implicit declarations for greptime_trace_v1 tables; the pod UID bridges
-# trace-side pods onto the kube-state-metrics entities below.
+# trace-side pods onto the kube-state-metrics entities below. host.name and
+# container.name vary by SDK and resource detector, so identity is the id.
+# qualified_by folds the namespace into the service identity the same way the
+# metric sources' `job` already carries it, so both signals name one entity.
 otlp_trace_entities:
   - entity: service
     id: [service_name]
+    qualified_by: resource_attributes.service.namespace
   - entity: service.instance
     id: [service_name, resource_attributes.service.instance.id]
+    qualified_by: resource_attributes.service.namespace
+  - entity: host
+    id: [resource_attributes.host.id]
+    descriptive:
+      - resource_attributes.host.name
+  - entity: container
+    id: [resource_attributes.container.id]
+    descriptive:
+      - resource_attributes.container.name
   - entity: k8s.pod
     id: [resource_attributes.k8s.pod.uid]
     descriptive:
@@ -113,3 +132,32 @@ prometheus_info_metrics:
     - entity: service.instance
       id: [job, instance]
       descriptive_rest: true
+
+# Implicit declarations for the OTLP resource descriptor synthesized at
+# metrics ingestion, gated on source=opentelemetry and metric.type=info.
+# job/instance are the derived Prometheus-compatible identity, so services
+# line up with target_info. Unlike target_info the attribute set is known, so
+# descriptive columns are listed instead of swept up by descriptive_rest,
+# which would misfile other entities' ids under service.instance.
+otel_info_metrics:
+  greptime_otel_resource_info:
+    - entity: service
+      id: [job]
+      descriptive:
+        - service.name
+        - service.namespace
+    - entity: service.instance
+      id: [job, instance]
+    - entity: host
+      id: [host.id]
+      descriptive:
+        - host.name
+    - entity: container
+      id: [container.id]
+      descriptive:
+        - container.name
+    - entity: k8s.pod
+      id: [k8s.pod.uid]
+      descriptive:
+        - k8s.pod.name
+        - k8s.namespace.name

src/operator/src/statement/semantic_graph/relationships.rs

@@ -200,9 +200,9 @@ fn co_declared_branch(
                     bin.clone() + bin_interval(),
                     bin.clone() + bin_interval(),
                     lit(src.entity_type.as_str()),
-                    entity_id_expr(&src.id_columns, &|c| ident(c)),
+                    entity_id_expr(&src.id_columns, src.id_qualifier.as_deref(), &|c| ident(c)),
                     lit(dst.entity_type.as_str()),
-                    entity_id_expr(&dst.id_columns, &|c| ident(c)),
+                    entity_id_expr(&dst.id_columns, dst.id_qualifier.as_deref(), &|c| ident(c)),
                     lit(*rel_type),
                     lit(*provenance),
                     lit(1.0_f64),
@@ -501,6 +501,13 @@ fn span_predicate(service: &EntityDeclaration, window: &GraphQueryWindow, strict
 }
 
 /// One trace table's client spans, normalized to
+/// A trace table's `duration_nano` is UInt64 on tables created before the
+/// signed-integer ingest change and Int64 after it. The per-table selects are
+/// unioned, and those two have no common integer type.
+fn duration_nano_expr() -> Expr {
+    cast(ident(DURATION_NANO_COLUMN), DataType::Int64).alias(DURATION_NANO_COLUMN)
+}
+
 /// `(timestamp, trace_id, span_id, src_id, status_code, duration_nano,
 /// virtual_dst, virtual_conn)`. `src_id` is built from the table's `service`
 /// declaration, so edges land on exactly the entity ids the registry emits (a
@@ -553,9 +560,12 @@ fn client_spans(
             ident(SPAN_ID_COLUMN),
             // The cast inside entity_id_expr also normalizes tag columns, which
             // come out of the storage engine dictionary-encoded.
-            entity_id_expr(&service.id_columns, &|c| ident(c)).alias(SRC_ID_COLUMN),
+            entity_id_expr(&service.id_columns, service.id_qualifier.as_deref(), &|c| {
+                ident(c)
+            })
+            .alias(SRC_ID_COLUMN),
             ident(SPAN_STATUS_CODE_COLUMN),
-            ident(DURATION_NANO_COLUMN),
+            duration_nano_expr(),
             virtual_dst.alias("virtual_dst"),
             virtual_conn.alias("virtual_conn"),
         ])
@@ -578,9 +588,12 @@ fn server_spans(
             ident(TRACE_TIMESTAMP_COLUMN),
             ident(TRACE_ID_COLUMN),
             ident(PARENT_SPAN_ID_COLUMN),
-            entity_id_expr(&service.id_columns, &|c| ident(c)).alias(DST_ID_COLUMN),
+            entity_id_expr(&service.id_columns, service.id_qualifier.as_deref(), &|c| {
+                ident(c)
+            })
+            .alias(DST_ID_COLUMN),
             ident(SPAN_STATUS_CODE_COLUMN),
-            ident(DURATION_NANO_COLUMN),
+            duration_nano_expr(),
         ])
 }
 
@@ -609,7 +622,10 @@ fn agent_calls_branch(
                 ident(TRACE_TIMESTAMP_COLUMN),
                 ident(TRACE_ID_COLUMN),
                 ident(SPAN_ID_COLUMN),
-                entity_id_expr(&agent.id_columns, &|c| ident(c)).alias(SRC_ID_COLUMN),
+                entity_id_expr(&agent.id_columns, agent.id_qualifier.as_deref(), &|c| {
+                    ident(c)
+                })
+                .alias(SRC_ID_COLUMN),
             ])?;
         let child = trace
             .scan
@@ -619,9 +635,12 @@ fn agent_calls_branch(
                 ident(TRACE_TIMESTAMP_COLUMN),
                 ident(TRACE_ID_COLUMN),
                 ident(PARENT_SPAN_ID_COLUMN),
-                entity_id_expr(&agent.id_columns, &|c| ident(c)).alias(DST_ID_COLUMN),
+                entity_id_expr(&agent.id_columns, agent.id_qualifier.as_deref(), &|c| {
+                    ident(c)
+                })
+                .alias(DST_ID_COLUMN),
                 ident(SPAN_STATUS_CODE_COLUMN),
-                ident(DURATION_NANO_COLUMN),
+                duration_nano_expr(),
             ])?;
         parents = union_all(parents, parent)?;
         children = union_all(children, child)?;
@@ -732,6 +751,16 @@ mod tests {
         name: &str,
         extra: &[(&str, &[Option<&str>])],
         spans: &[Span<'_>],
+    ) {
+        register_typed_trace_table(ctx, name, extra, spans, DataType::UInt64)
+    }
+
+    fn register_typed_trace_table(
+        ctx: &SessionContext,
+        name: &str,
+        extra: &[(&str, &[Option<&str>])],
+        spans: &[Span<'_>],
+        duration_type: DataType,
     ) {
         let mut fields = vec![
             Field::new(
@@ -745,7 +774,7 @@ mod tests {
             Field::new(SPAN_KIND_COLUMN, DataType::Utf8, false),
             Field::new(SPAN_STATUS_CODE_COLUMN, DataType::Utf8, false),
             Field::new(SERVICE_NAME_COLUMN, DataType::Utf8, false),
-            Field::new(DURATION_NANO_COLUMN, DataType::UInt64, false),
+            Field::new(DURATION_NANO_COLUMN, duration_type.clone(), false),
         ];
         for (column, _) in extra {
             fields.push(Field::new(*column, DataType::Utf8, true));
@@ -774,9 +803,14 @@ mod tests {
             Arc::new(StringArray::from(
                 spans.iter().map(|s| s.6).collect::<Vec<_>>(),
             )),
-            Arc::new(UInt64Array::from(
-                spans.iter().map(|s| s.7).collect::<Vec<_>>(),
-            )),
+            match duration_type {
+                DataType::Int64 => Arc::new(Int64Array::from(
+                    spans.iter().map(|s| s.7 as i64).collect::<Vec<_>>(),
+                )) as ArrayRef,
+                _ => Arc::new(UInt64Array::from(
+                    spans.iter().map(|s| s.7).collect::<Vec<_>>(),
+                )),
+            },
         ];
         for (_, values) in extra {
             columns.push(Arc::new(StringArray::from(values.to_vec())));
@@ -843,6 +877,7 @@ mod tests {
             time_index: TRACE_TIMESTAMP_COLUMN.to_string(),
             entity_type: "service".to_string(),
             id_columns: id_columns.iter().map(|s| s.to_string()).collect(),
+            id_qualifier: None,
             descriptive_columns: vec![],
             scope_columns: vec![],
         }
@@ -972,7 +1007,7 @@ mod tests {
                 ),
             ],
         );
-        register_trace_table(
+        register_typed_trace_table(
             &ctx,
             "trace_b",
             &[],
@@ -989,6 +1024,7 @@ mod tests {
                     1_500_000_000,
                 ),
             ],
+            DataType::Int64,
         );
         let a = ctx.table("trace_a").await.unwrap();
         let b = ctx.table("trace_b").await.unwrap();
@@ -1294,6 +1330,7 @@ mod tests {
             time_index: "ts".to_string(),
             entity_type: entity_type.to_string(),
             id_columns: id_columns.iter().map(|s| s.to_string()).collect(),
+            id_qualifier: None,
             descriptive_columns: vec![],
             scope_columns: vec![],
         }
@@ -1544,16 +1581,10 @@ mod tests {
         let total: usize = batches.iter().map(|b| b.num_rows()).sum();
         assert_eq!(total, 1);
         let batch = &batches[0];
-        // Composite service identity renders the same sorted `k=v` form the
-        // registry emits, so edges land on registry entity ids.
-        assert_eq!(
-            strings(batch, 5),
-            vec!["service_name=frontend,service_namespace=ns1"]
-        );
-        assert_eq!(
-            strings(batch, 7),
-            vec!["service_name=cart,service_namespace=ns1"]
-        );
+        // Composite service identity renders exactly as the registry emits it,
+        // so edges land on registry entity ids.
+        assert_eq!(strings(batch, 5), vec!["frontend,ns1"]);
+        assert_eq!(strings(batch, 7), vec!["cart,ns1"]);
     }
 
     fn build_relationships_plan_for_test(

src/servers/src/http/otlp.rs

@@ -118,8 +118,9 @@ pub async fn metrics(
         promote_scope_attrs: http_opts.promote_scope_attrs,
         with_metric_engine,
         experimental_enable_exponential_histogram,
-        // set is_legacy later
+        // set by the frontend from its config
         is_legacy: false,
+        resource_info: false,
         metric_type: MetricType::Init,
         metric_translation_strategy: http_opts.metric_translation_strategy,
     }));
@@ -128,7 +129,7 @@ pub async fn metrics(
     handler.metrics(request, query_ctx).await.map(|outcome| {
         if outcome.accepted_data_points == 0 && outcome.rejected_data_points > 0 {
             OtlpMetricsResponse::Failure(outcome)
-        } else if outcome.rejected_data_points > 0 {
+        } else if outcome.rejected_data_points > 0 || outcome.error_message.is_some() {
             OtlpMetricsResponse::PartialSuccess(outcome)
         } else {
             OtlpMetricsResponse::FullSuccess(outcome)

src/servers/src/otlp/metrics.rs

@@ -25,6 +25,7 @@ use common_query::native_histogram::{
 };
 use common_query::prelude::{GREPTIME_COUNT, greptime_timestamp, greptime_value};
 use common_query::prometheus::PROMETHEUS_STALE_NAN_BITS;
+use common_telemetry::warn;
 use lazy_static::lazy_static;
 use otel_arrow_rust::proto::opentelemetry::collector::metrics::v1::ExportMetricsServiceRequest;
 use otel_arrow_rust::proto::opentelemetry::common::v1::{AnyValue, KeyValue, any_value};
@@ -36,7 +37,7 @@ use table::requests::{
 };
 
 use crate::error::{self, Result};
-use crate::otlp::trace::{KEY_SERVICE_INSTANCE_ID, KEY_SERVICE_NAME};
+use crate::otlp::trace::{KEY_SERVICE_INSTANCE_ID, KEY_SERVICE_NAME, KEY_SERVICE_NAMESPACE};
 use crate::query_handler::MetricsIngestOutcome;
 use crate::row_writer::{self, MultiTableData, TableData};
 pub use crate::semantic::SemanticIndex;
@@ -45,8 +46,11 @@ use crate::semantic::{
     METRIC_TYPE_UPDOWN_COUNTER,
 };
 
+mod resource_info;
 mod translator;
 
+pub use resource_info::OTEL_RESOURCE_INFO_TABLE_NAME;
+use resource_info::ResourceInfoData;
 pub use translator::legacy_normalize_otlp_name;
 pub(crate) use translator::ucum_to_openmetrics_unit;
 use translator::{translate_label_name, translate_metric_name};
@@ -96,28 +100,43 @@ const MIN_EXPONENTIAL_HISTOGRAM_SCALE: i32 = -4;
 const MAX_EXPONENTIAL_HISTOGRAM_SCALE: i32 = 8;
 const MAX_REJECTION_MESSAGE_BYTES: usize = 512;
 
+/// Result of converting one OTLP metrics request.
+#[derive(Debug)]
+pub struct MetricsConversion {
+    pub requests: RowInsertRequests,
+    /// Row count of `requests`; the resource descriptor is not counted.
+    pub rows: usize,
+    /// Per-table semantic index for the auto-create path to stamp as table
+    /// options; covers the descriptor table too.
+    pub semantic_index: SemanticIndex,
+    /// The synthesized resource descriptor, written separately from the
+    /// metric data. See [`resource_info`].
+    pub resource_info: Option<RowInsertRequests>,
+    pub outcome: MetricsIngestOutcome,
+}
+
 /// Convert OpenTelemetry metrics to GreptimeDB insert requests
 ///
 /// See
 /// <https://github.com/open-telemetry/opentelemetry-proto/blob/main/opentelemetry/proto/metrics/v1/metrics.proto>
 /// for data structure of OTLP metrics.
-///
-/// Returns `InsertRequests`, total number of rows to ingest, and the per-table
-/// semantic index for the auto-create path to stamp as table options.
 pub fn to_grpc_insert_requests(
     request: ExportMetricsServiceRequest,
     metric_ctx: &mut OtlpMetricCtx,
-) -> Result<(
-    RowInsertRequests,
-    usize,
-    SemanticIndex,
-    MetricsIngestOutcome,
-)> {
+) -> Result<MetricsConversion> {
     let mut table_writer = MultiTableData::default();
     let mut semantic_index = SemanticIndex::default();
     let mut outcome = MetricsIngestOutcome::default();
+    let mut resource_info = ResourceInfoData::default();
 
     for resource in &request.resource_metrics {
+        if metric_ctx.resource_info
+            && !metric_ctx.is_legacy
+            && let Some(r) = resource.resource.as_ref()
+        {
+            resource_info.observe(&r.attributes, resource, metric_ctx);
+        }
+
         let resource_attrs = resource.resource.as_ref().map(|r| {
             let mut attrs = r.attributes.clone();
             process_resource_attrs(&mut attrs, metric_ctx);
@@ -149,8 +168,46 @@ pub fn to_grpc_insert_requests(
     }
 
     let (requests, rows) = table_writer.into_row_insert_requests();
+
     validate_sample_kinds(&requests)?;
-    Ok((requests, rows, semantic_index, outcome))
+
+    // The metric and the descriptor would fight over the same table across
+    // two engines. The user's metric wins.
+    let resource_info = if !metric_ctx.resource_info {
+        None
+    } else if requests
+        .inserts
+        .iter()
+        .any(|r| r.table_name == OTEL_RESOURCE_INFO_TABLE_NAME)
+    {
+        warn!(
+            "Skipping OTLP resource descriptor synthesis: the request writes \
+             a metric named `{OTEL_RESOURCE_INFO_TABLE_NAME}`"
+        );
+        None
+    } else {
+        resource_info.into_row_insert_requests()?
+    };
+    if resource_info.is_some() {
+        semantic_index.record_scalar(
+            OTEL_RESOURCE_INFO_TABLE_NAME,
+            SEMANTIC_METRIC_TYPE,
+            crate::semantic::METRIC_TYPE_INFO,
+        );
+        semantic_index.record_scalar(
+            OTEL_RESOURCE_INFO_TABLE_NAME,
+            SEMANTIC_METRIC_METADATA_QUALITY,
+            METADATA_QUALITY_DECLARED,
+        );
+    }
+
+    Ok(MetricsConversion {
+        requests,
+        rows,
+        semantic_index,
+        outcome,
+        resource_info,
+    })
 }
 
 fn validate_sample_kinds(requests: &RowInsertRequests) -> Result<()> {
@@ -287,30 +344,61 @@ fn from_metric_type(data: &metric::Data) -> MetricType {
     }
 }
 
+/// Non-scalar values (bool, arrays, maps, bytes) are not representable as tags.
+fn scalar_value_string(value: Option<&AnyValue>) -> Option<String> {
+    match value.and_then(|v| v.value.as_ref())? {
+        any_value::Value::StringValue(s) => Some(s.clone()),
+        any_value::Value::IntValue(v) => Some(v.to_string()),
+        any_value::Value::DoubleValue(v) => Some(v.to_string()),
+        _ => None,
+    }
+}
+
+/// Prometheus-style `(job, instance)` identity. Per the OTel Prometheus
+/// compatibility spec `job` folds in `service.namespace` when present; a
+/// resource without `service.name` gets no job rather than a fabricated one.
+/// Both fields are optional strings, so a named type keeps a swapped
+/// destructuring from type-checking at the call sites.
+pub(crate) struct ServiceIdentity {
+    pub job: Option<String>,
+    pub instance: Option<String>,
+}
+
+pub(crate) fn service_identity(attrs: &[KeyValue]) -> ServiceIdentity {
+    let mut name = None;
+    let mut namespace = None;
+    let mut instance = None;
+    for kv in attrs {
+        match kv.key.as_str() {
+            KEY_SERVICE_NAME => name = scalar_value_string(kv.value.as_ref()),
+            KEY_SERVICE_NAMESPACE => namespace = scalar_value_string(kv.value.as_ref()),
+            KEY_SERVICE_INSTANCE_ID => instance = scalar_value_string(kv.value.as_ref()),
+            _ => {}
+        }
+    }
+    let job = name.map(|name| match namespace {
+        Some(ns) if !ns.is_empty() => format!("{ns}/{name}"),
+        _ => name,
+    });
+    ServiceIdentity { job, instance }
+}
+
+fn string_key_value(key: &str, value: String) -> KeyValue {
+    KeyValue {
+        key: key.to_string(),
+        value: Some(AnyValue {
+            value: Some(any_value::Value::StringValue(value)),
+        }),
+    }
+}
+
 fn process_resource_attrs(attrs: &mut Vec<KeyValue>, metric_ctx: &OtlpMetricCtx) {
     if metric_ctx.is_legacy {
         return;
     }
 
-    // remap service.name and service.instance.id to job and instance
-    let mut tmp = Vec::with_capacity(2);
-    for kv in attrs.iter() {
-        match &kv.key as &str {
-            KEY_SERVICE_NAME => {
-                tmp.push(KeyValue {
-                    key: JOB_KEY.to_string(),
-                    value: kv.value.clone(),
-                });
-            }
-            KEY_SERVICE_INSTANCE_ID => {
-                tmp.push(KeyValue {
-                    key: INSTANCE_KEY.to_string(),
-                    value: kv.value.clone(),
-                });
-            }
-            _ => {}
-        }
-    }
+    // remap the service identity attributes to job and instance
+    let ServiceIdentity { job, instance } = service_identity(attrs);
 
     // if promote all, then exclude the list, else, include the list
     if metric_ctx.promote_all_resource_attrs {
@@ -322,7 +410,12 @@ fn process_resource_attrs(attrs: &mut Vec<KeyValue>, metric_ctx: &OtlpMetricCtx)
         });
     }
 
-    attrs.extend(tmp);
+    if let Some(job) = job {
+        attrs.push(string_key_value(JOB_KEY, job));
+    }
+    if let Some(instance) = instance {
+        attrs.push(string_key_value(INSTANCE_KEY, instance));
+    }
 }
 
 fn process_scope_attrs(scope: &ScopeMetrics, metric_ctx: &OtlpMetricCtx) -> Option<Vec<KeyValue>> {
@@ -536,35 +629,13 @@ fn encode_exponential_histogram(
     metric_ctx: &OtlpMetricCtx,
     outcome: &mut MetricsIngestOutcome,
 ) -> Result<bool> {
-    if !metric_ctx.experimental_enable_exponential_histogram {
+    if let Err(rejection) = exponential_histogram_gate(histogram, metric_ctx) {
         reject_data_points(outcome, histogram.data_points.len(), || {
-            format!(
-                "metric `{name}` uses OTLP exponential histograms; set otlp.experimental_enable_exponential_histogram = true to enable ingestion"
-            )
+            rejection.message(name)
         })?;
         return Ok(false);
     }
 
-    match AggregationTemporality::try_from(histogram.aggregation_temporality) {
-        Ok(AggregationTemporality::Cumulative) => {}
-        Ok(AggregationTemporality::Delta) => {
-            reject_data_points(outcome, histogram.data_points.len(), || {
-                format!(
-                    "metric `{name}` uses delta OTLP exponential histograms; only cumulative temporality is supported"
-                )
-            })?;
-            return Ok(false);
-        }
-        _ => {
-            reject_data_points(outcome, histogram.data_points.len(), || {
-                format!(
-                    "metric `{name}` has unspecified OTLP exponential histogram temporality; cumulative temporality is required"
-                )
-            })?;
-            return Ok(false);
-        }
-    }
-
     let column_schema = native_histogram_column_schema().map_err(|error| {
         error::InternalSnafu {
             err_msg: error.to_string(),
@@ -611,7 +682,46 @@ fn encode_exponential_histogram(
     Ok(emitted)
 }
 
-fn exponential_histogram_value(
+pub(crate) enum ExponentialHistogramRejection {
+    Disabled,
+    DeltaTemporality,
+    UnspecifiedTemporality,
+}
+
+impl ExponentialHistogramRejection {
+    fn message(&self, name: &str) -> String {
+        match self {
+            Self::Disabled => format!(
+                "metric `{name}` uses OTLP exponential histograms; set otlp.experimental_enable_exponential_histogram = true to enable ingestion"
+            ),
+            Self::DeltaTemporality => format!(
+                "metric `{name}` uses delta OTLP exponential histograms; only cumulative temporality is supported"
+            ),
+            Self::UnspecifiedTemporality => format!(
+                "metric `{name}` has unspecified OTLP exponential histogram temporality; cumulative temporality is required"
+            ),
+        }
+    }
+}
+
+/// Whole-metric acceptance, decided once for the encoder and for the resource
+/// descriptor, which must not describe a resource whose data was rejected.
+/// Individual points can still fail [`exponential_histogram_value`].
+pub(crate) fn exponential_histogram_gate(
+    histogram: &ExponentialHistogram,
+    metric_ctx: &OtlpMetricCtx,
+) -> std::result::Result<(), ExponentialHistogramRejection> {
+    if !metric_ctx.experimental_enable_exponential_histogram {
+        return Err(ExponentialHistogramRejection::Disabled);
+    }
+    match AggregationTemporality::try_from(histogram.aggregation_temporality) {
+        Ok(AggregationTemporality::Cumulative) => Ok(()),
+        Ok(AggregationTemporality::Delta) => Err(ExponentialHistogramRejection::DeltaTemporality),
+        _ => Err(ExponentialHistogramRejection::UnspecifiedTemporality),
+    }
+}
+
+pub(crate) fn exponential_histogram_value(
     data_point: &ExponentialHistogramDataPoint,
 ) -> std::result::Result<(ValueData, i64), String> {
     if data_point.start_time_unix_nano > data_point.time_unix_nano {
@@ -823,29 +933,21 @@ fn write_attributes(
     };
 
     let tags = attrs.iter().filter_map(|attr| {
-        attr.value
-            .as_ref()
-            .and_then(|v| v.value.as_ref())
-            .and_then(|val| {
-                let key = match attribute_type {
-                    AttributeType::Resource | AttributeType::DataPoint => {
-                        translate_label_name(&attr.key, metric_ctx.metric_translation_strategy)
-                    }
-                    AttributeType::Scope => {
-                        format!(
-                            "otel_scope_{}",
-                            translate_label_name(&attr.key, metric_ctx.metric_translation_strategy)
-                        )
-                    }
-                    AttributeType::Legacy => legacy_normalize_otlp_name(&attr.key),
-                };
-                match val {
-                    any_value::Value::StringValue(s) => Some((key, s.clone())),
-                    any_value::Value::IntValue(v) => Some((key, v.to_string())),
-                    any_value::Value::DoubleValue(v) => Some((key, v.to_string())),
-                    _ => None, // TODO(sunng87): allow different type of values
-                }
-            })
+        // TODO(sunng87): allow different type of values
+        let value = scalar_value_string(attr.value.as_ref())?;
+        let key = match attribute_type {
+            AttributeType::Resource | AttributeType::DataPoint => {
+                translate_label_name(&attr.key, metric_ctx.metric_translation_strategy)
+            }
+            AttributeType::Scope => {
+                format!(
+                    "otel_scope_{}",
+                    translate_label_name(&attr.key, metric_ctx.metric_translation_strategy)
+                )
+            }
+            AttributeType::Legacy => legacy_normalize_otlp_name(&attr.key),
+        };
+        Some((key, value))
     });
     row_writer::write_tags(writer, tags, row)?;
 
@@ -1295,6 +1397,169 @@ mod tests {
         }
     }
 
+    fn descriptor_ctx() -> OtlpMetricCtx {
+        OtlpMetricCtx {
+            resource_info: true,
+            ..Default::default()
+        }
+    }
+
+    fn attr_value(attrs: &[KeyValue], key: &str) -> Option<String> {
+        attrs
+            .iter()
+            .find(|kv| kv.key == key)
+            .and_then(|kv| scalar_value_string(kv.value.as_ref()))
+    }
+
+    fn gauge_request(
+        resource_attrs: Vec<KeyValue>,
+        metric_name: &str,
+    ) -> ExportMetricsServiceRequest {
+        use otel_arrow_rust::proto::opentelemetry::resource::v1::Resource;
+        ExportMetricsServiceRequest {
+            resource_metrics: vec![ResourceMetrics {
+                resource: Some(Resource {
+                    attributes: resource_attrs,
+                    ..Default::default()
+                }),
+                scope_metrics: vec![ScopeMetrics {
+                    metrics: vec![Metric {
+                        name: metric_name.to_string(),
+                        data: Some(metric::Data::Gauge(Gauge {
+                            data_points: vec![NumberDataPoint {
+                                time_unix_nano: 1_000_000,
+                                value: Some(Value::AsInt(1)),
+                                ..Default::default()
+                            }],
+                        })),
+                        ..Default::default()
+                    }],
+                    ..Default::default()
+                }],
+                ..Default::default()
+            }],
+        }
+    }
+
+    fn column_names(request: &RowInsertRequests, table: &str) -> Vec<String> {
+        request
+            .inserts
+            .iter()
+            .find(|r| r.table_name == table)
+            .unwrap_or_else(|| panic!("missing table {table}"))
+            .rows
+            .as_ref()
+            .unwrap()
+            .schema
+            .iter()
+            .map(|c| c.column_name.clone())
+            .collect()
+    }
+
+    #[test]
+    fn test_conversion_synthesizes_resource_descriptor() {
+        set_default_prefix(None).unwrap();
+        let request = gauge_request(
+            vec![keyvalue("service.name", "api"), keyvalue("host.id", "h-1")],
+            "my_gauge",
+        );
+        let conversion = to_grpc_insert_requests(request, &mut descriptor_ctx()).unwrap();
+
+        // descriptor keeps raw OTel keys while the metric table's labels went
+        // through the (default underscore-escaping) translation strategy
+        let resource_info = conversion.resource_info.expect("descriptor synthesized");
+        let descriptor_cols = column_names(&resource_info, OTEL_RESOURCE_INFO_TABLE_NAME);
+        assert!(descriptor_cols.contains(&"host.id".to_string()));
+        assert!(descriptor_cols.contains(&"service.name".to_string()));
+        assert!(descriptor_cols.contains(&"job".to_string()));
+        let metric_cols = column_names(&conversion.requests, "my_gauge");
+        assert!(metric_cols.contains(&"service_name".to_string()));
+        assert!(!metric_cols.contains(&"service.name".to_string()));
+        // host.id is not in the promote list: only the descriptor keeps it
+        assert!(!metric_cols.contains(&"host_id".to_string()));
+
+        let decoded = decode(&conversion.semantic_index);
+        let t = &decoded[OTEL_RESOURCE_INFO_TABLE_NAME];
+        assert_eq!(
+            t.get(SEMANTIC_METRIC_TYPE).map(String::as_str),
+            Some("info")
+        );
+        assert_eq!(
+            t.get(SEMANTIC_METRIC_METADATA_QUALITY).map(String::as_str),
+            Some("declared")
+        );
+    }
+
+    #[test]
+    fn test_conversion_skips_descriptor_for_legacy_mode() {
+        set_default_prefix(None).unwrap();
+        let request = gauge_request(
+            vec![keyvalue("service.name", "api"), keyvalue("host.id", "h-1")],
+            "my_gauge",
+        );
+        let mut ctx = OtlpMetricCtx {
+            is_legacy: true,
+            ..descriptor_ctx()
+        };
+        let conversion = to_grpc_insert_requests(request, &mut ctx).unwrap();
+        assert!(conversion.resource_info.is_none());
+
+        // legacy tables predate job/instance and the promote filter; adding
+        // either would alter the schema of tables already in use
+        let cols = column_names(&conversion.requests, "my_gauge");
+        assert!(!cols.contains(&"job".to_string()));
+        assert!(cols.contains(&"host_id".to_string()));
+    }
+
+    #[test]
+    fn test_conversion_skips_descriptor_on_metric_name_collision() {
+        set_default_prefix(None).unwrap();
+        let request = gauge_request(
+            vec![keyvalue("service.name", "api")],
+            OTEL_RESOURCE_INFO_TABLE_NAME,
+        );
+        let conversion = to_grpc_insert_requests(request, &mut descriptor_ctx()).unwrap();
+        assert!(conversion.resource_info.is_none());
+        // the metric itself still goes through the main path
+        assert!(
+            conversion
+                .requests
+                .inserts
+                .iter()
+                .any(|r| r.table_name == OTEL_RESOURCE_INFO_TABLE_NAME)
+        );
+    }
+
+    #[test]
+    fn test_job_composition_follows_service_namespace() {
+        let mut attrs = vec![
+            keyvalue("service.name", "api"),
+            keyvalue("service.namespace", "shop"),
+        ];
+        process_resource_attrs(&mut attrs, &OtlpMetricCtx::default());
+        assert_eq!(attr_value(&attrs, "job").as_deref(), Some("shop/api"));
+
+        let mut attrs = vec![keyvalue("service.name", "api")];
+        process_resource_attrs(&mut attrs, &OtlpMetricCtx::default());
+        assert_eq!(attr_value(&attrs, "job").as_deref(), Some("api"));
+
+        let mut attrs = vec![
+            keyvalue("service.name", "api"),
+            keyvalue("service.namespace", ""),
+        ];
+        process_resource_attrs(&mut attrs, &OtlpMetricCtx::default());
+        assert_eq!(attr_value(&attrs, "job").as_deref(), Some("api"));
+
+        // no service.name: no job is fabricated
+        let mut attrs = vec![
+            keyvalue("service.namespace", "shop"),
+            keyvalue("service.instance.id", "inst-1"),
+        ];
+        process_resource_attrs(&mut attrs, &OtlpMetricCtx::default());
+        assert_eq!(attr_value(&attrs, "job"), None);
+        assert_eq!(attr_value(&attrs, "instance").as_deref(), Some("inst-1"));
+    }
+
     #[test]
     fn test_encode_gauge() {
         let mut tables = MultiTableData::default();
@@ -2104,8 +2369,12 @@ mod tests {
                 AggregationTemporality::Cumulative,
             ),
         ]);
-        let (requests, _, semantic_index, outcome) =
-            to_grpc_insert_requests(request, &mut OtlpMetricCtx::default()).unwrap();
+        let MetricsConversion {
+            requests,
+            semantic_index,
+            outcome,
+            ..
+        } = to_grpc_insert_requests(request, &mut OtlpMetricCtx::default()).unwrap();
 
         assert_eq!(outcome.accepted_data_points, 1);
         assert_eq!(outcome.rejected_data_points, 1);
@@ -2127,8 +2396,9 @@ mod tests {
             vec![],
             AggregationTemporality::Cumulative,
         )]);
-        let (_, _, _, outcome) =
-            to_grpc_insert_requests(empty, &mut OtlpMetricCtx::default()).unwrap();
+        let outcome = to_grpc_insert_requests(empty, &mut OtlpMetricCtx::default())
+            .unwrap()
+            .outcome;
         assert_eq!(outcome.rejected_data_points, 0);
         assert_eq!(outcome.error_message, None);
     }
@@ -2208,8 +2478,12 @@ mod tests {
                 .collect(),
         };
 
-        let (requests, rows, _, outcome) =
-            to_grpc_insert_requests(request, &mut OtlpMetricCtx::default()).unwrap();
+        let MetricsConversion {
+            requests,
+            rows,
+            outcome,
+            ..
+        } = to_grpc_insert_requests(request, &mut OtlpMetricCtx::default()).unwrap();
 
         assert_eq!(outcome.accepted_data_points, 2);
         assert_eq!(rows, 6);
@@ -2243,8 +2517,13 @@ mod tests {
                 experimental_enable_exponential_histogram: true,
                 ..Default::default()
             };
-            let (requests, rows, semantic_index, outcome) =
-                to_grpc_insert_requests(request, &mut ctx).unwrap();
+            let MetricsConversion {
+                requests,
+                rows,
+                semantic_index,
+                outcome,
+                ..
+            } = to_grpc_insert_requests(request, &mut ctx).unwrap();
 
             assert_eq!(outcome.accepted_data_points, 0);
             assert_eq!(outcome.rejected_data_points, 1);
@@ -2269,14 +2548,17 @@ mod tests {
             experimental_enable_exponential_histogram: true,
             ..Default::default()
         };
-        let (new_requests, _, _, _) =
-            to_grpc_insert_requests(request.clone(), &mut new_ctx).unwrap();
+        let new_requests = to_grpc_insert_requests(request.clone(), &mut new_ctx)
+            .unwrap()
+            .requests;
         let mut legacy_ctx = OtlpMetricCtx {
             experimental_enable_exponential_histogram: true,
             is_legacy: true,
             ..Default::default()
         };
-        let (legacy_requests, _, _, _) = to_grpc_insert_requests(request, &mut legacy_ctx).unwrap();
+        let legacy_requests = to_grpc_insert_requests(request, &mut legacy_ctx)
+            .unwrap()
+            .requests;
 
         let new_insert = &new_requests.inserts[0];
         let legacy_insert = &legacy_requests.inserts[0];
@@ -2325,8 +2607,9 @@ mod tests {
             vec![ExponentialHistogramDataPoint::default()],
             AggregationTemporality::Cumulative,
         )]);
-        let (_, _, _, outcome) =
-            to_grpc_insert_requests(request, &mut OtlpMetricCtx::default()).unwrap();
+        let outcome = to_grpc_insert_requests(request, &mut OtlpMetricCtx::default())
+            .unwrap()
+            .outcome;
 
         assert_eq!(outcome.rejected_data_points, 1);
         assert!(outcome.error_message.unwrap().len() <= MAX_REJECTION_MESSAGE_BYTES);

src/servers/src/otlp/metrics/resource_info.rs

@@ -0,0 +1,393 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Resource descriptor synthesized from OTLP metrics requests, so the entity
+//! graph reads one table instead of scanning every logical metric table for
+//! attributes the promote filter may have dropped.
+//!
+//! Columns are a fixed allowlist under the raw OTel attribute names: the
+//! conventions whitelist matches fixed names, so they must not follow the
+//! per-request label translation strategy or the promote/ignore headers.
+
+use std::collections::BTreeMap;
+
+use api::v1::RowInsertRequests;
+use common_catalog::consts::SEMANTIC_GRAPH_WINDOW_NANOS;
+use common_grpc::precision::Precision;
+use common_query::prelude::{greptime_timestamp, greptime_value};
+use otel_arrow_rust::proto::opentelemetry::common::v1::KeyValue;
+use otel_arrow_rust::proto::opentelemetry::metrics::v1::{ResourceMetrics, metric};
+use session::protocol_ctx::OtlpMetricCtx;
+
+use crate::error::Result;
+use crate::otlp::metrics::{
+    INSTANCE_KEY, JOB_KEY, ServiceIdentity, exponential_histogram_gate,
+    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,
+};
+use crate::row_writer::{self, MultiTableData};
+
+/// Prefixed like the other engine-managed tables, so a user metric is
+/// unlikely to claim the name.
+pub const OTEL_RESOURCE_INFO_TABLE_NAME: &str = "greptime_otel_resource_info";
+
+/// Attributes projected under their raw OTel keys. `service.instance.id` is
+/// absent on purpose: it lands in `instance`.
+///
+/// Matched instead of scanned: this runs for every attribute of every
+/// resource, and the compiler turns it into a length-and-prefix dispatch.
+fn is_projected_attr(key: &str) -> bool {
+    matches!(
+        key,
+        KEY_SERVICE_NAME
+            | KEY_SERVICE_NAMESPACE
+            | KEY_HOST_ID
+            | KEY_HOST_NAME
+            | KEY_CONTAINER_ID
+            | KEY_CONTAINER_NAME
+            | KEY_K8S_POD_UID
+            | KEY_K8S_POD_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;
+
+/// Projected attributes (sorted `(name, value)` pairs) -> graph window ->
+/// the newest data-point time seen in that window, which is what the row for
+/// that window is stamped with.
+///
+/// Windows are an inner map so the attributes are stored, and moved, once per
+/// resource. They are keyed separately because one request may carry data for
+/// several of them: a single row per resource would describe only the newest
+/// window, leaving the earlier ones with metric rows but no entities.
+#[derive(Debug, Default)]
+pub struct ResourceInfoData {
+    rows: BTreeMap<Vec<(String, String)>, BTreeMap<i64, i64>>,
+}
+
+impl ResourceInfoData {
+    /// Takes the raw attributes, before the promote filter runs on them.
+    pub fn observe(
+        &mut self,
+        raw_attrs: &[KeyValue],
+        resource: &ResourceMetrics,
+        metric_ctx: &OtlpMetricCtx,
+    ) {
+        let mut tags = Vec::with_capacity(MAX_PROJECTED_TAGS);
+        let ServiceIdentity { job, instance } = service_identity(raw_attrs);
+        if let Some(job) = job {
+            tags.push((JOB_KEY.to_string(), job));
+        }
+        if let Some(instance) = instance {
+            tags.push((INSTANCE_KEY.to_string(), instance));
+        }
+        for kv in raw_attrs {
+            if is_projected_attr(&kv.key)
+                && let Some(value) = scalar_value_string(kv.value.as_ref())
+            {
+                tags.push((kv.key.clone(), value));
+            }
+        }
+        if tags.is_empty() {
+            return;
+        }
+        // Sorted so equal attribute sets share a key, and so the emitted
+        // columns keep a stable order.
+        tags.sort_unstable();
+
+        let mut observed: BTreeMap<i64, i64> = BTreeMap::new();
+        for_each_encoded_time(resource, metric_ctx, |ts| {
+            let window = ts - ts.rem_euclid(SEMANTIC_GRAPH_WINDOW_NANOS);
+            observed
+                .entry(window)
+                .and_modify(|newest| *newest = (*newest).max(ts))
+                .or_insert(ts);
+        });
+        if observed.is_empty() {
+            return;
+        }
+
+        let windows = self.rows.entry(tags).or_default();
+        for (window, newest) in observed {
+            windows
+                .entry(window)
+                .and_modify(|seen| *seen = (*seen).max(newest))
+                .or_insert(newest);
+        }
+    }
+
+    /// Every projected attribute becomes a tag, so auto-create puts it in the
+    /// primary key where the conventions expect it.
+    pub fn into_row_insert_requests(self) -> Result<Option<RowInsertRequests>> {
+        if self.rows.is_empty() {
+            return Ok(None);
+        }
+
+        let mut writer = MultiTableData::default();
+        let table = writer.get_or_default_table_data(
+            OTEL_RESOURCE_INFO_TABLE_NAME,
+            MAX_PROJECTED_TAGS + 2,
+            self.rows.values().map(BTreeMap::len).sum(),
+        );
+        for (tags, windows) in &self.rows {
+            for ts_nanos in windows.values().copied() {
+                let mut row = table.alloc_one_row();
+                row_writer::write_tags(table, tags.iter().cloned(), &mut row)?;
+                row_writer::write_f64(table, greptime_value(), 1.0, &mut row)?;
+                row_writer::write_ts_to_millis(
+                    table,
+                    greptime_timestamp(),
+                    Some(ts_nanos),
+                    Precision::Nanosecond,
+                    &mut row,
+                )?;
+                table.add_row(row);
+            }
+        }
+
+        let (requests, _) = writer.into_row_insert_requests();
+        Ok(Some(requests))
+    }
+}
+
+/// Visits the times of the data points the encoder writes rows for, so a
+/// resource is described exactly where it is measured rather than wherever
+/// its request happens to reach.
+fn for_each_encoded_time(
+    resource: &ResourceMetrics,
+    metric_ctx: &OtlpMetricCtx,
+    mut visit: impl FnMut(i64),
+) {
+    fn visit_all(points: impl Iterator<Item = u64>, visit: &mut impl FnMut(i64)) {
+        for ts in points {
+            visit(ts as i64);
+        }
+    }
+    for scope in &resource.scope_metrics {
+        for m in &scope.metrics {
+            match &m.data {
+                Some(metric::Data::Gauge(g)) => {
+                    visit_all(g.data_points.iter().map(|p| p.time_unix_nano), &mut visit)
+                }
+                Some(metric::Data::Sum(s)) => {
+                    visit_all(s.data_points.iter().map(|p| p.time_unix_nano), &mut visit)
+                }
+                Some(metric::Data::Histogram(h)) => {
+                    visit_all(h.data_points.iter().map(|p| p.time_unix_nano), &mut visit)
+                }
+                Some(metric::Data::Summary(s)) => {
+                    visit_all(s.data_points.iter().map(|p| p.time_unix_nano), &mut visit)
+                }
+                Some(metric::Data::ExponentialHistogram(h))
+                    if exponential_histogram_gate(h, metric_ctx).is_ok() =>
+                {
+                    for point in &h.data_points {
+                        if let Ok((_, ts)) = exponential_histogram_value(point) {
+                            visit(ts);
+                        }
+                    }
+                }
+                Some(metric::Data::ExponentialHistogram(_)) => {}
+                None => {}
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use api::v1::SemanticType;
+    use api::v1::value::ValueData;
+    use common_query::prelude::set_default_prefix;
+    use otel_arrow_rust::proto::opentelemetry::common::v1::{AnyValue, any_value};
+    use otel_arrow_rust::proto::opentelemetry::metrics::v1::{
+        AggregationTemporality, ExponentialHistogram, ExponentialHistogramDataPoint, Gauge, Metric,
+        NumberDataPoint, ScopeMetrics,
+    };
+
+    use super::*;
+
+    fn kv(key: &str, value: &str) -> KeyValue {
+        KeyValue {
+            key: key.into(),
+            value: Some(AnyValue {
+                value: Some(any_value::Value::StringValue(value.into())),
+            }),
+        }
+    }
+
+    fn gauge_at(times: &[i64]) -> ResourceMetrics {
+        ResourceMetrics {
+            scope_metrics: vec![ScopeMetrics {
+                metrics: vec![Metric {
+                    data: Some(metric::Data::Gauge(Gauge {
+                        data_points: times
+                            .iter()
+                            .map(|ts| NumberDataPoint {
+                                time_unix_nano: *ts as u64,
+                                ..Default::default()
+                            })
+                            .collect(),
+                    })),
+                    ..Default::default()
+                }],
+                ..Default::default()
+            }],
+            ..Default::default()
+        }
+    }
+
+    #[test]
+    fn observe_projects_allowlist_and_dedups_per_request() {
+        let mut data = ResourceInfoData::default();
+        let attrs = vec![
+            kv("service.name", "api"),
+            kv("service.namespace", "shop"),
+            kv("service.instance.id", "inst-1"),
+            kv("host.id", "h-1"),
+            kv("os.type", "linux"),
+        ];
+        data.observe(&attrs, &gauge_at(&[100, 50]), &OtlpMetricCtx::default());
+        assert_eq!(data.rows.len(), 1);
+        let (tags, windows) = data.rows.iter().next().unwrap();
+        assert_eq!(windows.values().copied().collect::<Vec<_>>(), vec![100]);
+        assert!(tags.contains(&("job".to_string(), "shop/api".to_string())));
+        assert!(tags.contains(&("instance".to_string(), "inst-1".to_string())));
+        assert!(tags.contains(&("service.name".to_string(), "api".to_string())));
+        assert!(
+            tags.iter()
+                .all(|(k, _)| k != "os.type" && k != "service.instance.id")
+        );
+
+        data.observe(
+            &[kv("host.id", "h-2")],
+            &gauge_at(&[10]),
+            &OtlpMetricCtx::default(),
+        );
+        assert_eq!(data.rows.len(), 2);
+
+        let mut empty = ResourceInfoData::default();
+        empty.observe(
+            &[kv("os.type", "linux")],
+            &gauge_at(&[100]),
+            &OtlpMetricCtx::default(),
+        );
+        assert!(empty.into_row_insert_requests().unwrap().is_none());
+    }
+
+    /// Earlier windows would keep their metric rows but lose their entities.
+    #[test]
+    fn observe_keeps_one_row_per_graph_window() {
+        let window = SEMANTIC_GRAPH_WINDOW_NANOS;
+        let mut data = ResourceInfoData::default();
+        data.observe(
+            &[kv("service.name", "api")],
+            &gauge_at(&[window + 1, window + 2, 3 * window + 7]),
+            &OtlpMetricCtx::default(),
+        );
+
+        let windows = data.rows.values().next().unwrap();
+        assert_eq!(
+            windows.iter().collect::<Vec<_>>(),
+            vec![(&window, &(window + 2)), (&(3 * window), &(3 * window + 7))]
+        );
+    }
+
+    /// Describing a resource whose only data the encoder drops invents an
+    /// entity with no measurements.
+    #[test]
+    fn observe_ignores_data_the_encoder_drops() {
+        let exponential = |temporality: AggregationTemporality| ResourceMetrics {
+            scope_metrics: vec![ScopeMetrics {
+                metrics: vec![Metric {
+                    data: Some(metric::Data::ExponentialHistogram(ExponentialHistogram {
+                        data_points: vec![ExponentialHistogramDataPoint {
+                            time_unix_nano: 100,
+                            ..Default::default()
+                        }],
+                        aggregation_temporality: temporality as i32,
+                    })),
+                    ..Default::default()
+                }],
+                ..Default::default()
+            }],
+            ..Default::default()
+        };
+        let enabled = OtlpMetricCtx {
+            experimental_enable_exponential_histogram: true,
+            ..Default::default()
+        };
+
+        for (resource, ctx) in [
+            (
+                exponential(AggregationTemporality::Cumulative),
+                OtlpMetricCtx::default(),
+            ),
+            (exponential(AggregationTemporality::Delta), enabled),
+        ] {
+            let mut data = ResourceInfoData::default();
+            data.observe(&[kv("service.name", "api")], &resource, &ctx);
+            assert!(data.into_row_insert_requests().unwrap().is_none());
+        }
+    }
+
+    #[test]
+    fn rows_carry_raw_key_tags_value_and_millis_timestamp() {
+        set_default_prefix(None).unwrap();
+        let mut data = ResourceInfoData::default();
+        data.observe(
+            &[kv("service.name", "api"), kv("host.id", "h-1")],
+            &gauge_at(&[1_700_000_000_123_456_789]),
+            &OtlpMetricCtx::default(),
+        );
+        let requests = data.into_row_insert_requests().unwrap().unwrap();
+        assert_eq!(requests.inserts.len(), 1);
+        let insert = &requests.inserts[0];
+        assert_eq!(insert.table_name, OTEL_RESOURCE_INFO_TABLE_NAME);
+
+        let rows = insert.rows.as_ref().unwrap();
+        let names = rows
+            .schema
+            .iter()
+            .map(|c| c.column_name.as_str())
+            .collect::<Vec<_>>();
+        assert_eq!(
+            names,
+            vec![
+                "host.id",
+                "job",
+                "service.name",
+                greptime_value(),
+                greptime_timestamp()
+            ]
+        );
+        for column in &rows.schema[..3] {
+            assert_eq!(column.semantic_type, SemanticType::Tag as i32);
+        }
+
+        assert_eq!(rows.rows.len(), 1);
+        let values = &rows.rows[0].values;
+        assert_eq!(values[3].value_data, Some(ValueData::F64Value(1.0)));
+        assert_eq!(
+            values[4].value_data,
+            Some(ValueData::TimestampMillisecondValue(1_700_000_000_123))
+        );
+    }
+}

src/servers/src/otlp/trace.rs

@@ -46,7 +46,15 @@ pub const TRACE_STATE_COLUMN: &str = "trace_state";
 
 // const keys
 pub const KEY_SERVICE_NAME: &str = "service.name";
+pub const KEY_SERVICE_NAMESPACE: &str = "service.namespace";
 pub const KEY_SERVICE_INSTANCE_ID: &str = "service.instance.id";
+pub const KEY_HOST_ID: &str = "host.id";
+pub const KEY_HOST_NAME: &str = "host.name";
+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_NAMESPACE_NAME: &str = "k8s.namespace.name";
 pub const KEY_SPAN_KIND: &str = "span.kind";
 
 // jaeger const keys, not sure if they are general

src/session/src/protocol_ctx.rs

@@ -56,6 +56,9 @@ pub struct OtlpMetricCtx {
     pub with_metric_engine: bool,
     pub experimental_enable_exponential_histogram: bool,
     pub is_legacy: bool,
+    /// Set from the server's `otlp.experimental_enable_resource_info`; off
+    /// means the resource descriptor is not synthesized at all.
+    pub resource_info: bool,
     pub metric_type: MetricType,
     pub metric_translation_strategy: OtlpMetricTranslationStrategy,
 }

src/table/src/requests/semantic.rs

@@ -98,6 +98,10 @@ pub const SEMANTIC_ENTITY_SERVICE_ID: &str = "greptime.semantic.entity.service.i
 /// ([`has_stable_string_form`]).
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum EntityRole {
+    /// Identifying columns. Their **order is part of the identity**: the
+    /// entity id is their values joined in the declared order, so two tables
+    /// naming the same entity must list them the same way (broad to narrow)
+    /// or they name two entities.
     Id,
     Descriptive,
     Scope,

tests-integration/src/instance.rs

@@ -691,4 +691,119 @@ mod tests {
             "denied source leaked into entities:\n{pretty_print}"
         );
     }
+
+    /// Denying the descriptor's table must not fail the request: the metric
+    /// data is already committed, so the client would retry what the server
+    /// accepted.
+    #[tokio::test(flavor = "multi_thread")]
+    async fn test_otlp_descriptor_permission_denial_degrades() {
+        use otel_arrow_rust::proto::opentelemetry::collector::metrics::v1::ExportMetricsServiceRequest;
+        use otel_arrow_rust::proto::opentelemetry::common::v1::{AnyValue, KeyValue, any_value};
+        use otel_arrow_rust::proto::opentelemetry::metrics::v1::number_data_point::Value;
+        use otel_arrow_rust::proto::opentelemetry::metrics::v1::{
+            Gauge, Metric, NumberDataPoint, ResourceMetrics, ScopeMetrics, metric,
+        };
+        use otel_arrow_rust::proto::opentelemetry::resource::v1::Resource;
+        use servers::query_handler::OpenTelemetryProtocolHandler;
+
+        common_telemetry::init_default_ut_logging();
+
+        /// Denies writes to the descriptor table only.
+        struct DenyResourceInfo;
+
+        impl auth::PermissionChecker for DenyResourceInfo {
+            fn check_permission(
+                &self,
+                _user_info: auth::UserInfoRef,
+                _req: auth::PermissionReq,
+            ) -> auth::error::Result<auth::PermissionResp> {
+                Ok(auth::PermissionResp::Allow)
+            }
+
+            fn check_permission_with_table_targets(
+                &self,
+                _user_info: auth::UserInfoRef,
+                _req: auth::PermissionReq,
+                targets: auth::PermissionTableTargets,
+            ) -> auth::error::Result<auth::PermissionResp> {
+                if let auth::PermissionTableTargets::Resolved(targets) = targets
+                    && targets
+                        .iter()
+                        .any(|target| target.table == "greptime_otel_resource_info")
+                {
+                    return Ok(auth::PermissionResp::Reject);
+                }
+                Ok(auth::PermissionResp::Allow)
+            }
+        }
+
+        let plugins = Plugins::new();
+        plugins.insert::<auth::PermissionCheckerRef>(Arc::new(DenyResourceInfo));
+
+        let standalone = GreptimeDbStandaloneBuilder::new("test_otlp_descriptor_denied")
+            .with_plugin(plugins)
+            .build()
+            .await;
+        let instance = standalone.fe_instance().clone();
+
+        let attr = |key: &str, value: &str| KeyValue {
+            key: key.to_string(),
+            value: Some(AnyValue {
+                value: Some(any_value::Value::StringValue(value.to_string())),
+            }),
+        };
+        let request = ExportMetricsServiceRequest {
+            resource_metrics: vec![ResourceMetrics {
+                resource: Some(Resource {
+                    attributes: vec![attr("service.name", "api"), attr("host.id", "h-1")],
+                    ..Default::default()
+                }),
+                scope_metrics: vec![ScopeMetrics {
+                    metrics: vec![Metric {
+                        name: "denied_descriptor_gauge".to_string(),
+                        data: Some(metric::Data::Gauge(Gauge {
+                            data_points: vec![NumberDataPoint {
+                                time_unix_nano: 1_700_000_000_000_000_000,
+                                value: Some(Value::AsDouble(1.0)),
+                                ..Default::default()
+                            }],
+                        })),
+                        ..Default::default()
+                    }],
+                    ..Default::default()
+                }],
+                ..Default::default()
+            }],
+        };
+
+        let outcome =
+            OpenTelemetryProtocolHandler::metrics(&*instance, request, QueryContext::arc())
+                .await
+                .unwrap();
+        let warning = outcome
+            .error_message
+            .expect("denial must surface as a warning");
+        assert!(
+            warning.contains("greptime_otel_resource_info"),
+            "unexpected warning: {warning}"
+        );
+
+        // the metric data is committed, the descriptor table was never created
+        let sql = "select table_name from information_schema.tables \
+                   where table_schema = 'public' order by table_name";
+        let output = query(&instance, sql).await;
+        let OutputData::Stream(s) = output.data else {
+            unreachable!()
+        };
+        let batches = common_recordbatch::util::collect_batches(s).await.unwrap();
+        let pretty_print = batches.pretty_print().unwrap();
+        assert!(
+            pretty_print.contains("denied_descriptor_gauge"),
+            "metric data was not committed:\n{pretty_print}"
+        );
+        assert!(
+            !pretty_print.contains("greptime_otel_resource_info"),
+            "denied descriptor table was created:\n{pretty_print}"
+        );
+    }
 }

tests-integration/src/standalone.rs

@@ -369,6 +369,11 @@ impl GreptimeDbStandaloneBuilder {
             grpc: GrpcOptions::default().with_server_addr("127.0.0.1:4001"),
             slow_query: self.slow_query_options.clone(),
             auto_create_table: self.auto_create_table,
+            // Tests cover the descriptor, so they run with it enabled.
+            otlp: frontend::service_config::OtlpOptions {
+                experimental_enable_resource_info: true,
+                ..Default::default()
+            },
             ..StandaloneOptions::default()
         };
 

tests-integration/tests/http.rs

@@ -164,6 +164,7 @@ macro_rules! http_tests {
                 test_otlp_metrics_new,
                 test_otlp_exponential_histogram,
                 test_otlp_metric_translation_strategies,
+                test_otlp_metrics_resource_info_conflicts,
                 test_otlp_traces_v0,
                 test_otlp_traces_v1,
                 test_otlp_traces_v1_entity_graph,
@@ -2208,6 +2209,7 @@ enable = true
 enable = true
 experimental_enable_exponential_histogram = false
 trace_ingest_chunk_size = 512
+experimental_enable_resource_info = true
 
 [prom_store]
 enable = true
@@ -6500,7 +6502,7 @@ pub async fn test_otlp_metrics_new(store_type: StorageType) {
     .await;
     assert_eq!(StatusCode::OK, res.status());
 
-    let expected = "[[\"claude_code_cost_usage_USD_total\"],[\"claude_code_token_usage_tokens_total\"],[\"demo\"],[\"greptime_physical_table\"],[\"numbers\"]]";
+    let expected = "[[\"claude_code_cost_usage_USD_total\"],[\"claude_code_token_usage_tokens_total\"],[\"demo\"],[\"greptime_otel_resource_info\"],[\"greptime_physical_table\"],[\"numbers\"]]";
     validate_data("otlp_metrics_all_tables", &client, "show tables;", expected).await;
 
     // Metric-engine logical table carries the semantic identity. Match substrings
@@ -6567,6 +6569,33 @@ pub async fn test_otlp_metrics_new(store_type: StorageType) {
     )
     .await;
 
+    // The synthesized resource descriptor: a plain mito info table keyed by
+    // the raw OTel attribute keys (service.name is the only allowlisted
+    // resource attribute in this payload), independent of the promote/ignore
+    // headers. The timestamp is the newest data-point time.
+    validate_data(
+        "otlp_metrics_resource_info",
+        &client,
+        "select job, \"service.name\", greptime_value, greptime_timestamp from greptime_otel_resource_info;",
+        "[[\"claude-code\",\"claude-code\",1.0,1753780559836]]",
+    )
+    .await;
+    // All four ownership markers must land on auto-create: the descriptor
+    // write path refuses tables that miss any of them, so a missing stamp
+    // here would silently stop every follow-up descriptor write.
+    validate_data(
+        "otlp_metrics_resource_info_options",
+        &client,
+        "select count(*) from information_schema.tables where table_name = 'greptime_otel_resource_info' \
+         and engine = 'mito' \
+         and create_options like '%greptime.semantic.metric.type=info%' \
+         and create_options like '%greptime.semantic.metric.metadata_quality=declared%' \
+         and create_options like '%greptime.semantic.signal_type=metric%' \
+         and create_options like '%greptime.semantic.source=opentelemetry%';",
+        "[[1]]",
+    )
+    .await;
+
     // drop table
     let res = client
         .get("/v1/sql?sql=drop table `claude_code_cost_usage_USD_total`;")
@@ -6578,6 +6607,11 @@ pub async fn test_otlp_metrics_new(store_type: StorageType) {
         .send()
         .await;
     assert_eq!(res.status(), StatusCode::OK);
+    let res = client
+        .get("/v1/sql?sql=drop table greptime_otel_resource_info;")
+        .send()
+        .await;
+    assert_eq!(res.status(), StatusCode::OK);
 
     // write metrics data
     // with scope attrs
@@ -6704,6 +6738,16 @@ pub async fn test_otlp_metrics_new(store_type: StorageType) {
     )
     .await;
 
+    // the descriptor was auto-created again after the earlier drop
+    validate_data(
+        "otlp_metrics_resource_info_recreated",
+        &client,
+        "select count(*) from information_schema.tables \
+         where table_name = 'greptime_otel_resource_info' and engine = 'mito';",
+        "[[1]]",
+    )
+    .await;
+
     // drop table
     let res = client
         .get("/v1/sql?sql=drop table `claude_code_cost_usage_USD_total`;")
@@ -6715,6 +6759,44 @@ pub async fn test_otlp_metrics_new(store_type: StorageType) {
         .send()
         .await;
     assert_eq!(res.status(), StatusCode::OK);
+    let res = client
+        .get("/v1/sql?sql=drop table greptime_otel_resource_info;")
+        .send()
+        .await;
+    assert_eq!(res.status(), StatusCode::OK);
+
+    // job composition end-to-end: service.namespace folds into "<ns>/<name>"
+    // on both the descriptor and the metric tables
+    let content = r#"
+{"resourceMetrics":[{"resource":{"attributes":[{"key":"service.name","value":{"stringValue":"api"}},{"key":"service.namespace","value":{"stringValue":"shop"}},{"key":"host.id","value":{"stringValue":"h-1"}}]},"scopeMetrics":[{"scope":{"name":"s"},"metrics":[{"name":"ns_gauge","gauge":{"dataPoints":[{"timeUnixNano":"1753780559836000000","asDouble":1.0}]}}]}]}]}
+    "#;
+    let req: ExportMetricsServiceRequest = serde_json::from_str(content).unwrap();
+    let res = send_req(
+        &client,
+        vec![(
+            HeaderName::from_static("content-type"),
+            HeaderValue::from_static("application/x-protobuf"),
+        )],
+        "/v1/otlp/v1/metrics",
+        req.encode_to_vec(),
+        false,
+    )
+    .await;
+    assert_eq!(StatusCode::OK, res.status());
+    validate_data(
+        "otlp_metrics_namespace_job_descriptor",
+        &client,
+        "select job, \"service.name\", \"service.namespace\", \"host.id\" from greptime_otel_resource_info;",
+        "[[\"shop/api\",\"api\",\"shop\",\"h-1\"]]",
+    )
+    .await;
+    validate_data(
+        "otlp_metrics_namespace_job_metric_table",
+        &client,
+        "select job from ns_gauge;",
+        "[[\"shop/api\"]]",
+    )
+    .await;
 
     guard.remove_all().await;
 }
@@ -6912,6 +6994,24 @@ pub async fn test_otlp_metric_translation_strategies(store_type: StorageType) {
     )
     .await;
 
+    // the descriptor's columns are the fixed raw keys regardless of the
+    // translation strategy, and only allowlisted attributes are projected
+    validate_data(
+        "otlp_metric_translation_strategy_resource_info",
+        &client,
+        "select job, \"service.name\" from greptime_otel_resource_info;",
+        "[[\"strategy-service\",\"strategy-service\"]]",
+    )
+    .await;
+    validate_data(
+        "otlp_metric_translation_strategy_resource_info_allowlist",
+        &client,
+        "select count(*) from information_schema.columns \
+         where table_name = 'greptime_otel_resource_info' and column_name = 'resource.attr';",
+        "[[0]]",
+    )
+    .await;
+
     let res = send_req(
         &client,
         vec![
@@ -6934,6 +7034,122 @@ pub async fn test_otlp_metric_translation_strategies(store_type: StorageType) {
     guard.remove_all().await;
 }
 
+pub async fn test_otlp_metrics_resource_info_conflicts(store_type: StorageType) {
+    common_telemetry::init_default_ut_logging();
+    let (app, mut guard) =
+        setup_test_http_app_with_frontend(store_type, "test_otlp_metrics_resource_info_conflicts")
+            .await;
+    let client = TestClient::new(app).await;
+
+    let content_type = || {
+        (
+            HeaderName::from_static("content-type"),
+            HeaderValue::from_static("application/x-protobuf"),
+        )
+    };
+
+    // A metric named like the descriptor wins the table name: it goes through
+    // the normal metric-engine path and no descriptor is synthesized.
+    let content = r#"
+{"resourceMetrics":[{"resource":{"attributes":[{"key":"service.name","value":{"stringValue":"api"}},{"key":"host.id","value":{"stringValue":"h-1"}}]},"scopeMetrics":[{"scope":{"name":"s"},"metrics":[{"name":"greptime_otel_resource_info","gauge":{"dataPoints":[{"timeUnixNano":"1753780559836000000","asDouble":1.0}]}}]}]}]}
+    "#;
+    let req: ExportMetricsServiceRequest = serde_json::from_str(content).unwrap();
+    let res = send_req(
+        &client,
+        vec![content_type()],
+        "/v1/otlp/v1/metrics",
+        req.encode_to_vec(),
+        false,
+    )
+    .await;
+    assert_eq!(StatusCode::OK, res.status());
+    let body = ExportMetricsServiceResponse::decode(res.bytes().await).unwrap();
+    assert!(body.partial_success.is_none());
+    validate_data(
+        "otlp_metrics_name_collision_engine",
+        &client,
+        "select count(*) from information_schema.tables \
+         where table_name = 'greptime_otel_resource_info' and engine = 'metric';",
+        "[[1]]",
+    )
+    .await;
+    let res = execute_sql(&client, "drop table greptime_otel_resource_info;").await;
+    assert_eq!(res.status(), StatusCode::OK);
+
+    // A pre-existing table with an incompatible schema fails the descriptor
+    // write, which degrades to an OTLP partial-success warning while the
+    // metric data itself is accepted.
+    let res = execute_sql(
+        &client,
+        "create table greptime_otel_resource_info (ts timestamp time index, job double, greptime_value double);",
+    )
+    .await;
+    assert_eq!(res.status(), StatusCode::OK);
+
+    let content = r#"
+{"resourceMetrics":[{"resource":{"attributes":[{"key":"service.name","value":{"stringValue":"api"}}]},"scopeMetrics":[{"scope":{"name":"s"},"metrics":[{"name":"conflict_gauge","gauge":{"dataPoints":[{"timeUnixNano":"1753780559836000000","asDouble":7.0}]}}]}]}]}
+    "#;
+    let req: ExportMetricsServiceRequest = serde_json::from_str(content).unwrap();
+    let res = send_req(
+        &client,
+        vec![content_type()],
+        "/v1/otlp/v1/metrics",
+        req.encode_to_vec(),
+        false,
+    )
+    .await;
+    assert_eq!(StatusCode::OK, res.status());
+    let body = ExportMetricsServiceResponse::decode(res.bytes().await).unwrap();
+    let partial_success = body.partial_success.as_ref().unwrap();
+    assert_eq!(partial_success.rejected_data_points, 0);
+    assert!(
+        partial_success
+            .error_message
+            .contains("greptime_otel_resource_info"),
+        "unexpected warning: {}",
+        partial_success.error_message
+    );
+    validate_data(
+        "otlp_metrics_conflict_main_data_accepted",
+        &client,
+        "select greptime_value from conflict_gauge;",
+        "[[7.0]]",
+    )
+    .await;
+
+    // The descriptor is auto-created again once the conflicting table is
+    // gone, and a follow-up write into it does not degrade.
+    let res = execute_sql(&client, "drop table greptime_otel_resource_info;").await;
+    assert_eq!(res.status(), StatusCode::OK);
+    for round in 0..2 {
+        let req: ExportMetricsServiceRequest = serde_json::from_str(content).unwrap();
+        let res = send_req(
+            &client,
+            vec![content_type()],
+            "/v1/otlp/v1/metrics",
+            req.encode_to_vec(),
+            false,
+        )
+        .await;
+        assert_eq!(StatusCode::OK, res.status());
+        let body = ExportMetricsServiceResponse::decode(res.bytes().await).unwrap();
+        assert!(
+            body.partial_success.is_none(),
+            "round {round} unexpectedly degraded: {:?}",
+            body.partial_success
+        );
+    }
+    validate_data(
+        "otlp_metrics_owned_descriptor_keeps_accepting",
+        &client,
+        "select count(*) from greptime_otel_resource_info;",
+        "[[1]]",
+    )
+    .await;
+
+    guard.remove_all().await;
+}
+
 pub async fn test_otlp_traces_v0(store_type: StorageType) {
     // init
     common_telemetry::init_default_ut_logging();
@@ -7136,7 +7352,7 @@ pub async fn test_otlp_traces_v1_entity_graph(store_type: StorageType) {
     let res = send_trace_v1_req(&client, "graph_traces", req, false).await;
     assert_eq!(StatusCode::OK, res.status());
 
-    let expected = r#"[["service","frontend","service","cart","calls"],["service.instance","resource_attributes.service.instance.id=binst-1,service_name=batch","service","batch","part_of"],["service.instance","resource_attributes.service.instance.id=inst-1,service_name=frontend","service","frontend","part_of"],["service.instance","resource_attributes.service.instance.id=inst-1,service_name=frontend","k8s.pod","uid-9","runs_on"]]"#;
+    let expected = r#"[["service","frontend","service","cart","calls"],["service.instance","batch,binst-1","service","batch","part_of"],["service.instance","frontend,inst-1","service","frontend","part_of"],["service.instance","frontend,inst-1","k8s.pod","uid-9","runs_on"]]"#;
     validate_data(
         "otlp_traces_entity_graph",
         &client,

tests/cases/standalone/common/system/semantic_graph.result

@@ -94,26 +94,26 @@ select entity_type, entity_id, entity_id_attrs, scope, descriptive, source_table
 from greptime_private.semantic_entities
 order by entity_type, entity_id;
 
-+------------------+---------------------------+-------------------------------------+---------+-------------------+------------------------------+
-| entity_type      | entity_id                 | entity_id_attrs                     | scope   | descriptive       | source_tables                |
-+------------------+---------------------------+-------------------------------------+---------+-------------------+------------------------------+
-| host             | h1                        |                                     |         |                   | ["public.graph_app_metrics"] |
-| process          | host=h1,service_name=cart | {"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"] |
-+------------------+---------------------------+-------------------------------------+---------+-------------------+------------------------------+
++------------------+-----------+-------------------------------------+---------+-------------------+------------------------------+
+| entity_type      | entity_id | entity_id_attrs                     | scope   | descriptive       | source_tables                |
++------------------+-----------+-------------------------------------+---------+-------------------+------------------------------+
+| 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"] |
++------------------+-----------+-------------------------------------+---------+-------------------+------------------------------+
 
 select src_type, src_id, dst_type, dst_id, rel_type, provenance, confidence
 from greptime_private.semantic_relationships
 order by rel_type, src_id;
 
-+------------------+---------------------------+----------+--------+----------+------------+------------+
-| src_type         | src_id                    | dst_type | dst_id | rel_type | provenance | confidence |
-+------------------+---------------------------+----------+--------+----------+------------+------------+
-| service.instance | cart-0                    | service  | cart   | part_of  | attribute  | 1.0        |
-| service.instance | cart-0                    | host     | h1     | runs_on  | attribute  | 1.0        |
-| process          | host=h1,service_name=cart | host     | h1     | runs_on  | attribute  | 1.0        |
-+------------------+---------------------------+----------+--------+----------+------------+------------+
++------------------+---------+----------+--------+----------+------------+------------+
+| src_type         | src_id  | dst_type | dst_id | rel_type | provenance | confidence |
++------------------+---------+----------+--------+----------+------------+------------+
+| service.instance | cart-0  | service  | cart   | part_of  | attribute  | 1.0        |
+| process          | cart,h1 | host     | h1     | runs_on  | attribute  | 1.0        |
+| service.instance | cart-0  | host     | h1     | runs_on  | attribute  | 1.0        |
++------------------+---------+----------+--------+----------+------------+------------+
 
 drop table graph_app_metrics;
 
@@ -570,38 +570,38 @@ select entity_type, entity_id, source_tables
 from greptime_private.semantic_entities
 order by entity_type, entity_id, source_tables;
 
-+------------------+-----------------------------------------------------------+------------------------------------+
-| entity_type      | entity_id                                                 | source_tables                      |
-+------------------+-----------------------------------------------------------+------------------------------------+
-| k8s.container    | container=main,uid=uid-1                                  | ["public.kube_pod_container_info"] |
-| k8s.node         | node-a                                                    | ["public.kube_pod_info"]           |
-| k8s.pod          | uid-1                                                     | ["public.kube_pod_info"]           |
-| k8s.pod          | uid-1                                                     | ["public.kube_pod_owner"]          |
-| k8s.pod          | uid-1                                                     | ["public.kube_pod_container_info"] |
-| k8s.pod          | uid-2                                                     | ["public.kube_pod_info"]           |
-| k8s.pod          | uid-2                                                     | ["public.kube_pod_owner"]          |
-| k8s.pod          | uid-3                                                     | ["public.kube_pod_info"]           |
-| k8s.service      | svc-uid-1                                                 | ["public.kube_service_info"]       |
-| k8s.workload     | namespace=default,owner_kind=ReplicaSet,owner_name=api-rs | ["public.kube_pod_owner"]          |
-| service          | shop/api                                                  | ["public.target_info"]             |
-| service.instance | instance=inst-1,job=shop/api                              | ["public.target_info"]             |
-+------------------+-----------------------------------------------------------+------------------------------------+
++------------------+---------------------------+------------------------------------+
+| entity_type      | entity_id                 | source_tables                      |
++------------------+---------------------------+------------------------------------+
+| k8s.container    | uid-1,main                | ["public.kube_pod_container_info"] |
+| k8s.node         | node-a                    | ["public.kube_pod_info"]           |
+| k8s.pod          | uid-1                     | ["public.kube_pod_info"]           |
+| k8s.pod          | uid-1                     | ["public.kube_pod_owner"]          |
+| k8s.pod          | uid-1                     | ["public.kube_pod_container_info"] |
+| k8s.pod          | uid-2                     | ["public.kube_pod_info"]           |
+| k8s.pod          | uid-2                     | ["public.kube_pod_owner"]          |
+| k8s.pod          | uid-3                     | ["public.kube_pod_info"]           |
+| k8s.service      | svc-uid-1                 | ["public.kube_service_info"]       |
+| k8s.workload     | default,ReplicaSet,api-rs | ["public.kube_pod_owner"]          |
+| service          | shop/api                  | ["public.target_info"]             |
+| service.instance | shop/api,inst-1           | ["public.target_info"]             |
++------------------+---------------------------+------------------------------------+
 
 -- 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 |
-+------------------+------------------------------+---------------+-----------------------------------------------------------+----------+------------+
-| k8s.pod          | uid-1                        | k8s.container | container=main,uid=uid-1                                  | contains | attribute  |
-| service.instance | instance=inst-1,job=shop/api | service       | shop/api                                                  | part_of  | attribute  |
-| k8s.pod          | uid-1                        | k8s.workload  | namespace=default,owner_kind=ReplicaSet,owner_name=api-rs | part_of  | attribute  |
-| k8s.pod          | uid-2                        | k8s.workload  | namespace=default,owner_kind=ReplicaSet,owner_name=api-rs | part_of  | attribute  |
-| k8s.pod          | uid-1                        | k8s.node      | node-a                                                    | runs_on  | attribute  |
-| k8s.pod          | uid-2                        | k8s.node      | node-a                                                    | runs_on  | attribute  |
-+------------------+------------------------------+---------------+-----------------------------------------------------------+----------+------------+
++------------------+-----------------+---------------+---------------------------+----------+------------+
+| src_type         | src_id          | dst_type      | dst_id                    | rel_type | provenance |
++------------------+-----------------+---------------+---------------------------+----------+------------+
+| k8s.pod          | uid-1           | k8s.container | uid-1,main                | contains | attribute  |
+| service.instance | shop/api,inst-1 | service       | shop/api                  | part_of  | attribute  |
+| k8s.pod          | uid-1           | k8s.workload  | default,ReplicaSet,api-rs | part_of  | attribute  |
+| k8s.pod          | uid-2           | k8s.workload  | default,ReplicaSet,api-rs | part_of  | attribute  |
+| k8s.pod          | uid-1           | k8s.node      | node-a                    | runs_on  | attribute  |
+| k8s.pod          | uid-2           | k8s.node      | node-a                    | runs_on  | attribute  |
++------------------+-----------------+---------------+---------------------------+----------+------------+
 
 drop table kube_pod_info;
 
@@ -627,3 +627,182 @@ drop table http_requests_total;
 
 Affected Rows: 0
 
+-- OTel conventions: the ingestion-synthesized greptime_otel_resource_info descriptor
+-- (stamped signal_type=metric + source=opentelemetry + metric.type=info) gets implicit
+-- declarations under the raw OTel column names. host/container identities are
+-- 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.
+create table greptime_otel_resource_info (
+  greptime_timestamp timestamp(3) time index,
+  job string,
+  instance string,
+  "service.name" string,
+  "service.namespace" string,
+  "host.id" string,
+  "host.name" string,
+  "container.id" string,
+  "container.name" string,
+  greptime_value double,
+  primary key (job, instance, "service.name", "service.namespace",
+    "host.id", "host.name", "container.id", "container.name")
+) with (
+  'greptime.semantic.signal_type' = 'metric',
+  'greptime.semantic.source' = 'opentelemetry',
+  'greptime.semantic.metric.type' = '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);
+
+Affected Rows: 3
+
+-- SQLNESS PROTOCOL MYSQL
+select entity_type, entity_id, source_tables
+from greptime_private.semantic_entities
+order by entity_type, entity_id, source_tables;
+
++------------------+-----------------+----------------------------------------+
+| entity_type      | entity_id       | source_tables                          |
++------------------+-----------------+----------------------------------------+
+| container        | c-1             | ["public.greptime_otel_resource_info"] |
+| host             | h-1             | ["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"] |
++------------------+-----------------+----------------------------------------+
+
+-- 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
+
+drop table greptime_otel_resource_info;
+
+Affected Rows: 0
+
+-- 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
+-- job, so this asserts one entity per service and per instance, sourced from
+-- both tables.
+create table graph_xsignal_traces (
+  "timestamp" timestamp(9) time index,
+  trace_id string,
+  span_id string,
+  parent_span_id string,
+  span_kind string,
+  span_status_code string,
+  service_name string,
+  duration_nano bigint unsigned,
+  "resource_attributes.service.namespace" string,
+  "resource_attributes.service.instance.id" string,
+  primary key (service_name, "resource_attributes.service.namespace",
+    "resource_attributes.service.instance.id")
+) with ('table_data_model' = 'greptime_trace_v1', 'append_mode' = 'true');
+
+Affected Rows: 0
+
+create table greptime_otel_resource_info (
+  greptime_timestamp timestamp(3) time index,
+  job string,
+  instance string,
+  "service.name" string,
+  "service.namespace" string,
+  greptime_value double,
+  primary key (job, instance, "service.name", "service.namespace")
+) with (
+  'greptime.semantic.signal_type' = 'metric',
+  'greptime.semantic.source' = 'opentelemetry',
+  'greptime.semantic.metric.type' = 'info'
+);
+
+Affected Rows: 0
+
+insert into graph_xsignal_traces values
+  (now(), 't1', 's1', NULL, 'SPAN_KIND_SERVER', 'STATUS_CODE_UNSET', 'api', 100, 'shop', 'inst-1'),
+  (now(), 't2', 's2', NULL, 'SPAN_KIND_SERVER', 'STATUS_CODE_UNSET', 'worker', 100, '', 'inst-2');
+
+Affected Rows: 2
+
+insert into greptime_otel_resource_info values
+  (now(), 'shop/api', 'inst-1', 'api', 'shop', 1),
+  (now(), 'worker', 'inst-2', 'worker', '', 1);
+
+Affected Rows: 2
+
+-- SQLNESS PROTOCOL MYSQL
+select entity_type, entity_id, source_tables
+from greptime_private.semantic_entities
+where entity_type in ('service', 'service.instance')
+order by entity_type, entity_id, source_tables;
+
++------------------+-----------------+----------------------------------------+
+| entity_type      | entity_id       | source_tables                          |
++------------------+-----------------+----------------------------------------+
+| service          | shop/api        | ["public.graph_xsignal_traces"]        |
+| service          | shop/api        | ["public.greptime_otel_resource_info"] |
+| service          | worker          | ["public.graph_xsignal_traces"]        |
+| service          | worker          | ["public.greptime_otel_resource_info"] |
+| service.instance | shop/api,inst-1 | ["public.graph_xsignal_traces"]        |
+| service.instance | shop/api,inst-1 | ["public.greptime_otel_resource_info"] |
+| service.instance | worker,inst-2   | ["public.graph_xsignal_traces"]        |
+| service.instance | worker,inst-2   | ["public.greptime_otel_resource_info"] |
++------------------+-----------------+----------------------------------------+
+
+drop table graph_xsignal_traces;
+
+Affected Rows: 0
+
+drop table greptime_otel_resource_info;
+
+Affected Rows: 0
+

tests/cases/standalone/common/system/semantic_graph.sql

@@ -387,3 +387,122 @@ drop table kube_service_info;
 drop table target_info;
 
 drop table http_requests_total;
+
+-- OTel conventions: the ingestion-synthesized greptime_otel_resource_info descriptor
+-- (stamped signal_type=metric + source=opentelemetry + metric.type=info) gets implicit
+-- declarations under the raw OTel column names. host/container identities are
+-- 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.
+create table greptime_otel_resource_info (
+  greptime_timestamp timestamp(3) time index,
+  job string,
+  instance string,
+  "service.name" string,
+  "service.namespace" string,
+  "host.id" string,
+  "host.name" string,
+  "container.id" string,
+  "container.name" string,
+  greptime_value double,
+  primary key (job, instance, "service.name", "service.namespace",
+    "host.id", "host.name", "container.id", "container.name")
+) with (
+  'greptime.semantic.signal_type' = 'metric',
+  'greptime.semantic.source' = 'opentelemetry',
+  'greptime.semantic.metric.type' = '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);
+
+-- SQLNESS PROTOCOL MYSQL
+select entity_type, entity_id, source_tables
+from greptime_private.semantic_entities
+order by entity_type, entity_id, source_tables;
+
+-- 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;
+
+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
+-- job, so this asserts one entity per service and per instance, sourced from
+-- both tables.
+create table graph_xsignal_traces (
+  "timestamp" timestamp(9) time index,
+  trace_id string,
+  span_id string,
+  parent_span_id string,
+  span_kind string,
+  span_status_code string,
+  service_name string,
+  duration_nano bigint unsigned,
+  "resource_attributes.service.namespace" string,
+  "resource_attributes.service.instance.id" string,
+  primary key (service_name, "resource_attributes.service.namespace",
+    "resource_attributes.service.instance.id")
+) with ('table_data_model' = 'greptime_trace_v1', 'append_mode' = 'true');
+
+create table greptime_otel_resource_info (
+  greptime_timestamp timestamp(3) time index,
+  job string,
+  instance string,
+  "service.name" string,
+  "service.namespace" string,
+  greptime_value double,
+  primary key (job, instance, "service.name", "service.namespace")
+) with (
+  'greptime.semantic.signal_type' = 'metric',
+  'greptime.semantic.source' = 'opentelemetry',
+  'greptime.semantic.metric.type' = 'info'
+);
+
+insert into graph_xsignal_traces values
+  (now(), 't1', 's1', NULL, 'SPAN_KIND_SERVER', 'STATUS_CODE_UNSET', 'api', 100, 'shop', 'inst-1'),
+  (now(), 't2', 's2', NULL, 'SPAN_KIND_SERVER', 'STATUS_CODE_UNSET', 'worker', 100, '', 'inst-2');
+
+insert into greptime_otel_resource_info values
+  (now(), 'shop/api', 'inst-1', 'api', 'shop', 1),
+  (now(), 'worker', 'inst-2', 'worker', '', 1);
+
+-- SQLNESS PROTOCOL MYSQL
+select entity_type, entity_id, source_tables
+from greptime_private.semantic_entities
+where entity_type in ('service', 'service.instance')
+order by entity_type, entity_id, source_tables;
+
+drop table graph_xsignal_traces;
+
+drop table greptime_otel_resource_info;

tests/conf/frontend-test.toml.template

@@ -7,3 +7,6 @@ parallelism = 4
 
 [event_recorder]
 event_types = []
+
+[otlp]
+experimental_enable_resource_info = true

tests/conf/standalone-test.toml.template

@@ -53,3 +53,6 @@ retry_delay = "500ms"
 
 [event_recorder]
 event_types = []
+
+[otlp]
+experimental_enable_resource_info = true