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 carries at least one
17//! `greptime.semantic.*` option, so a consumer can discover the observability
18//! concept a table stands for with a single SQL query instead of parsing every
19//! table's `create_options`.
20//!
21//! The few signal-agnostic keys are promoted to their own columns
22//! (`signal_type` / `source` / `source_version` / `pipeline` /
23//! `metadata_quality`); the remaining signal-specific keys are folded into a
24//! `semantic_options` JSON string, keyed by the option name with the
25//! `greptime.semantic.` prefix stripped.
26
27use std::collections::BTreeMap;
28use std::sync::{Arc, Weak};
29
30use arrow_schema::SchemaRef as ArrowSchemaRef;
31use common_catalog::consts::INFORMATION_SCHEMA_TABLE_SEMANTICS_TABLE_ID;
32use common_error::ext::BoxedError;
33use common_recordbatch::adapter::RecordBatchStreamAdapter;
34use common_recordbatch::{RecordBatch, SendableRecordBatchStream};
35use datafusion::execution::TaskContext;
36use datafusion::physical_plan::SendableRecordBatchStream as DfSendableRecordBatchStream;
37use datafusion::physical_plan::stream::RecordBatchStreamAdapter as DfRecordBatchStreamAdapter;
38use datafusion::physical_plan::streaming::PartitionStream as DfPartitionStream;
39use datatypes::prelude::{ConcreteDataType, ScalarVectorBuilder, VectorRef};
40use datatypes::schema::{ColumnSchema, Schema, SchemaRef};
41use datatypes::value::Value;
42use datatypes::vectors::{StringVectorBuilder, UInt32VectorBuilder};
43use futures::TryStreamExt;
44use snafu::{OptionExt, ResultExt};
45use store_api::storage::{ScanRequest, TableId};
46use table::metadata::TableInfo;
47use table::requests::{
48    SEMANTIC_METRIC_METADATA_QUALITY, SEMANTIC_PIPELINE, SEMANTIC_PREFIX, SEMANTIC_SIGNAL_TYPE,
49    SEMANTIC_SOURCE, SEMANTIC_SOURCE_VERSION, is_semantic_option_key,
50};
51
52use crate::CatalogManager;
53use crate::error::{
54    CreateRecordBatchSnafu, InternalSnafu, Result, UpgradeWeakCatalogManagerRefSnafu,
55};
56use crate::system_schema::information_schema::{InformationTable, Predicates, TABLE_SEMANTICS};
57
58pub const TABLE_CATALOG: &str = "table_catalog";
59pub const TABLE_SCHEMA: &str = "table_schema";
60pub const TABLE_NAME: &str = "table_name";
61pub const TABLE_ID: &str = "table_id";
62pub const SIGNAL_TYPE: &str = "signal_type";
63pub const SOURCE: &str = "source";
64pub const SOURCE_VERSION: &str = "source_version";
65pub const PIPELINE: &str = "pipeline";
66pub const METADATA_QUALITY: &str = "metadata_quality";
67pub const SEMANTIC_OPTIONS: &str = "semantic_options";
68
69const INIT_CAPACITY: usize = 42;
70
71fn optional_value(v: Option<&str>) -> Value {
72    v.map(Value::from).unwrap_or(Value::Null)
73}
74
75/// The semantic projection of a single table: the signal-agnostic keys promoted
76/// to columns, plus a JSON tail for the rest. Borrows from the table's options.
77struct SemanticRow<'a> {
78    signal_type: Option<&'a str>,
79    source: Option<&'a str>,
80    source_version: Option<&'a str>,
81    pipeline: Option<&'a str>,
82    metadata_quality: Option<&'a str>,
83    options_json: Option<String>,
84}
85
86impl<'a> SemanticRow<'a> {
87    /// Projects a table's options onto the semantic schema, or `None` when the
88    /// table carries no semantic key at all.
89    fn extract(table_info: &'a TableInfo) -> Option<Self> {
90        let mut signal_type = None;
91        let mut source = None;
92        let mut source_version = None;
93        let mut pipeline = None;
94        let mut metadata_quality = None;
95        let mut rest = BTreeMap::new();
96
97        for (key, value) in &table_info.meta.options.extra_options {
98            if !is_semantic_option_key(key) {
99                continue;
100            }
101            match key.as_str() {
102                SEMANTIC_SIGNAL_TYPE => signal_type = Some(value.as_str()),
103                SEMANTIC_SOURCE => source = Some(value.as_str()),
104                SEMANTIC_SOURCE_VERSION => source_version = Some(value.as_str()),
105                SEMANTIC_PIPELINE => pipeline = Some(value.as_str()),
106                SEMANTIC_METRIC_METADATA_QUALITY => metadata_quality = Some(value.as_str()),
107                _ => {
108                    let short = key.strip_prefix(SEMANTIC_PREFIX).unwrap_or(key);
109                    rest.insert(short, value.as_str());
110                }
111            }
112        }
113
114        let has_any = signal_type.is_some()
115            || source.is_some()
116            || source_version.is_some()
117            || pipeline.is_some()
118            || metadata_quality.is_some()
119            || !rest.is_empty();
120        if !has_any {
121            return None;
122        }
123
124        // `rest` is a `BTreeMap`, so the JSON keys come out sorted and the output
125        // is stable across runs. Serializing a string map can't realistically fail,
126        // but fold a failure into `None` rather than panicking the query path.
127        let options_json = (!rest.is_empty())
128            .then(|| serde_json::to_string(&rest).ok())
129            .flatten();
130
131        Some(Self {
132            signal_type,
133            source,
134            source_version,
135            pipeline,
136            metadata_quality,
137            options_json,
138        })
139    }
140}
141
142#[derive(Debug)]
143pub(super) struct InformationSchemaTableSemantics {
144    schema: SchemaRef,
145    catalog_name: String,
146    catalog_manager: Weak<dyn CatalogManager>,
147}
148
149impl InformationSchemaTableSemantics {
150    pub(super) fn new(catalog_name: String, catalog_manager: Weak<dyn CatalogManager>) -> Self {
151        Self {
152            schema: Self::schema(),
153            catalog_name,
154            catalog_manager,
155        }
156    }
157
158    fn schema() -> SchemaRef {
159        Arc::new(Schema::new(vec![
160            ColumnSchema::new(TABLE_CATALOG, ConcreteDataType::string_datatype(), false),
161            ColumnSchema::new(TABLE_SCHEMA, ConcreteDataType::string_datatype(), false),
162            ColumnSchema::new(TABLE_NAME, ConcreteDataType::string_datatype(), false),
163            ColumnSchema::new(TABLE_ID, ConcreteDataType::uint32_datatype(), false),
164            ColumnSchema::new(SIGNAL_TYPE, ConcreteDataType::string_datatype(), true),
165            ColumnSchema::new(SOURCE, ConcreteDataType::string_datatype(), true),
166            ColumnSchema::new(SOURCE_VERSION, ConcreteDataType::string_datatype(), true),
167            ColumnSchema::new(PIPELINE, ConcreteDataType::string_datatype(), true),
168            ColumnSchema::new(METADATA_QUALITY, ConcreteDataType::string_datatype(), true),
169            ColumnSchema::new(SEMANTIC_OPTIONS, ConcreteDataType::string_datatype(), true),
170        ]))
171    }
172
173    fn builder(&self) -> InformationSchemaSemanticTablesBuilder {
174        InformationSchemaSemanticTablesBuilder::new(
175            self.schema.clone(),
176            self.catalog_name.clone(),
177            self.catalog_manager.clone(),
178        )
179    }
180}
181
182impl InformationTable for InformationSchemaTableSemantics {
183    fn table_id(&self) -> TableId {
184        INFORMATION_SCHEMA_TABLE_SEMANTICS_TABLE_ID
185    }
186
187    fn table_name(&self) -> &'static str {
188        TABLE_SEMANTICS
189    }
190
191    fn schema(&self) -> SchemaRef {
192        self.schema.clone()
193    }
194
195    fn to_stream(&self, request: ScanRequest) -> Result<SendableRecordBatchStream> {
196        let schema = self.schema.arrow_schema().clone();
197        let mut builder = self.builder();
198        let stream = Box::pin(DfRecordBatchStreamAdapter::new(
199            schema,
200            futures::stream::once(async move {
201                builder
202                    .make_tables(Some(request))
203                    .await
204                    .map(|x| x.into_df_record_batch())
205                    .map_err(|err| datafusion::error::DataFusionError::External(Box::new(err)))
206            }),
207        ));
208        Ok(Box::pin(
209            RecordBatchStreamAdapter::try_new(stream)
210                .map_err(BoxedError::new)
211                .context(InternalSnafu)?,
212        ))
213    }
214}
215
216struct InformationSchemaSemanticTablesBuilder {
217    schema: SchemaRef,
218    catalog_name: String,
219    catalog_manager: Weak<dyn CatalogManager>,
220
221    catalog_names: StringVectorBuilder,
222    schema_names: StringVectorBuilder,
223    table_names: StringVectorBuilder,
224    table_ids: UInt32VectorBuilder,
225    signal_types: StringVectorBuilder,
226    sources: StringVectorBuilder,
227    source_versions: StringVectorBuilder,
228    pipelines: StringVectorBuilder,
229    metadata_qualities: StringVectorBuilder,
230    semantic_options: StringVectorBuilder,
231}
232
233impl InformationSchemaSemanticTablesBuilder {
234    fn new(
235        schema: SchemaRef,
236        catalog_name: String,
237        catalog_manager: Weak<dyn CatalogManager>,
238    ) -> Self {
239        Self {
240            schema,
241            catalog_name,
242            catalog_manager,
243            catalog_names: StringVectorBuilder::with_capacity(INIT_CAPACITY),
244            schema_names: StringVectorBuilder::with_capacity(INIT_CAPACITY),
245            table_names: StringVectorBuilder::with_capacity(INIT_CAPACITY),
246            table_ids: UInt32VectorBuilder::with_capacity(INIT_CAPACITY),
247            signal_types: StringVectorBuilder::with_capacity(INIT_CAPACITY),
248            sources: StringVectorBuilder::with_capacity(INIT_CAPACITY),
249            source_versions: StringVectorBuilder::with_capacity(INIT_CAPACITY),
250            pipelines: StringVectorBuilder::with_capacity(INIT_CAPACITY),
251            metadata_qualities: StringVectorBuilder::with_capacity(INIT_CAPACITY),
252            semantic_options: StringVectorBuilder::with_capacity(INIT_CAPACITY),
253        }
254    }
255
256    async fn make_tables(&mut self, request: Option<ScanRequest>) -> Result<RecordBatch> {
257        let catalog_name = self.catalog_name.clone();
258        let catalog_manager = self
259            .catalog_manager
260            .upgrade()
261            .context(UpgradeWeakCatalogManagerRefSnafu)?;
262        let predicates = Predicates::from_scan_request(&request);
263
264        for schema_name in catalog_manager.schema_names(&catalog_name, None).await? {
265            let mut table_stream = catalog_manager.tables(&catalog_name, &schema_name, None);
266            while let Some(table) = table_stream.try_next().await? {
267                self.add_table(&predicates, &catalog_name, &schema_name, table.table_info());
268            }
269        }
270
271        self.finish()
272    }
273
274    fn add_table(
275        &mut self,
276        predicates: &Predicates,
277        catalog_name: &str,
278        schema_name: &str,
279        table_info: Arc<TableInfo>,
280    ) {
281        // A table with no semantic key is not part of the semantic layer.
282        let Some(row) = SemanticRow::extract(&table_info) else {
283            return;
284        };
285
286        let table_name = table_info.name.as_ref();
287        let catalog_v = Value::from(catalog_name);
288        let schema_v = Value::from(schema_name);
289        let name_v = Value::from(table_name);
290        let signal_v = optional_value(row.signal_type);
291        let source_v = optional_value(row.source);
292        let source_version_v = optional_value(row.source_version);
293        let pipeline_v = optional_value(row.pipeline);
294        let quality_v = optional_value(row.metadata_quality);
295        let predicate_row = [
296            (TABLE_CATALOG, &catalog_v),
297            (TABLE_SCHEMA, &schema_v),
298            (TABLE_NAME, &name_v),
299            (SIGNAL_TYPE, &signal_v),
300            (SOURCE, &source_v),
301            (SOURCE_VERSION, &source_version_v),
302            (PIPELINE, &pipeline_v),
303            (METADATA_QUALITY, &quality_v),
304        ];
305        if !predicates.eval(&predicate_row) {
306            return;
307        }
308
309        self.catalog_names.push(Some(catalog_name));
310        self.schema_names.push(Some(schema_name));
311        self.table_names.push(Some(table_name));
312        self.table_ids.push(Some(table_info.table_id()));
313        self.signal_types.push(row.signal_type);
314        self.sources.push(row.source);
315        self.source_versions.push(row.source_version);
316        self.pipelines.push(row.pipeline);
317        self.metadata_qualities.push(row.metadata_quality);
318        self.semantic_options.push(row.options_json.as_deref());
319    }
320
321    fn finish(&mut self) -> Result<RecordBatch> {
322        let columns: Vec<VectorRef> = vec![
323            Arc::new(self.catalog_names.finish()),
324            Arc::new(self.schema_names.finish()),
325            Arc::new(self.table_names.finish()),
326            Arc::new(self.table_ids.finish()),
327            Arc::new(self.signal_types.finish()),
328            Arc::new(self.sources.finish()),
329            Arc::new(self.source_versions.finish()),
330            Arc::new(self.pipelines.finish()),
331            Arc::new(self.metadata_qualities.finish()),
332            Arc::new(self.semantic_options.finish()),
333        ];
334        RecordBatch::new(self.schema.clone(), columns).context(CreateRecordBatchSnafu)
335    }
336}
337
338impl DfPartitionStream for InformationSchemaTableSemantics {
339    fn schema(&self) -> &ArrowSchemaRef {
340        self.schema.arrow_schema()
341    }
342
343    fn execute(&self, _: Arc<TaskContext>) -> DfSendableRecordBatchStream {
344        let schema = self.schema.arrow_schema().clone();
345        let mut builder = self.builder();
346        Box::pin(DfRecordBatchStreamAdapter::new(
347            schema,
348            futures::stream::once(async move {
349                builder
350                    .make_tables(None)
351                    .await
352                    .map(|x| x.into_df_record_batch())
353                    .map_err(Into::into)
354            }),
355        ))
356    }
357}
358
359#[cfg(test)]
360mod tests {
361    use std::collections::HashMap;
362
363    use common_catalog::consts::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, MITO_ENGINE};
364    use datatypes::schema::SchemaBuilder;
365    use table::metadata::{TableInfoBuilder, TableMeta, TableType};
366    use table::requests::{
367        SEMANTIC_METRIC_TYPE, SEMANTIC_METRIC_UNIT, SEMANTIC_SOURCE_VERSION, TableOptions,
368    };
369
370    use super::*;
371
372    fn table_info(extra: &[(&str, &str)]) -> TableInfo {
373        let schema = Arc::new(
374            SchemaBuilder::try_from_columns(vec![ColumnSchema::new(
375                "ts",
376                ConcreteDataType::timestamp_millisecond_datatype(),
377                false,
378            )])
379            .unwrap()
380            .build()
381            .unwrap(),
382        );
383        let options = TableOptions {
384            extra_options: extra
385                .iter()
386                .map(|(k, v)| (k.to_string(), v.to_string()))
387                .collect::<HashMap<_, _>>(),
388            ..Default::default()
389        };
390        let meta = TableMeta {
391            schema,
392            primary_key_indices: vec![],
393            value_indices: vec![],
394            engine: MITO_ENGINE.to_string(),
395            next_column_id: 1,
396            options,
397            created_on: Default::default(),
398            updated_on: Default::default(),
399            partition_key_indices: vec![],
400            column_ids: vec![],
401        };
402        TableInfoBuilder::default()
403            .table_id(1)
404            .name("t")
405            .catalog_name(DEFAULT_CATALOG_NAME)
406            .schema_name(DEFAULT_SCHEMA_NAME)
407            .table_version(0)
408            .table_type(TableType::Base)
409            .meta(meta)
410            .build()
411            .unwrap()
412    }
413
414    #[test]
415    fn extract_promotes_core_keys_and_folds_the_rest() {
416        let info = table_info(&[
417            (SEMANTIC_SIGNAL_TYPE, "metric"),
418            (SEMANTIC_SOURCE, "opentelemetry"),
419            (SEMANTIC_SOURCE_VERSION, "2.0"),
420            (SEMANTIC_PIPELINE, "greptime_metric_v1"),
421            (SEMANTIC_METRIC_METADATA_QUALITY, "declared"),
422            (SEMANTIC_METRIC_TYPE, "counter"),
423            (SEMANTIC_METRIC_UNIT, "By"),
424            // A non-semantic option must be ignored entirely.
425            ("ttl", "7d"),
426        ]);
427
428        let row = SemanticRow::extract(&info).unwrap();
429        assert_eq!(row.signal_type, Some("metric"));
430        assert_eq!(row.source, Some("opentelemetry"));
431        assert_eq!(row.source_version, Some("2.0"));
432        assert_eq!(row.pipeline, Some("greptime_metric_v1"));
433        assert_eq!(row.metadata_quality, Some("declared"));
434        // Promoted keys stay out of the JSON tail; remaining keys are sorted and
435        // prefix-stripped.
436        assert_eq!(
437            row.options_json.as_deref(),
438            Some(r#"{"metric.type":"counter","metric.unit":"By"}"#)
439        );
440    }
441
442    #[test]
443    fn extract_skips_untagged_table() {
444        let info = table_info(&[("ttl", "7d")]);
445        assert!(SemanticRow::extract(&info).is_none());
446    }
447
448    #[test]
449    fn extract_omits_json_when_only_core_keys_present() {
450        let info = table_info(&[(SEMANTIC_SIGNAL_TYPE, "log")]);
451        let row = SemanticRow::extract(&info).unwrap();
452        assert_eq!(row.signal_type, Some("log"));
453        assert!(row.source.is_none());
454        assert!(row.options_json.is_none());
455    }
456}
catalog/system_schema/information_schema/table_semantics.rs

catalog/system_schema/information_schema/
table_semantics.rs
