mirror of
https://github.com/GreptimeTeam/greptimedb.git
synced 2026-07-06 14:00:40 +00:00
feat: table semantic layer information_schema view (Phase 3) (#8240)
* feat: table semantic layer information_schema view (Phase 3) Add `information_schema.table_semantics`, a queryable view over the table semantic layer. One row per table that carries at least one `greptime.semantic.*` option: the signal-agnostic keys (signal_type/source/pipeline/metadata_quality) are promoted to columns and the remaining signal-specific keys are folded into a `semantic_options` JSON string. Tables with no semantic key are excluded. Stacked on Phase 2. Signed-off-by: Dennis Zhuang <killme2008@gmail.com> * chore: address PR review on table_semantics - fold JSON serialization failure into None instead of unwrap/panic - drop per-row Vec allocation in predicate eval; use a fixed array - align RFC view name with the shipped `table_semantics` Signed-off-by: Dennis Zhuang <killme2008@gmail.com> * chore: update results Signed-off-by: Dennis Zhuang <killme2008@gmail.com> --------- Signed-off-by: Dennis Zhuang <killme2008@gmail.com>
This commit is contained in:
dennis zhuang
GitHub
@@ -9,7 +9,7 @@ Author: "Dennis Zhuang <killme2008@gmail.com>"
|
||||
|
||||
Attach a thin layer of semantic metadata to each table so machine consumers — LLM agents, alert generators, dashboard builders, MCP servers, ETL pipelines — can align it with the observability concepts they already know (OTel instrument kinds, Prometheus naming conventions, UCUM units, semantic conventions, severity numbers, OTel ↔ Prometheus translation rules).
|
||||
|
||||
The mechanism reuses what already exists in `table_options` (the same slot that today carries `table_data_model` and `otlp_metric_compat`): a reserved `greptime.semantic.*` namespace, plus standard SQL column `COMMENT` for field-level supplements, plus an `information_schema.semantic_tables` view as the discovery entry point. No new protocol, no new DDL keyword.
|
||||
The mechanism reuses what already exists in `table_options` (the same slot that today carries `table_data_model` and `otlp_metric_compat`): a reserved `greptime.semantic.*` namespace, plus standard SQL column `COMMENT` for field-level supplements, plus an `information_schema.table_semantics` view as the discovery entry point. No new protocol, no new DDL keyword.
|
||||
|
||||
Per-table identity only. Cross-table relationships are deferred.
|
||||
|
||||
@@ -44,7 +44,7 @@ The audience is broader than LLM agents. Alert generators need to choose between
|
||||
|
||||
1. **`greptime.semantic.*` table options** — table-level identity and lineage. Carried inside the existing `table_options` blob. This is the same slot that today carries `table_data_model = 'greptime_trace_v1'` and `otlp_metric_compat = 'prom'`, so the mechanism is generalising what the OTLP trace auto-create path already does.
|
||||
2. **Column `COMMENT`** — column-level supplements ("this column is `resource.service.name`"; "this column carries delta values"). Standard SQL.
|
||||
3. **`information_schema.semantic_tables` view** — a denormalised projection of the options, registered through the existing `with_extra_table_factories()` hook. Tables without a `greptime.semantic.*` option do not appear in the view.
|
||||
3. **`information_schema.table_semantics` view** — a denormalised projection of the options, registered through the existing `with_extra_table_factories()` hook. Tables without a `greptime.semantic.*` option do not appear in the view.
|
||||
|
||||
## Vocabulary
|
||||
|
||||
@@ -92,13 +92,13 @@ Two design decisions worth pinning down up front, because they constrain everyth
|
||||
- **Conflict.** Some table-level keys (`trace.conventions` lifted from `schema_url`, `metric.temporality`, ...) cannot represent the truth when a long-lived table sees rows from multiple sources. v1 records `mixed` or `unknown` rather than a fictitious single value. Downstream consumers must treat any single-valued semantic key as best-effort, not strong evidence.
|
||||
- **Update.** Semantic options are stamped at table creation. v1 does not specify an update path; promoting `metadata_quality` from `inferred` to `declared`, refreshing `resource.attributes_preserved`, or revising `trace.conventions` on later writes is deferred. If real usage shows update is needed, it lands as a separate RFC.
|
||||
|
||||
## `information_schema.semantic_tables`
|
||||
## `information_schema.table_semantics`
|
||||
|
||||
A consumer's first SQL on connect:
|
||||
|
||||
```sql
|
||||
SELECT table_catalog, table_schema, table_name, signal_type, source, pipeline
|
||||
FROM information_schema.semantic_tables;
|
||||
FROM information_schema.table_semantics;
|
||||
```
|
||||
|
||||
returns one row per semantic-tagged table. The view exposes a stable set of core columns (`table_catalog`, `table_schema`, `table_name`, `signal_type`, `source`, `source_version`, `pipeline`) plus a `semantic_options` JSON column carrying the rest of the `greptime.semantic.*` keys verbatim. Future keys appear inside `semantic_options` without forcing a view-schema change; only widely-used keys are ever promoted to first-class columns.
|
||||
@@ -110,7 +110,7 @@ Four phases, each independently shippable.
|
||||
1. **Identity.** Stamp `signal_type` and `source` on every auto-create path. The OTLP paths already have natural injection points; Prom remote write is the one non-trivial path because metric-engine logical tables share physical storage (see Open Question 2).
|
||||
2. **Metric specifics.** Add type / unit / temporality / monotonic / metadata_quality / original_name at OTel metric and Prom RW ingestion sites; the data is already at hand inside the OTel translator.
|
||||
3. **Resource / scope lineage.** Record what the OTel-to-Prometheus translation kept and dropped.
|
||||
4. **`information_schema.semantic_tables` view + documentation** as a stable user-facing contract.
|
||||
4. **`information_schema.table_semantics` view + documentation** as a stable user-facing contract.
|
||||
|
||||
# Relationship to OpenTelemetry standardisation
|
||||
|
||||
|
||||
@@ -27,6 +27,7 @@ pub mod schemata;
|
||||
mod ssts;
|
||||
mod table_constraints;
|
||||
mod table_names;
|
||||
mod table_semantics;
|
||||
pub mod tables;
|
||||
mod views;
|
||||
|
||||
@@ -72,6 +73,7 @@ use crate::system_schema::information_schema::ssts::{
|
||||
InformationSchemaSstsIndexMeta, InformationSchemaSstsManifest, InformationSchemaSstsStorage,
|
||||
};
|
||||
use crate::system_schema::information_schema::table_constraints::InformationSchemaTableConstraints;
|
||||
use crate::system_schema::information_schema::table_semantics::InformationSchemaTableSemantics;
|
||||
use crate::system_schema::information_schema::tables::InformationSchemaTables;
|
||||
use crate::system_schema::memory_table::MemoryTable;
|
||||
pub(crate) use crate::system_schema::predicate::Predicates;
|
||||
@@ -261,6 +263,10 @@ impl SystemSchemaProviderInner for InformationSchemaProvider {
|
||||
SSTS_INDEX_META => Some(Arc::new(InformationSchemaSstsIndexMeta::new(
|
||||
self.catalog_manager.clone(),
|
||||
)) as _),
|
||||
TABLE_SEMANTICS => Some(Arc::new(InformationSchemaTableSemantics::new(
|
||||
self.catalog_name.clone(),
|
||||
self.catalog_manager.clone(),
|
||||
)) as _),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
@@ -357,6 +363,10 @@ impl InformationSchemaProvider {
|
||||
self.build_table(TABLE_CONSTRAINTS).unwrap(),
|
||||
);
|
||||
tables.insert(FLOWS.to_string(), self.build_table(FLOWS).unwrap());
|
||||
tables.insert(
|
||||
TABLE_SEMANTICS.to_string(),
|
||||
self.build_table(TABLE_SEMANTICS).unwrap(),
|
||||
);
|
||||
if let Some(process_list) = self.build_table(PROCESS_LIST) {
|
||||
tables.insert(PROCESS_LIST.to_string(), process_list);
|
||||
}
|
||||
|
||||
@@ -51,3 +51,4 @@ pub const PROCESS_LIST: &str = "process_list";
|
||||
pub const SSTS_MANIFEST: &str = "ssts_manifest";
|
||||
pub const SSTS_STORAGE: &str = "ssts_storage";
|
||||
pub const SSTS_INDEX_META: &str = "ssts_index_meta";
|
||||
pub const TABLE_SEMANTICS: &str = "table_semantics";
|
||||
|
||||
@@ -0,0 +1,438 @@
|
||||
// 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.
|
||||
|
||||
//! `information_schema.table_semantics`: the queryable view over the table
|
||||
//! semantic layer. One row per table that carries at least one
|
||||
//! `greptime.semantic.*` option, so a consumer can discover the observability
|
||||
//! concept a table stands for with a single SQL query instead of parsing every
|
||||
//! table's `create_options`.
|
||||
//!
|
||||
//! The few signal-agnostic keys are promoted to their own columns
|
||||
//! (`signal_type` / `source` / `pipeline` / `metadata_quality`); the remaining
|
||||
//! signal-specific keys are folded into a `semantic_options` JSON string, keyed
|
||||
//! by the option name with the `greptime.semantic.` prefix stripped.
|
||||
|
||||
use std::collections::BTreeMap;
|
||||
use std::sync::{Arc, Weak};
|
||||
|
||||
use arrow_schema::SchemaRef as ArrowSchemaRef;
|
||||
use common_catalog::consts::INFORMATION_SCHEMA_TABLE_SEMANTICS_TABLE_ID;
|
||||
use common_error::ext::BoxedError;
|
||||
use common_recordbatch::adapter::RecordBatchStreamAdapter;
|
||||
use common_recordbatch::{RecordBatch, SendableRecordBatchStream};
|
||||
use datafusion::execution::TaskContext;
|
||||
use datafusion::physical_plan::SendableRecordBatchStream as DfSendableRecordBatchStream;
|
||||
use datafusion::physical_plan::stream::RecordBatchStreamAdapter as DfRecordBatchStreamAdapter;
|
||||
use datafusion::physical_plan::streaming::PartitionStream as DfPartitionStream;
|
||||
use datatypes::prelude::{ConcreteDataType, ScalarVectorBuilder, VectorRef};
|
||||
use datatypes::schema::{ColumnSchema, Schema, SchemaRef};
|
||||
use datatypes::value::Value;
|
||||
use datatypes::vectors::{StringVectorBuilder, UInt32VectorBuilder};
|
||||
use futures::TryStreamExt;
|
||||
use snafu::{OptionExt, ResultExt};
|
||||
use store_api::storage::{ScanRequest, TableId};
|
||||
use table::metadata::TableInfo;
|
||||
use table::requests::{
|
||||
SEMANTIC_METRIC_METADATA_QUALITY, SEMANTIC_PIPELINE, SEMANTIC_PREFIX, SEMANTIC_SIGNAL_TYPE,
|
||||
SEMANTIC_SOURCE, is_semantic_option_key,
|
||||
};
|
||||
|
||||
use crate::CatalogManager;
|
||||
use crate::error::{
|
||||
CreateRecordBatchSnafu, InternalSnafu, Result, UpgradeWeakCatalogManagerRefSnafu,
|
||||
};
|
||||
use crate::system_schema::information_schema::{InformationTable, Predicates, TABLE_SEMANTICS};
|
||||
|
||||
pub const TABLE_CATALOG: &str = "table_catalog";
|
||||
pub const TABLE_SCHEMA: &str = "table_schema";
|
||||
pub const TABLE_NAME: &str = "table_name";
|
||||
pub const TABLE_ID: &str = "table_id";
|
||||
pub const SIGNAL_TYPE: &str = "signal_type";
|
||||
pub const SOURCE: &str = "source";
|
||||
pub const PIPELINE: &str = "pipeline";
|
||||
pub const METADATA_QUALITY: &str = "metadata_quality";
|
||||
pub const SEMANTIC_OPTIONS: &str = "semantic_options";
|
||||
|
||||
const INIT_CAPACITY: usize = 42;
|
||||
|
||||
fn optional_value(v: Option<&str>) -> Value {
|
||||
v.map(Value::from).unwrap_or(Value::Null)
|
||||
}
|
||||
|
||||
/// The semantic projection of a single table: the signal-agnostic keys promoted
|
||||
/// to columns, plus a JSON tail for the rest. Borrows from the table's options.
|
||||
struct SemanticRow<'a> {
|
||||
signal_type: Option<&'a str>,
|
||||
source: Option<&'a str>,
|
||||
pipeline: Option<&'a str>,
|
||||
metadata_quality: Option<&'a str>,
|
||||
options_json: Option<String>,
|
||||
}
|
||||
|
||||
impl<'a> SemanticRow<'a> {
|
||||
/// Projects a table's options onto the semantic schema, or `None` when the
|
||||
/// table carries no semantic key at all.
|
||||
fn extract(table_info: &'a TableInfo) -> Option<Self> {
|
||||
let mut signal_type = None;
|
||||
let mut source = None;
|
||||
let mut pipeline = None;
|
||||
let mut metadata_quality = None;
|
||||
let mut rest = BTreeMap::new();
|
||||
|
||||
for (key, value) in &table_info.meta.options.extra_options {
|
||||
if !is_semantic_option_key(key) {
|
||||
continue;
|
||||
}
|
||||
match key.as_str() {
|
||||
SEMANTIC_SIGNAL_TYPE => signal_type = Some(value.as_str()),
|
||||
SEMANTIC_SOURCE => source = Some(value.as_str()),
|
||||
SEMANTIC_PIPELINE => pipeline = Some(value.as_str()),
|
||||
SEMANTIC_METRIC_METADATA_QUALITY => metadata_quality = Some(value.as_str()),
|
||||
_ => {
|
||||
let short = key.strip_prefix(SEMANTIC_PREFIX).unwrap_or(key);
|
||||
rest.insert(short, value.as_str());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let has_any = signal_type.is_some()
|
||||
|| source.is_some()
|
||||
|| pipeline.is_some()
|
||||
|| metadata_quality.is_some()
|
||||
|| !rest.is_empty();
|
||||
if !has_any {
|
||||
return None;
|
||||
}
|
||||
|
||||
// `rest` is a `BTreeMap`, so the JSON keys come out sorted and the output
|
||||
// is stable across runs. Serializing a string map can't realistically fail,
|
||||
// but fold a failure into `None` rather than panicking the query path.
|
||||
let options_json = (!rest.is_empty())
|
||||
.then(|| serde_json::to_string(&rest).ok())
|
||||
.flatten();
|
||||
|
||||
Some(Self {
|
||||
signal_type,
|
||||
source,
|
||||
pipeline,
|
||||
metadata_quality,
|
||||
options_json,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
pub(super) struct InformationSchemaTableSemantics {
|
||||
schema: SchemaRef,
|
||||
catalog_name: String,
|
||||
catalog_manager: Weak<dyn CatalogManager>,
|
||||
}
|
||||
|
||||
impl InformationSchemaTableSemantics {
|
||||
pub(super) fn new(catalog_name: String, catalog_manager: Weak<dyn CatalogManager>) -> Self {
|
||||
Self {
|
||||
schema: Self::schema(),
|
||||
catalog_name,
|
||||
catalog_manager,
|
||||
}
|
||||
}
|
||||
|
||||
fn schema() -> SchemaRef {
|
||||
Arc::new(Schema::new(vec![
|
||||
ColumnSchema::new(TABLE_CATALOG, ConcreteDataType::string_datatype(), false),
|
||||
ColumnSchema::new(TABLE_SCHEMA, ConcreteDataType::string_datatype(), false),
|
||||
ColumnSchema::new(TABLE_NAME, ConcreteDataType::string_datatype(), false),
|
||||
ColumnSchema::new(TABLE_ID, ConcreteDataType::uint32_datatype(), false),
|
||||
ColumnSchema::new(SIGNAL_TYPE, ConcreteDataType::string_datatype(), true),
|
||||
ColumnSchema::new(SOURCE, ConcreteDataType::string_datatype(), true),
|
||||
ColumnSchema::new(PIPELINE, ConcreteDataType::string_datatype(), true),
|
||||
ColumnSchema::new(METADATA_QUALITY, ConcreteDataType::string_datatype(), true),
|
||||
ColumnSchema::new(SEMANTIC_OPTIONS, ConcreteDataType::string_datatype(), true),
|
||||
]))
|
||||
}
|
||||
|
||||
fn builder(&self) -> InformationSchemaSemanticTablesBuilder {
|
||||
InformationSchemaSemanticTablesBuilder::new(
|
||||
self.schema.clone(),
|
||||
self.catalog_name.clone(),
|
||||
self.catalog_manager.clone(),
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
impl InformationTable for InformationSchemaTableSemantics {
|
||||
fn table_id(&self) -> TableId {
|
||||
INFORMATION_SCHEMA_TABLE_SEMANTICS_TABLE_ID
|
||||
}
|
||||
|
||||
fn table_name(&self) -> &'static str {
|
||||
TABLE_SEMANTICS
|
||||
}
|
||||
|
||||
fn schema(&self) -> SchemaRef {
|
||||
self.schema.clone()
|
||||
}
|
||||
|
||||
fn to_stream(&self, request: ScanRequest) -> Result<SendableRecordBatchStream> {
|
||||
let schema = self.schema.arrow_schema().clone();
|
||||
let mut builder = self.builder();
|
||||
let stream = Box::pin(DfRecordBatchStreamAdapter::new(
|
||||
schema,
|
||||
futures::stream::once(async move {
|
||||
builder
|
||||
.make_tables(Some(request))
|
||||
.await
|
||||
.map(|x| x.into_df_record_batch())
|
||||
.map_err(|err| datafusion::error::DataFusionError::External(Box::new(err)))
|
||||
}),
|
||||
));
|
||||
Ok(Box::pin(
|
||||
RecordBatchStreamAdapter::try_new(stream)
|
||||
.map_err(BoxedError::new)
|
||||
.context(InternalSnafu)?,
|
||||
))
|
||||
}
|
||||
}
|
||||
|
||||
struct InformationSchemaSemanticTablesBuilder {
|
||||
schema: SchemaRef,
|
||||
catalog_name: String,
|
||||
catalog_manager: Weak<dyn CatalogManager>,
|
||||
|
||||
catalog_names: StringVectorBuilder,
|
||||
schema_names: StringVectorBuilder,
|
||||
table_names: StringVectorBuilder,
|
||||
table_ids: UInt32VectorBuilder,
|
||||
signal_types: StringVectorBuilder,
|
||||
sources: StringVectorBuilder,
|
||||
pipelines: StringVectorBuilder,
|
||||
metadata_qualities: StringVectorBuilder,
|
||||
semantic_options: StringVectorBuilder,
|
||||
}
|
||||
|
||||
impl InformationSchemaSemanticTablesBuilder {
|
||||
fn new(
|
||||
schema: SchemaRef,
|
||||
catalog_name: String,
|
||||
catalog_manager: Weak<dyn CatalogManager>,
|
||||
) -> Self {
|
||||
Self {
|
||||
schema,
|
||||
catalog_name,
|
||||
catalog_manager,
|
||||
catalog_names: StringVectorBuilder::with_capacity(INIT_CAPACITY),
|
||||
schema_names: StringVectorBuilder::with_capacity(INIT_CAPACITY),
|
||||
table_names: StringVectorBuilder::with_capacity(INIT_CAPACITY),
|
||||
table_ids: UInt32VectorBuilder::with_capacity(INIT_CAPACITY),
|
||||
signal_types: StringVectorBuilder::with_capacity(INIT_CAPACITY),
|
||||
sources: StringVectorBuilder::with_capacity(INIT_CAPACITY),
|
||||
pipelines: StringVectorBuilder::with_capacity(INIT_CAPACITY),
|
||||
metadata_qualities: StringVectorBuilder::with_capacity(INIT_CAPACITY),
|
||||
semantic_options: StringVectorBuilder::with_capacity(INIT_CAPACITY),
|
||||
}
|
||||
}
|
||||
|
||||
async fn make_tables(&mut self, request: Option<ScanRequest>) -> Result<RecordBatch> {
|
||||
let catalog_name = self.catalog_name.clone();
|
||||
let catalog_manager = self
|
||||
.catalog_manager
|
||||
.upgrade()
|
||||
.context(UpgradeWeakCatalogManagerRefSnafu)?;
|
||||
let predicates = Predicates::from_scan_request(&request);
|
||||
|
||||
for schema_name in catalog_manager.schema_names(&catalog_name, None).await? {
|
||||
let mut table_stream = catalog_manager.tables(&catalog_name, &schema_name, None);
|
||||
while let Some(table) = table_stream.try_next().await? {
|
||||
self.add_table(&predicates, &catalog_name, &schema_name, table.table_info());
|
||||
}
|
||||
}
|
||||
|
||||
self.finish()
|
||||
}
|
||||
|
||||
fn add_table(
|
||||
&mut self,
|
||||
predicates: &Predicates,
|
||||
catalog_name: &str,
|
||||
schema_name: &str,
|
||||
table_info: Arc<TableInfo>,
|
||||
) {
|
||||
// A table with no semantic key is not part of the semantic layer.
|
||||
let Some(row) = SemanticRow::extract(&table_info) else {
|
||||
return;
|
||||
};
|
||||
|
||||
let table_name = table_info.name.as_ref();
|
||||
let catalog_v = Value::from(catalog_name);
|
||||
let schema_v = Value::from(schema_name);
|
||||
let name_v = Value::from(table_name);
|
||||
let signal_v = optional_value(row.signal_type);
|
||||
let source_v = optional_value(row.source);
|
||||
let pipeline_v = optional_value(row.pipeline);
|
||||
let quality_v = optional_value(row.metadata_quality);
|
||||
let predicate_row = [
|
||||
(TABLE_CATALOG, &catalog_v),
|
||||
(TABLE_SCHEMA, &schema_v),
|
||||
(TABLE_NAME, &name_v),
|
||||
(SIGNAL_TYPE, &signal_v),
|
||||
(SOURCE, &source_v),
|
||||
(PIPELINE, &pipeline_v),
|
||||
(METADATA_QUALITY, &quality_v),
|
||||
];
|
||||
if !predicates.eval(&predicate_row) {
|
||||
return;
|
||||
}
|
||||
|
||||
self.catalog_names.push(Some(catalog_name));
|
||||
self.schema_names.push(Some(schema_name));
|
||||
self.table_names.push(Some(table_name));
|
||||
self.table_ids.push(Some(table_info.table_id()));
|
||||
self.signal_types.push(row.signal_type);
|
||||
self.sources.push(row.source);
|
||||
self.pipelines.push(row.pipeline);
|
||||
self.metadata_qualities.push(row.metadata_quality);
|
||||
self.semantic_options.push(row.options_json.as_deref());
|
||||
}
|
||||
|
||||
fn finish(&mut self) -> Result<RecordBatch> {
|
||||
let columns: Vec<VectorRef> = vec![
|
||||
Arc::new(self.catalog_names.finish()),
|
||||
Arc::new(self.schema_names.finish()),
|
||||
Arc::new(self.table_names.finish()),
|
||||
Arc::new(self.table_ids.finish()),
|
||||
Arc::new(self.signal_types.finish()),
|
||||
Arc::new(self.sources.finish()),
|
||||
Arc::new(self.pipelines.finish()),
|
||||
Arc::new(self.metadata_qualities.finish()),
|
||||
Arc::new(self.semantic_options.finish()),
|
||||
];
|
||||
RecordBatch::new(self.schema.clone(), columns).context(CreateRecordBatchSnafu)
|
||||
}
|
||||
}
|
||||
|
||||
impl DfPartitionStream for InformationSchemaTableSemantics {
|
||||
fn schema(&self) -> &ArrowSchemaRef {
|
||||
self.schema.arrow_schema()
|
||||
}
|
||||
|
||||
fn execute(&self, _: Arc<TaskContext>) -> DfSendableRecordBatchStream {
|
||||
let schema = self.schema.arrow_schema().clone();
|
||||
let mut builder = self.builder();
|
||||
Box::pin(DfRecordBatchStreamAdapter::new(
|
||||
schema,
|
||||
futures::stream::once(async move {
|
||||
builder
|
||||
.make_tables(None)
|
||||
.await
|
||||
.map(|x| x.into_df_record_batch())
|
||||
.map_err(Into::into)
|
||||
}),
|
||||
))
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use std::collections::HashMap;
|
||||
|
||||
use common_catalog::consts::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, MITO_ENGINE};
|
||||
use datatypes::schema::SchemaBuilder;
|
||||
use table::metadata::{TableInfoBuilder, TableMeta, TableType};
|
||||
use table::requests::{SEMANTIC_METRIC_TYPE, SEMANTIC_METRIC_UNIT, TableOptions};
|
||||
|
||||
use super::*;
|
||||
|
||||
fn table_info(extra: &[(&str, &str)]) -> TableInfo {
|
||||
let schema = Arc::new(
|
||||
SchemaBuilder::try_from_columns(vec![ColumnSchema::new(
|
||||
"ts",
|
||||
ConcreteDataType::timestamp_millisecond_datatype(),
|
||||
false,
|
||||
)])
|
||||
.unwrap()
|
||||
.build()
|
||||
.unwrap(),
|
||||
);
|
||||
let options = TableOptions {
|
||||
extra_options: extra
|
||||
.iter()
|
||||
.map(|(k, v)| (k.to_string(), v.to_string()))
|
||||
.collect::<HashMap<_, _>>(),
|
||||
..Default::default()
|
||||
};
|
||||
let meta = TableMeta {
|
||||
schema,
|
||||
primary_key_indices: vec![],
|
||||
value_indices: vec![],
|
||||
engine: MITO_ENGINE.to_string(),
|
||||
next_column_id: 1,
|
||||
options,
|
||||
created_on: Default::default(),
|
||||
updated_on: Default::default(),
|
||||
partition_key_indices: vec![],
|
||||
column_ids: vec![],
|
||||
};
|
||||
TableInfoBuilder::default()
|
||||
.table_id(1)
|
||||
.name("t")
|
||||
.catalog_name(DEFAULT_CATALOG_NAME)
|
||||
.schema_name(DEFAULT_SCHEMA_NAME)
|
||||
.table_version(0)
|
||||
.table_type(TableType::Base)
|
||||
.meta(meta)
|
||||
.build()
|
||||
.unwrap()
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn extract_promotes_core_keys_and_folds_the_rest() {
|
||||
let info = table_info(&[
|
||||
(SEMANTIC_SIGNAL_TYPE, "metric"),
|
||||
(SEMANTIC_SOURCE, "opentelemetry"),
|
||||
(SEMANTIC_PIPELINE, "greptime_metric_v1"),
|
||||
(SEMANTIC_METRIC_METADATA_QUALITY, "declared"),
|
||||
(SEMANTIC_METRIC_TYPE, "counter"),
|
||||
(SEMANTIC_METRIC_UNIT, "By"),
|
||||
// A non-semantic option must be ignored entirely.
|
||||
("ttl", "7d"),
|
||||
]);
|
||||
|
||||
let row = SemanticRow::extract(&info).unwrap();
|
||||
assert_eq!(row.signal_type, Some("metric"));
|
||||
assert_eq!(row.source, Some("opentelemetry"));
|
||||
assert_eq!(row.pipeline, Some("greptime_metric_v1"));
|
||||
assert_eq!(row.metadata_quality, Some("declared"));
|
||||
// Promoted keys stay out of the JSON tail; remaining keys are sorted and
|
||||
// prefix-stripped.
|
||||
assert_eq!(
|
||||
row.options_json.as_deref(),
|
||||
Some(r#"{"metric.type":"counter","metric.unit":"By"}"#)
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn extract_skips_untagged_table() {
|
||||
let info = table_info(&[("ttl", "7d")]);
|
||||
assert!(SemanticRow::extract(&info).is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn extract_omits_json_when_only_core_keys_present() {
|
||||
let info = table_info(&[(SEMANTIC_SIGNAL_TYPE, "log")]);
|
||||
let row = SemanticRow::extract(&info).unwrap();
|
||||
assert_eq!(row.signal_type, Some("log"));
|
||||
assert!(row.source.is_none());
|
||||
assert!(row.options_json.is_none());
|
||||
}
|
||||
}
|
||||
@@ -114,6 +114,8 @@ pub const INFORMATION_SCHEMA_SSTS_INDEX_META_TABLE_ID: u32 = 39;
|
||||
pub const INFORMATION_SCHEMA_ALERTS_TABLE_ID: u32 = 40;
|
||||
/// id for information_schema.region_info
|
||||
pub const INFORMATION_SCHEMA_REGION_INFO_TABLE_ID: u32 = 41;
|
||||
/// id for information_schema.table_semantics
|
||||
pub const INFORMATION_SCHEMA_TABLE_SEMANTICS_TABLE_ID: u32 = 42;
|
||||
|
||||
// ----- End of information_schema tables -----
|
||||
|
||||
|
||||
@@ -0,0 +1,86 @@
|
||||
DESC TABLE information_schema.table_semantics;
|
||||
|
||||
+------------------+--------+-----+------+---------+---------------+
|
||||
| Column | Type | Key | Null | Default | Semantic Type |
|
||||
+------------------+--------+-----+------+---------+---------------+
|
||||
| table_catalog | String | | NO | | FIELD |
|
||||
| table_schema | String | | NO | | FIELD |
|
||||
| table_name | String | | NO | | FIELD |
|
||||
| table_id | UInt32 | | NO | | FIELD |
|
||||
| signal_type | String | | YES | | FIELD |
|
||||
| source | String | | YES | | FIELD |
|
||||
| pipeline | String | | YES | | FIELD |
|
||||
| metadata_quality | String | | YES | | FIELD |
|
||||
| semantic_options | String | | YES | | FIELD |
|
||||
+------------------+--------+-----+------+---------+---------------+
|
||||
|
||||
CREATE TABLE metrics_tagged (
|
||||
ts TIMESTAMP TIME INDEX,
|
||||
val DOUBLE,
|
||||
)
|
||||
WITH (
|
||||
'greptime.semantic.signal_type' = 'metric',
|
||||
'greptime.semantic.source' = 'opentelemetry',
|
||||
'greptime.semantic.pipeline' = 'greptime_metric_v1',
|
||||
'greptime.semantic.metric.metadata_quality' = 'declared',
|
||||
'greptime.semantic.metric.type' = 'counter',
|
||||
'greptime.semantic.metric.unit' = 'By'
|
||||
);
|
||||
|
||||
Affected Rows: 0
|
||||
|
||||
CREATE TABLE traces_tagged (
|
||||
ts TIMESTAMP TIME INDEX,
|
||||
span_name STRING,
|
||||
)
|
||||
WITH (
|
||||
'greptime.semantic.signal_type' = 'trace',
|
||||
'greptime.semantic.source' = 'opentelemetry',
|
||||
'greptime.semantic.trace.conventions' = 'https://opentelemetry.io/schemas/1.27.0'
|
||||
);
|
||||
|
||||
Affected Rows: 0
|
||||
|
||||
-- A table with no semantic options must not appear in the view.
|
||||
CREATE TABLE plain_table (
|
||||
ts TIMESTAMP TIME INDEX,
|
||||
val DOUBLE,
|
||||
);
|
||||
|
||||
Affected Rows: 0
|
||||
|
||||
SELECT table_schema, table_name, signal_type, source, pipeline, metadata_quality, semantic_options
|
||||
FROM information_schema.table_semantics
|
||||
ORDER BY table_name;
|
||||
|
||||
+--------------+----------------+-------------+---------------+--------------------+------------------+-----------------------------------------------------------------+
|
||||
| table_schema | table_name | signal_type | source | pipeline | metadata_quality | semantic_options |
|
||||
+--------------+----------------+-------------+---------------+--------------------+------------------+-----------------------------------------------------------------+
|
||||
| public | metrics_tagged | metric | opentelemetry | greptime_metric_v1 | declared | {"metric.type":"counter","metric.unit":"By"} |
|
||||
| public | traces_tagged | trace | opentelemetry | | | {"trace.conventions":"https://opentelemetry.io/schemas/1.27.0"} |
|
||||
+--------------+----------------+-------------+---------------+--------------------+------------------+-----------------------------------------------------------------+
|
||||
|
||||
-- Predicate pushdown on a promoted column.
|
||||
SELECT table_name, signal_type
|
||||
FROM information_schema.table_semantics
|
||||
WHERE signal_type = 'metric'
|
||||
ORDER BY table_name;
|
||||
|
||||
+----------------+-------------+
|
||||
| table_name | signal_type |
|
||||
+----------------+-------------+
|
||||
| metrics_tagged | metric |
|
||||
+----------------+-------------+
|
||||
|
||||
DROP TABLE metrics_tagged;
|
||||
|
||||
Affected Rows: 0
|
||||
|
||||
DROP TABLE traces_tagged;
|
||||
|
||||
Affected Rows: 0
|
||||
|
||||
DROP TABLE plain_table;
|
||||
|
||||
Affected Rows: 0
|
||||
|
||||
@@ -0,0 +1,46 @@
|
||||
DESC TABLE information_schema.table_semantics;
|
||||
|
||||
CREATE TABLE metrics_tagged (
|
||||
ts TIMESTAMP TIME INDEX,
|
||||
val DOUBLE,
|
||||
)
|
||||
WITH (
|
||||
'greptime.semantic.signal_type' = 'metric',
|
||||
'greptime.semantic.source' = 'opentelemetry',
|
||||
'greptime.semantic.pipeline' = 'greptime_metric_v1',
|
||||
'greptime.semantic.metric.metadata_quality' = 'declared',
|
||||
'greptime.semantic.metric.type' = 'counter',
|
||||
'greptime.semantic.metric.unit' = 'By'
|
||||
);
|
||||
|
||||
CREATE TABLE traces_tagged (
|
||||
ts TIMESTAMP TIME INDEX,
|
||||
span_name STRING,
|
||||
)
|
||||
WITH (
|
||||
'greptime.semantic.signal_type' = 'trace',
|
||||
'greptime.semantic.source' = 'opentelemetry',
|
||||
'greptime.semantic.trace.conventions' = 'https://opentelemetry.io/schemas/1.27.0'
|
||||
);
|
||||
|
||||
-- A table with no semantic options must not appear in the view.
|
||||
CREATE TABLE plain_table (
|
||||
ts TIMESTAMP TIME INDEX,
|
||||
val DOUBLE,
|
||||
);
|
||||
|
||||
SELECT table_schema, table_name, signal_type, source, pipeline, metadata_quality, semantic_options
|
||||
FROM information_schema.table_semantics
|
||||
ORDER BY table_name;
|
||||
|
||||
-- Predicate pushdown on a promoted column.
|
||||
SELECT table_name, signal_type
|
||||
FROM information_schema.table_semantics
|
||||
WHERE signal_type = 'metric'
|
||||
ORDER BY table_name;
|
||||
|
||||
DROP TABLE metrics_tagged;
|
||||
|
||||
DROP TABLE traces_tagged;
|
||||
|
||||
DROP TABLE plain_table;
|
||||
@@ -61,6 +61,7 @@ SHOW TABLES;
|
||||
| ssts_storage |
|
||||
| table_constraints |
|
||||
| table_privileges |
|
||||
| table_semantics |
|
||||
| tables |
|
||||
| views |
|
||||
+---------------------------------------+
|
||||
@@ -112,6 +113,7 @@ SHOW FULL TABLES;
|
||||
| ssts_storage | LOCAL TEMPORARY |
|
||||
| table_constraints | LOCAL TEMPORARY |
|
||||
| table_privileges | LOCAL TEMPORARY |
|
||||
| table_semantics | LOCAL TEMPORARY |
|
||||
| tables | LOCAL TEMPORARY |
|
||||
| views | LOCAL TEMPORARY |
|
||||
+---------------------------------------+-----------------+
|
||||
@@ -157,6 +159,7 @@ SHOW TABLE STATUS;
|
||||
|ssts_storage||11|Fixed|0|0|0|0|0|0|0|DATETIME|DATETIME||utf8_bin|0|||
|
||||
|table_constraints||11|Fixed|0|0|0|0|0|0|0|DATETIME|DATETIME||utf8_bin|0|||
|
||||
|table_privileges||11|Fixed|0|0|0|0|0|0|0|DATETIME|DATETIME||utf8_bin|0|||
|
||||
|table_semantics||11|Fixed|0|0|0|0|0|0|0|DATETIME|DATETIME||utf8_bin|0|||
|
||||
|tables||11|Fixed|0|0|0|0|0|0|0|DATETIME|DATETIME||utf8_bin|0|||
|
||||
|views||11|Fixed|0|0|0|0|0|0|0|DATETIME|DATETIME||utf8_bin|0|||
|
||||
+++++++++++++++++++
|
||||
|
||||
@@ -48,6 +48,7 @@ order by table_schema, table_name;
|
||||
|greptime|information_schema|ssts_storage|LOCALTEMPORARY|38|0|0|0|0|0||11|Fixed|0|0|0|DATETIME|DATETIME||utf8_bin|0|||Y|
|
||||
|greptime|information_schema|table_constraints|LOCALTEMPORARY|30|0|0|0|0|0||11|Fixed|0|0|0|DATETIME|DATETIME||utf8_bin|0|||Y|
|
||||
|greptime|information_schema|table_privileges|LOCALTEMPORARY|23|0|0|0|0|0||11|Fixed|0|0|0|DATETIME|DATETIME||utf8_bin|0|||Y|
|
||||
|greptime|information_schema|table_semantics|LOCALTEMPORARY|42|0|0|0|0|0||11|Fixed|0|0|0|DATETIME|DATETIME||utf8_bin|0|||Y|
|
||||
|greptime|information_schema|tables|LOCALTEMPORARY|3|0|0|0|0|0||11|Fixed|0|0|0|DATETIME|DATETIME||utf8_bin|0|||Y|
|
||||
|greptime|information_schema|views|LOCALTEMPORARY|32|0|0|0|0|0||11|Fixed|0|0|0|DATETIME|DATETIME||utf8_bin|0|||Y|
|
||||
|greptime|public|numbers|LOCALTEMPORARY|2|0|0|0|0|0|test_engine|11|Fixed|0|0|0|DATETIME|DATETIME||utf8_bin|0|||Y|
|
||||
@@ -455,6 +456,15 @@ select * from information_schema.columns order by table_schema, table_name, colu
|
||||
| greptime | information_schema | table_privileges | table_catalog | 2 | 2147483647 | 2147483647 | | | | utf8 | utf8_bin | | | select,insert | | String | string | FIELD | | No | string | | |
|
||||
| greptime | information_schema | table_privileges | table_name | 4 | 2147483647 | 2147483647 | | | | utf8 | utf8_bin | | | select,insert | | String | string | FIELD | | No | string | | |
|
||||
| greptime | information_schema | table_privileges | table_schema | 3 | 2147483647 | 2147483647 | | | | utf8 | utf8_bin | | | select,insert | | String | string | FIELD | | No | string | | |
|
||||
| greptime | information_schema | table_semantics | metadata_quality | 8 | 2147483647 | 2147483647 | | | | utf8 | utf8_bin | | | select,insert | | String | string | FIELD | | Yes | string | | |
|
||||
| greptime | information_schema | table_semantics | pipeline | 7 | 2147483647 | 2147483647 | | | | utf8 | utf8_bin | | | select,insert | | String | string | FIELD | | Yes | string | | |
|
||||
| greptime | information_schema | table_semantics | semantic_options | 9 | 2147483647 | 2147483647 | | | | utf8 | utf8_bin | | | select,insert | | String | string | FIELD | | Yes | string | | |
|
||||
| greptime | information_schema | table_semantics | signal_type | 5 | 2147483647 | 2147483647 | | | | utf8 | utf8_bin | | | select,insert | | String | string | FIELD | | Yes | string | | |
|
||||
| greptime | information_schema | table_semantics | source | 6 | 2147483647 | 2147483647 | | | | utf8 | utf8_bin | | | select,insert | | String | string | FIELD | | Yes | string | | |
|
||||
| greptime | information_schema | table_semantics | table_catalog | 1 | 2147483647 | 2147483647 | | | | utf8 | utf8_bin | | | select,insert | | String | string | FIELD | | No | string | | |
|
||||
| greptime | information_schema | table_semantics | table_id | 4 | | | 10 | 0 | | | | | | select,insert | | UInt32 | int unsigned | FIELD | | No | int unsigned | | |
|
||||
| greptime | information_schema | table_semantics | table_name | 3 | 2147483647 | 2147483647 | | | | utf8 | utf8_bin | | | select,insert | | String | string | FIELD | | No | string | | |
|
||||
| greptime | information_schema | table_semantics | table_schema | 2 | 2147483647 | 2147483647 | | | | utf8 | utf8_bin | | | select,insert | | String | string | FIELD | | No | string | | |
|
||||
| greptime | information_schema | tables | auto_increment | 16 | | | 20 | 0 | | | | | | select,insert | | UInt64 | bigint unsigned | FIELD | | Yes | bigint unsigned | | |
|
||||
| greptime | information_schema | tables | avg_row_length | 10 | | | 20 | 0 | | | | | | select,insert | | UInt64 | bigint unsigned | FIELD | | Yes | bigint unsigned | | |
|
||||
| greptime | information_schema | tables | check_time | 19 | | | | | 0 | | | | | select,insert | | TimestampSecond | timestamp(0) | FIELD | | Yes | timestamp(0) | | |
|
||||
|
||||
@@ -128,6 +128,7 @@ SELECT * FROM INFORMATION_SCHEMA.TABLES ORDER BY TABLE_NAME, TABLE_TYPE;
|
||||
|greptime|information_schema|ssts_storage|LOCALTEMPORARY|ID|ID|ID|ID|ID|ID||ID|Fixed|ID|ID|ID|DATETIME|DATETIME||utf8_bin|ID|||Y|
|
||||
|greptime|information_schema|table_constraints|LOCALTEMPORARY|ID|ID|ID|ID|ID|ID||ID|Fixed|ID|ID|ID|DATETIME|DATETIME||utf8_bin|ID|||Y|
|
||||
|greptime|information_schema|table_privileges|LOCALTEMPORARY|ID|ID|ID|ID|ID|ID||ID|Fixed|ID|ID|ID|DATETIME|DATETIME||utf8_bin|ID|||Y|
|
||||
|greptime|information_schema|table_semantics|LOCALTEMPORARY|ID|ID|ID|ID|ID|ID||ID|Fixed|ID|ID|ID|DATETIME|DATETIME||utf8_bin|ID|||Y|
|
||||
|greptime|information_schema|tables|LOCALTEMPORARY|ID|ID|ID|ID|ID|ID||ID|Fixed|ID|ID|ID|DATETIME|DATETIME||utf8_bin|ID|||Y|
|
||||
|greptime|public|test_table|BASETABLE|ID|ID|ID|ID|ID|ID|mito|ID|Fixed|ID|ID|ID|DATETIME|DATETIME||utf8_bin|ID|||N|
|
||||
|greptime|public|test_view|VIEW|ID|ID|ID|ID|ID|ID||ID|Fixed|ID|ID|ID|DATETIME|DATETIME||utf8_bin|ID|||N|
|
||||
|
||||
Reference in New Issue
Block a user