catalog/system_schema/information_schema/
table_semantics.rs

1// Copyright 2023 Greptime Team
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15//! `information_schema.table_semantics`: the queryable view over the table
16//! semantic layer. One row per table that is part of it, so a consumer can
17//! discover the observability concept a table stands for with a single SQL
18//! query instead of parsing every table's `create_options`.
19//!
20//! The few signal-agnostic keys are promoted to their own columns
21//! (`signal_type` / `source` / `source_version` / `pipeline` /
22//! `metadata_quality`); the remaining signal-specific keys are folded into a
23//! `semantic_options` JSON string, keyed by the option name with the
24//! `greptime.semantic.` prefix stripped.
25//!
26//! `entity_declarations` reports the entities the table contributes to the
27//! graph, including the ones the conventions derive with no option set — so a
28//! table with no semantic option still gets a row. It shows the outcome, not
29//! the reasoning: a declaration dropped for naming a missing column is simply
30//! absent, and only the log says why.
31
32use std::collections::BTreeMap;
33use std::sync::{Arc, Weak};
34
35use arrow_schema::SchemaRef as ArrowSchemaRef;
36use common_catalog::consts::INFORMATION_SCHEMA_TABLE_SEMANTICS_TABLE_ID;
37use common_error::ext::BoxedError;
38use common_recordbatch::adapter::RecordBatchStreamAdapter;
39use common_recordbatch::{RecordBatch, SendableRecordBatchStream};
40use datafusion::execution::TaskContext;
41use datafusion::physical_plan::SendableRecordBatchStream as DfSendableRecordBatchStream;
42use datafusion::physical_plan::stream::RecordBatchStreamAdapter as DfRecordBatchStreamAdapter;
43use datafusion::physical_plan::streaming::PartitionStream as DfPartitionStream;
44use datatypes::prelude::{ConcreteDataType, ScalarVectorBuilder, VectorRef};
45use datatypes::schema::{ColumnSchema, Schema, SchemaRef};
46use datatypes::value::Value;
47use datatypes::vectors::{StringVectorBuilder, UInt32VectorBuilder};
48use futures::TryStreamExt;
49use serde::Serialize;
50use snafu::{OptionExt, ResultExt};
51use store_api::storage::{ScanRequest, TableId};
52use table::metadata::TableInfo;
53use table::requests::{
54    SEMANTIC_METRIC_METADATA_QUALITY, SEMANTIC_PIPELINE, SEMANTIC_PREFIX, SEMANTIC_SIGNAL_TYPE,
55    SEMANTIC_SOURCE, SEMANTIC_SOURCE_VERSION, is_semantic_option_key,
56};
57
58use crate::CatalogManager;
59use crate::error::{
60    CreateRecordBatchSnafu, InternalSnafu, Result, UpgradeWeakCatalogManagerRefSnafu,
61};
62use crate::system_schema::information_schema::{InformationTable, Predicates, TABLE_SEMANTICS};
63use crate::system_schema::semantic_graph::{EntityGraphProviderRef, TableEntityDeclaration};
64use crate::system_schema::utils;
65
66pub const TABLE_CATALOG: &str = "table_catalog";
67pub const TABLE_SCHEMA: &str = "table_schema";
68pub const TABLE_NAME: &str = "table_name";
69pub const TABLE_ID: &str = "table_id";
70pub const SIGNAL_TYPE: &str = "signal_type";
71pub const SOURCE: &str = "source";
72pub const SOURCE_VERSION: &str = "source_version";
73pub const PIPELINE: &str = "pipeline";
74pub const METADATA_QUALITY: &str = "metadata_quality";
75pub const SEMANTIC_OPTIONS: &str = "semantic_options";
76pub const ENTITY_DECLARATIONS: &str = "entity_declarations";
77
78const INIT_CAPACITY: usize = 42;
79
80fn optional_value(v: Option<&str>) -> Value {
81    v.map(Value::from).unwrap_or(Value::Null)
82}
83
84/// The JSON form of one declaration in the `entity_declarations` column.
85#[derive(Serialize)]
86struct DeclarationEntry {
87    entity_type: String,
88    origin: &'static str,
89    id: Vec<String>,
90    #[serde(skip_serializing_if = "Option::is_none")]
91    id_qualifier: Option<String>,
92    #[serde(skip_serializing_if = "Vec::is_empty")]
93    superseded_by: Vec<String>,
94    #[serde(skip_serializing_if = "Vec::is_empty")]
95    descriptive: Vec<String>,
96    #[serde(skip_serializing_if = "Vec::is_empty")]
97    scope: Vec<String>,
98}
99
100impl From<TableEntityDeclaration> for DeclarationEntry {
101    fn from(declaration: TableEntityDeclaration) -> Self {
102        Self {
103            entity_type: declaration.entity_type,
104            origin: declaration.origin.as_str(),
105            id: declaration.id_columns,
106            id_qualifier: declaration.id_qualifier,
107            superseded_by: declaration.superseded_by_columns,
108            descriptive: declaration.descriptive_columns,
109            scope: declaration.scope_columns,
110        }
111    }
112}
113
114fn entity_declarations_json(declarations: Vec<TableEntityDeclaration>) -> Option<String> {
115    if declarations.is_empty() {
116        return None;
117    }
118    let entries: Vec<DeclarationEntry> = declarations.into_iter().map(Into::into).collect();
119    // Fold a failure into `None` rather than panicking the query path, as the
120    // options tail does.
121    serde_json::to_string(&entries).ok()
122}
123
124/// The semantic projection of a single table: the signal-agnostic keys promoted
125/// to columns, plus a JSON tail for the rest. Borrows from the table's options.
126#[derive(Default)]
127struct SemanticRow<'a> {
128    signal_type: Option<&'a str>,
129    source: Option<&'a str>,
130    source_version: Option<&'a str>,
131    pipeline: Option<&'a str>,
132    metadata_quality: Option<&'a str>,
133    options_json: Option<String>,
134}
135
136impl<'a> SemanticRow<'a> {
137    /// Projects a table's options onto the semantic schema, or `None` when the
138    /// table carries no semantic key at all.
139    fn extract(table_info: &'a TableInfo) -> Option<Self> {
140        let mut signal_type = None;
141        let mut source = None;
142        let mut source_version = None;
143        let mut pipeline = None;
144        let mut metadata_quality = None;
145        let mut rest = BTreeMap::new();
146
147        for (key, value) in &table_info.meta.options.extra_options {
148            if !is_semantic_option_key(key) {
149                continue;
150            }
151            match key.as_str() {
152                SEMANTIC_SIGNAL_TYPE => signal_type = Some(value.as_str()),
153                SEMANTIC_SOURCE => source = Some(value.as_str()),
154                SEMANTIC_SOURCE_VERSION => source_version = Some(value.as_str()),
155                SEMANTIC_PIPELINE => pipeline = Some(value.as_str()),
156                SEMANTIC_METRIC_METADATA_QUALITY => metadata_quality = Some(value.as_str()),
157                _ => {
158                    let short = key.strip_prefix(SEMANTIC_PREFIX).unwrap_or(key);
159                    rest.insert(short, value.as_str());
160                }
161            }
162        }
163
164        let has_any = signal_type.is_some()
165            || source.is_some()
166            || source_version.is_some()
167            || pipeline.is_some()
168            || metadata_quality.is_some()
169            || !rest.is_empty();
170        if !has_any {
171            return None;
172        }
173
174        // `rest` is a `BTreeMap`, so the JSON keys come out sorted and the output
175        // is stable across runs. Serializing a string map can't realistically fail,
176        // but fold a failure into `None` rather than panicking the query path.
177        let options_json = (!rest.is_empty())
178            .then(|| serde_json::to_string(&rest).ok())
179            .flatten();
180
181        Some(Self {
182            signal_type,
183            source,
184            source_version,
185            pipeline,
186            metadata_quality,
187            options_json,
188        })
189    }
190}
191
192#[derive(Debug)]
193pub(super) struct InformationSchemaTableSemantics {
194    schema: SchemaRef,
195    catalog_name: String,
196    catalog_manager: Weak<dyn CatalogManager>,
197}
198
199impl InformationSchemaTableSemantics {
200    pub(super) fn new(catalog_name: String, catalog_manager: Weak<dyn CatalogManager>) -> Self {
201        Self {
202            schema: Self::schema(),
203            catalog_name,
204            catalog_manager,
205        }
206    }
207
208    fn schema() -> SchemaRef {
209        Arc::new(Schema::new(vec![
210            ColumnSchema::new(TABLE_CATALOG, ConcreteDataType::string_datatype(), false),
211            ColumnSchema::new(TABLE_SCHEMA, ConcreteDataType::string_datatype(), false),
212            ColumnSchema::new(TABLE_NAME, ConcreteDataType::string_datatype(), false),
213            ColumnSchema::new(TABLE_ID, ConcreteDataType::uint32_datatype(), false),
214            ColumnSchema::new(SIGNAL_TYPE, ConcreteDataType::string_datatype(), true),
215            ColumnSchema::new(SOURCE, ConcreteDataType::string_datatype(), true),
216            ColumnSchema::new(SOURCE_VERSION, ConcreteDataType::string_datatype(), true),
217            ColumnSchema::new(PIPELINE, ConcreteDataType::string_datatype(), true),
218            ColumnSchema::new(METADATA_QUALITY, ConcreteDataType::string_datatype(), true),
219            ColumnSchema::new(SEMANTIC_OPTIONS, ConcreteDataType::string_datatype(), true),
220            ColumnSchema::new(
221                ENTITY_DECLARATIONS,
222                ConcreteDataType::string_datatype(),
223                true,
224            ),
225        ]))
226    }
227
228    fn builder(&self) -> InformationSchemaSemanticTablesBuilder {
229        InformationSchemaSemanticTablesBuilder::new(
230            self.schema.clone(),
231            self.catalog_name.clone(),
232            self.catalog_manager.clone(),
233        )
234    }
235}
236
237impl InformationTable for InformationSchemaTableSemantics {
238    fn table_id(&self) -> TableId {
239        INFORMATION_SCHEMA_TABLE_SEMANTICS_TABLE_ID
240    }
241
242    fn table_name(&self) -> &'static str {
243        TABLE_SEMANTICS
244    }
245
246    fn schema(&self) -> SchemaRef {
247        self.schema.clone()
248    }
249
250    fn to_stream(&self, request: ScanRequest) -> Result<SendableRecordBatchStream> {
251        let schema = self.schema.arrow_schema().clone();
252        let mut builder = self.builder();
253        let stream = Box::pin(DfRecordBatchStreamAdapter::new(
254            schema,
255            futures::stream::once(async move {
256                builder
257                    .make_tables(Some(request))
258                    .await
259                    .map(|x| x.into_df_record_batch())
260                    .map_err(|err| datafusion::error::DataFusionError::External(Box::new(err)))
261            }),
262        ));
263        Ok(Box::pin(
264            RecordBatchStreamAdapter::try_new(stream)
265                .map_err(BoxedError::new)
266                .context(InternalSnafu)?,
267        ))
268    }
269}
270
271struct InformationSchemaSemanticTablesBuilder {
272    schema: SchemaRef,
273    catalog_name: String,
274    catalog_manager: Weak<dyn CatalogManager>,
275
276    catalog_names: StringVectorBuilder,
277    schema_names: StringVectorBuilder,
278    table_names: StringVectorBuilder,
279    table_ids: UInt32VectorBuilder,
280    signal_types: StringVectorBuilder,
281    sources: StringVectorBuilder,
282    source_versions: StringVectorBuilder,
283    pipelines: StringVectorBuilder,
284    metadata_qualities: StringVectorBuilder,
285    semantic_options: StringVectorBuilder,
286    entity_declarations: StringVectorBuilder,
287}
288
289impl InformationSchemaSemanticTablesBuilder {
290    fn new(
291        schema: SchemaRef,
292        catalog_name: String,
293        catalog_manager: Weak<dyn CatalogManager>,
294    ) -> Self {
295        Self {
296            schema,
297            catalog_name,
298            catalog_manager,
299            catalog_names: StringVectorBuilder::with_capacity(INIT_CAPACITY),
300            schema_names: StringVectorBuilder::with_capacity(INIT_CAPACITY),
301            table_names: StringVectorBuilder::with_capacity(INIT_CAPACITY),
302            table_ids: UInt32VectorBuilder::with_capacity(INIT_CAPACITY),
303            signal_types: StringVectorBuilder::with_capacity(INIT_CAPACITY),
304            sources: StringVectorBuilder::with_capacity(INIT_CAPACITY),
305            source_versions: StringVectorBuilder::with_capacity(INIT_CAPACITY),
306            pipelines: StringVectorBuilder::with_capacity(INIT_CAPACITY),
307            metadata_qualities: StringVectorBuilder::with_capacity(INIT_CAPACITY),
308            semantic_options: StringVectorBuilder::with_capacity(INIT_CAPACITY),
309            entity_declarations: StringVectorBuilder::with_capacity(INIT_CAPACITY),
310        }
311    }
312
313    async fn make_tables(&mut self, request: Option<ScanRequest>) -> Result<RecordBatch> {
314        let catalog_name = self.catalog_name.clone();
315        let catalog_manager = self
316            .catalog_manager
317            .upgrade()
318            .context(UpgradeWeakCatalogManagerRefSnafu)?;
319        let predicates = Predicates::from_scan_request(&request);
320        // Resolved once for the whole scan: the lookup downcasts the catalog
321        // manager, while the per-table call behind it is pure metadata work.
322        let graph_provider = utils::entity_graph_provider(&self.catalog_manager)?;
323
324        for schema_name in catalog_manager.schema_names(&catalog_name, None).await? {
325            let mut table_stream = catalog_manager.tables(&catalog_name, &schema_name, None);
326            while let Some(table) = table_stream.try_next().await? {
327                self.add_table(
328                    &predicates,
329                    &catalog_name,
330                    &schema_name,
331                    table.table_info(),
332                    graph_provider.as_ref(),
333                );
334            }
335        }
336
337        self.finish()
338    }
339
340    fn add_table(
341        &mut self,
342        predicates: &Predicates,
343        catalog_name: &str,
344        schema_name: &str,
345        table_info: Arc<TableInfo>,
346        graph_provider: Option<&EntityGraphProviderRef>,
347    ) {
348        let semantic_row = SemanticRow::extract(&table_info);
349        let carries_options = semantic_row.is_some();
350        let row = semantic_row.unwrap_or_default();
351
352        let table_name = table_info.name.as_ref();
353        let catalog_v = Value::from(catalog_name);
354        let schema_v = Value::from(schema_name);
355        let name_v = Value::from(table_name);
356        let signal_v = optional_value(row.signal_type);
357        let source_v = optional_value(row.source);
358        let source_version_v = optional_value(row.source_version);
359        let pipeline_v = optional_value(row.pipeline);
360        let quality_v = optional_value(row.metadata_quality);
361        let predicate_row = [
362            (TABLE_CATALOG, &catalog_v),
363            (TABLE_SCHEMA, &schema_v),
364            (TABLE_NAME, &name_v),
365            (SIGNAL_TYPE, &signal_v),
366            (SOURCE, &source_v),
367            (SOURCE_VERSION, &source_version_v),
368            (PIPELINE, &pipeline_v),
369            (METADATA_QUALITY, &quality_v),
370        ];
371        if !predicates.eval(&predicate_row) {
372            return;
373        }
374
375        let declarations_json = graph_provider
376            .map(|provider| provider.table_declarations(&table_info))
377            .and_then(entity_declarations_json);
378        if !carries_options && declarations_json.is_none() {
379            return;
380        }
381
382        self.catalog_names.push(Some(catalog_name));
383        self.schema_names.push(Some(schema_name));
384        self.table_names.push(Some(table_name));
385        self.table_ids.push(Some(table_info.table_id()));
386        self.signal_types.push(row.signal_type);
387        self.sources.push(row.source);
388        self.source_versions.push(row.source_version);
389        self.pipelines.push(row.pipeline);
390        self.metadata_qualities.push(row.metadata_quality);
391        self.semantic_options.push(row.options_json.as_deref());
392        self.entity_declarations.push(declarations_json.as_deref());
393    }
394
395    fn finish(&mut self) -> Result<RecordBatch> {
396        let columns: Vec<VectorRef> = vec![
397            Arc::new(self.catalog_names.finish()),
398            Arc::new(self.schema_names.finish()),
399            Arc::new(self.table_names.finish()),
400            Arc::new(self.table_ids.finish()),
401            Arc::new(self.signal_types.finish()),
402            Arc::new(self.sources.finish()),
403            Arc::new(self.source_versions.finish()),
404            Arc::new(self.pipelines.finish()),
405            Arc::new(self.metadata_qualities.finish()),
406            Arc::new(self.semantic_options.finish()),
407            Arc::new(self.entity_declarations.finish()),
408        ];
409        RecordBatch::new(self.schema.clone(), columns).context(CreateRecordBatchSnafu)
410    }
411}
412
413impl DfPartitionStream for InformationSchemaTableSemantics {
414    fn schema(&self) -> &ArrowSchemaRef {
415        self.schema.arrow_schema()
416    }
417
418    fn execute(&self, _: Arc<TaskContext>) -> DfSendableRecordBatchStream {
419        let schema = self.schema.arrow_schema().clone();
420        let mut builder = self.builder();
421        Box::pin(DfRecordBatchStreamAdapter::new(
422            schema,
423            futures::stream::once(async move {
424                builder
425                    .make_tables(None)
426                    .await
427                    .map(|x| x.into_df_record_batch())
428                    .map_err(Into::into)
429            }),
430        ))
431    }
432}
433
434#[cfg(test)]
435mod tests {
436    use std::collections::HashMap;
437
438    use common_catalog::consts::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, MITO_ENGINE};
439    use datatypes::schema::SchemaBuilder;
440    use table::metadata::{TableInfoBuilder, TableMeta, TableType};
441    use table::requests::{
442        SEMANTIC_METRIC_TYPE, SEMANTIC_METRIC_UNIT, SEMANTIC_SOURCE_VERSION, TableOptions,
443    };
444
445    use super::*;
446
447    fn table_info(extra: &[(&str, &str)]) -> TableInfo {
448        let schema = Arc::new(
449            SchemaBuilder::try_from_columns(vec![ColumnSchema::new(
450                "ts",
451                ConcreteDataType::timestamp_millisecond_datatype(),
452                false,
453            )])
454            .unwrap()
455            .build()
456            .unwrap(),
457        );
458        let options = TableOptions {
459            extra_options: extra
460                .iter()
461                .map(|(k, v)| (k.to_string(), v.to_string()))
462                .collect::<HashMap<_, _>>(),
463            ..Default::default()
464        };
465        let meta = TableMeta {
466            schema,
467            primary_key_indices: vec![],
468            value_indices: vec![],
469            engine: MITO_ENGINE.to_string(),
470            next_column_id: 1,
471            options,
472            created_on: Default::default(),
473            updated_on: Default::default(),
474            partition_key_indices: vec![],
475            column_ids: vec![],
476        };
477        TableInfoBuilder::default()
478            .table_id(1)
479            .name("t")
480            .catalog_name(DEFAULT_CATALOG_NAME)
481            .schema_name(DEFAULT_SCHEMA_NAME)
482            .table_version(0)
483            .table_type(TableType::Base)
484            .meta(meta)
485            .build()
486            .unwrap()
487    }
488
489    #[test]
490    fn extract_promotes_core_keys_and_folds_the_rest() {
491        let info = table_info(&[
492            (SEMANTIC_SIGNAL_TYPE, "metric"),
493            (SEMANTIC_SOURCE, "opentelemetry"),
494            (SEMANTIC_SOURCE_VERSION, "2.0"),
495            (SEMANTIC_PIPELINE, "greptime_metric_v1"),
496            (SEMANTIC_METRIC_METADATA_QUALITY, "declared"),
497            (SEMANTIC_METRIC_TYPE, "counter"),
498            (SEMANTIC_METRIC_UNIT, "By"),
499            // A non-semantic option must be ignored entirely.
500            ("ttl", "7d"),
501        ]);
502
503        let row = SemanticRow::extract(&info).unwrap();
504        assert_eq!(row.signal_type, Some("metric"));
505        assert_eq!(row.source, Some("opentelemetry"));
506        assert_eq!(row.source_version, Some("2.0"));
507        assert_eq!(row.pipeline, Some("greptime_metric_v1"));
508        assert_eq!(row.metadata_quality, Some("declared"));
509        // Promoted keys stay out of the JSON tail; remaining keys are sorted and
510        // prefix-stripped.
511        assert_eq!(
512            row.options_json.as_deref(),
513            Some(r#"{"metric.type":"counter","metric.unit":"By"}"#)
514        );
515    }
516
517    #[test]
518    fn extract_skips_untagged_table() {
519        let info = table_info(&[("ttl", "7d")]);
520        assert!(SemanticRow::extract(&info).is_none());
521    }
522
523    #[test]
524    fn extract_omits_json_when_only_core_keys_present() {
525        let info = table_info(&[(SEMANTIC_SIGNAL_TYPE, "log")]);
526        let row = SemanticRow::extract(&info).unwrap();
527        assert_eq!(row.signal_type, Some("log"));
528        assert!(row.source.is_none());
529        assert!(row.options_json.is_none());
530    }
531
532    #[test]
533    fn extract_folds_entity_keys_into_json_tail() {
534        // Entity keys are not promoted columns; they surface verbatim (prefix-
535        // stripped, sorted) in the JSON tail with no code change to this table.
536        let info = table_info(&[
537            (SEMANTIC_SIGNAL_TYPE, "trace"),
538            ("greptime.semantic.entity.service.id", "service_name"),
539            ("greptime.semantic.entity.host.id", "host_id"),
540        ]);
541        let row = SemanticRow::extract(&info).unwrap();
542        assert_eq!(row.signal_type, Some("trace"));
543        assert_eq!(
544            row.options_json.as_deref(),
545            Some(r#"{"entity.host.id":"host_id","entity.service.id":"service_name"}"#)
546        );
547    }
548}
catalog/system_schema/information_schema/table_semantics.rs

catalog/system_schema/information_schema/
table_semantics.rs
