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

catalog/system_schema/information_schema/
table_semantics.rs
