catalog/system_schema/information_schema/
columns.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
15use std::sync::{Arc, Weak};
16
17use arrow_schema::SchemaRef as ArrowSchemaRef;
18use common_catalog::consts::{
19    INFORMATION_SCHEMA_COLUMNS_TABLE_ID, SEMANTIC_TYPE_FIELD, SEMANTIC_TYPE_PRIMARY_KEY,
20    SEMANTIC_TYPE_TIME_INDEX,
21};
22use common_error::ext::BoxedError;
23use common_recordbatch::adapter::RecordBatchStreamAdapter;
24use common_recordbatch::{RecordBatch, SendableRecordBatchStream};
25use datafusion::execution::TaskContext;
26use datafusion::physical_plan::SendableRecordBatchStream as DfSendableRecordBatchStream;
27use datafusion::physical_plan::stream::RecordBatchStreamAdapter as DfRecordBatchStreamAdapter;
28use datafusion::physical_plan::streaming::PartitionStream as DfPartitionStream;
29use datatypes::prelude::{ConcreteDataType, DataType, MutableVector};
30use datatypes::scalars::ScalarVectorBuilder;
31use datatypes::schema::{ColumnSchema, Schema, SchemaRef};
32use datatypes::value::Value;
33use datatypes::vectors::{
34    ConstantVector, Int64Vector, Int64VectorBuilder, StringVector, StringVectorBuilder, VectorRef,
35};
36use futures::TryStreamExt;
37use snafu::{OptionExt, ResultExt};
38use sql::statements;
39use store_api::storage::{ScanRequest, TableId};
40
41use crate::CatalogManager;
42use crate::error::{
43    CreateRecordBatchSnafu, InternalSnafu, Result, UpgradeWeakCatalogManagerRefSnafu,
44};
45use crate::information_schema::Predicates;
46use crate::system_schema::information_schema::{COLUMNS, InformationTable};
47
48#[derive(Debug)]
49pub(super) struct InformationSchemaColumns {
50    schema: SchemaRef,
51    catalog_name: String,
52    catalog_manager: Weak<dyn CatalogManager>,
53}
54
55pub const TABLE_CATALOG: &str = "table_catalog";
56pub const TABLE_SCHEMA: &str = "table_schema";
57pub const TABLE_NAME: &str = "table_name";
58pub const COLUMN_NAME: &str = "column_name";
59pub const REGION_ID: &str = "region_id";
60pub const PEER_ID: &str = "peer_id";
61const ORDINAL_POSITION: &str = "ordinal_position";
62const CHARACTER_MAXIMUM_LENGTH: &str = "character_maximum_length";
63const CHARACTER_OCTET_LENGTH: &str = "character_octet_length";
64const NUMERIC_PRECISION: &str = "numeric_precision";
65const NUMERIC_SCALE: &str = "numeric_scale";
66const DATETIME_PRECISION: &str = "datetime_precision";
67const CHARACTER_SET_NAME: &str = "character_set_name";
68pub const COLLATION_NAME: &str = "collation_name";
69pub const COLUMN_KEY: &str = "column_key";
70pub const EXTRA: &str = "extra";
71pub const PRIVILEGES: &str = "privileges";
72const GENERATION_EXPRESSION: &str = "generation_expression";
73// Extension field to keep greptime data type name
74pub const GREPTIME_DATA_TYPE: &str = "greptime_data_type";
75pub const DATA_TYPE: &str = "data_type";
76pub const SEMANTIC_TYPE: &str = "semantic_type";
77pub const COLUMN_DEFAULT: &str = "column_default";
78pub const IS_NULLABLE: &str = "is_nullable";
79const COLUMN_TYPE: &str = "column_type";
80pub const COLUMN_COMMENT: &str = "column_comment";
81const SRS_ID: &str = "srs_id";
82const INIT_CAPACITY: usize = 42;
83
84// The maximum length of string type
85const MAX_STRING_LENGTH: i64 = 2147483647;
86const UTF8_CHARSET_NAME: &str = "utf8";
87const UTF8_COLLATE_NAME: &str = "utf8_bin";
88const PRI_COLUMN_KEY: &str = "PRI";
89const TIME_INDEX_COLUMN_KEY: &str = "TIME INDEX";
90const DEFAULT_PRIVILEGES: &str = "select,insert";
91const EMPTY_STR: &str = "";
92const YES: &str = "YES";
93const NO: &str = "NO";
94
95impl InformationSchemaColumns {
96    pub(super) fn new(catalog_name: String, catalog_manager: Weak<dyn CatalogManager>) -> Self {
97        Self {
98            schema: Self::schema(),
99            catalog_name,
100            catalog_manager,
101        }
102    }
103
104    fn schema() -> SchemaRef {
105        Arc::new(Schema::new(vec![
106            ColumnSchema::new(TABLE_CATALOG, ConcreteDataType::string_datatype(), false),
107            ColumnSchema::new(TABLE_SCHEMA, ConcreteDataType::string_datatype(), false),
108            ColumnSchema::new(TABLE_NAME, ConcreteDataType::string_datatype(), false),
109            ColumnSchema::new(COLUMN_NAME, ConcreteDataType::string_datatype(), false),
110            ColumnSchema::new(ORDINAL_POSITION, ConcreteDataType::int64_datatype(), false),
111            ColumnSchema::new(
112                CHARACTER_MAXIMUM_LENGTH,
113                ConcreteDataType::int64_datatype(),
114                true,
115            ),
116            ColumnSchema::new(
117                CHARACTER_OCTET_LENGTH,
118                ConcreteDataType::int64_datatype(),
119                true,
120            ),
121            ColumnSchema::new(NUMERIC_PRECISION, ConcreteDataType::int64_datatype(), true),
122            ColumnSchema::new(NUMERIC_SCALE, ConcreteDataType::int64_datatype(), true),
123            ColumnSchema::new(DATETIME_PRECISION, ConcreteDataType::int64_datatype(), true),
124            ColumnSchema::new(
125                CHARACTER_SET_NAME,
126                ConcreteDataType::string_datatype(),
127                true,
128            ),
129            ColumnSchema::new(COLLATION_NAME, ConcreteDataType::string_datatype(), true),
130            ColumnSchema::new(COLUMN_KEY, ConcreteDataType::string_datatype(), false),
131            ColumnSchema::new(EXTRA, ConcreteDataType::string_datatype(), false),
132            ColumnSchema::new(PRIVILEGES, ConcreteDataType::string_datatype(), false),
133            ColumnSchema::new(
134                GENERATION_EXPRESSION,
135                ConcreteDataType::string_datatype(),
136                false,
137            ),
138            ColumnSchema::new(
139                GREPTIME_DATA_TYPE,
140                ConcreteDataType::string_datatype(),
141                false,
142            ),
143            ColumnSchema::new(DATA_TYPE, ConcreteDataType::string_datatype(), false),
144            ColumnSchema::new(SEMANTIC_TYPE, ConcreteDataType::string_datatype(), false),
145            ColumnSchema::new(COLUMN_DEFAULT, ConcreteDataType::string_datatype(), true),
146            ColumnSchema::new(IS_NULLABLE, ConcreteDataType::string_datatype(), false),
147            ColumnSchema::new(COLUMN_TYPE, ConcreteDataType::string_datatype(), false),
148            ColumnSchema::new(COLUMN_COMMENT, ConcreteDataType::string_datatype(), true),
149            ColumnSchema::new(SRS_ID, ConcreteDataType::int64_datatype(), true),
150        ]))
151    }
152
153    fn builder(&self) -> InformationSchemaColumnsBuilder {
154        InformationSchemaColumnsBuilder::new(
155            self.schema.clone(),
156            self.catalog_name.clone(),
157            self.catalog_manager.clone(),
158        )
159    }
160}
161
162impl InformationTable for InformationSchemaColumns {
163    fn table_id(&self) -> TableId {
164        INFORMATION_SCHEMA_COLUMNS_TABLE_ID
165    }
166
167    fn table_name(&self) -> &'static str {
168        COLUMNS
169    }
170
171    fn schema(&self) -> SchemaRef {
172        self.schema.clone()
173    }
174
175    fn to_stream(&self, request: ScanRequest) -> Result<SendableRecordBatchStream> {
176        let schema = self.schema.arrow_schema().clone();
177        let mut builder = self.builder();
178        let stream = Box::pin(DfRecordBatchStreamAdapter::new(
179            schema,
180            futures::stream::once(async move {
181                builder
182                    .make_columns(Some(request))
183                    .await
184                    .map(|x| x.into_df_record_batch())
185                    .map_err(Into::into)
186            }),
187        ));
188        Ok(Box::pin(
189            RecordBatchStreamAdapter::try_new(stream)
190                .map_err(BoxedError::new)
191                .context(InternalSnafu)?,
192        ))
193    }
194}
195
196struct InformationSchemaColumnsBuilder {
197    schema: SchemaRef,
198    catalog_name: String,
199    catalog_manager: Weak<dyn CatalogManager>,
200
201    catalog_names: StringVectorBuilder,
202    schema_names: StringVectorBuilder,
203    table_names: StringVectorBuilder,
204    column_names: StringVectorBuilder,
205    ordinal_positions: Int64VectorBuilder,
206    character_maximum_lengths: Int64VectorBuilder,
207    character_octet_lengths: Int64VectorBuilder,
208    numeric_precisions: Int64VectorBuilder,
209    numeric_scales: Int64VectorBuilder,
210    datetime_precisions: Int64VectorBuilder,
211    character_set_names: StringVectorBuilder,
212    collation_names: StringVectorBuilder,
213    column_keys: StringVectorBuilder,
214    greptime_data_types: StringVectorBuilder,
215    data_types: StringVectorBuilder,
216    semantic_types: StringVectorBuilder,
217    column_defaults: StringVectorBuilder,
218    is_nullables: StringVectorBuilder,
219    column_types: StringVectorBuilder,
220    column_comments: StringVectorBuilder,
221}
222
223impl InformationSchemaColumnsBuilder {
224    fn new(
225        schema: SchemaRef,
226        catalog_name: String,
227        catalog_manager: Weak<dyn CatalogManager>,
228    ) -> Self {
229        Self {
230            schema,
231            catalog_name,
232            catalog_manager,
233            catalog_names: StringVectorBuilder::with_capacity(INIT_CAPACITY),
234            schema_names: StringVectorBuilder::with_capacity(INIT_CAPACITY),
235            table_names: StringVectorBuilder::with_capacity(INIT_CAPACITY),
236            column_names: StringVectorBuilder::with_capacity(INIT_CAPACITY),
237            ordinal_positions: Int64VectorBuilder::with_capacity(INIT_CAPACITY),
238            character_maximum_lengths: Int64VectorBuilder::with_capacity(INIT_CAPACITY),
239            character_octet_lengths: Int64VectorBuilder::with_capacity(INIT_CAPACITY),
240            numeric_precisions: Int64VectorBuilder::with_capacity(INIT_CAPACITY),
241            numeric_scales: Int64VectorBuilder::with_capacity(INIT_CAPACITY),
242            datetime_precisions: Int64VectorBuilder::with_capacity(INIT_CAPACITY),
243            character_set_names: StringVectorBuilder::with_capacity(INIT_CAPACITY),
244            collation_names: StringVectorBuilder::with_capacity(INIT_CAPACITY),
245            column_keys: StringVectorBuilder::with_capacity(INIT_CAPACITY),
246            greptime_data_types: StringVectorBuilder::with_capacity(INIT_CAPACITY),
247            data_types: StringVectorBuilder::with_capacity(INIT_CAPACITY),
248            semantic_types: StringVectorBuilder::with_capacity(INIT_CAPACITY),
249            column_defaults: StringVectorBuilder::with_capacity(INIT_CAPACITY),
250            is_nullables: StringVectorBuilder::with_capacity(INIT_CAPACITY),
251            column_types: StringVectorBuilder::with_capacity(INIT_CAPACITY),
252            column_comments: StringVectorBuilder::with_capacity(INIT_CAPACITY),
253        }
254    }
255
256    /// Construct the `information_schema.columns` virtual table
257    async fn make_columns(&mut self, request: Option<ScanRequest>) -> Result<RecordBatch> {
258        let catalog_name = self.catalog_name.clone();
259        let catalog_manager = self
260            .catalog_manager
261            .upgrade()
262            .context(UpgradeWeakCatalogManagerRefSnafu)?;
263        let predicates = Predicates::from_scan_request(&request);
264
265        for schema_name in catalog_manager.schema_names(&catalog_name, None).await? {
266            let mut stream = catalog_manager.tables(&catalog_name, &schema_name, None);
267
268            while let Some(table) = stream.try_next().await? {
269                let keys = &table.table_info().meta.primary_key_indices;
270                let schema = table.schema();
271
272                for (idx, column) in schema.column_schemas().iter().enumerate() {
273                    let semantic_type = if column.is_time_index() {
274                        SEMANTIC_TYPE_TIME_INDEX
275                    } else if keys.contains(&idx) {
276                        SEMANTIC_TYPE_PRIMARY_KEY
277                    } else {
278                        SEMANTIC_TYPE_FIELD
279                    };
280
281                    self.add_column(
282                        &predicates,
283                        idx,
284                        &catalog_name,
285                        &schema_name,
286                        &table.table_info().name,
287                        semantic_type,
288                        column,
289                    );
290                }
291            }
292        }
293
294        self.finish()
295    }
296
297    #[allow(clippy::too_many_arguments)]
298    fn add_column(
299        &mut self,
300        predicates: &Predicates,
301        index: usize,
302        catalog_name: &str,
303        schema_name: &str,
304        table_name: &str,
305        semantic_type: &str,
306        column_schema: &ColumnSchema,
307    ) {
308        // Use sql data type name
309        let data_type = statements::concrete_data_type_to_sql_data_type(&column_schema.data_type)
310            .map(|dt| dt.to_string().to_lowercase())
311            .unwrap_or_else(|_| column_schema.data_type.name());
312
313        let column_key = match semantic_type {
314            SEMANTIC_TYPE_PRIMARY_KEY => PRI_COLUMN_KEY,
315            SEMANTIC_TYPE_TIME_INDEX => TIME_INDEX_COLUMN_KEY,
316            _ => EMPTY_STR,
317        };
318
319        let row = [
320            (TABLE_CATALOG, &Value::from(catalog_name)),
321            (TABLE_SCHEMA, &Value::from(schema_name)),
322            (TABLE_NAME, &Value::from(table_name)),
323            (COLUMN_NAME, &Value::from(column_schema.name.as_str())),
324            (DATA_TYPE, &Value::from(data_type.as_str())),
325            (SEMANTIC_TYPE, &Value::from(semantic_type)),
326            (ORDINAL_POSITION, &Value::from((index + 1) as i64)),
327            (COLUMN_KEY, &Value::from(column_key)),
328        ];
329
330        if !predicates.eval(&row) {
331            return;
332        }
333
334        self.catalog_names.push(Some(catalog_name));
335        self.schema_names.push(Some(schema_name));
336        self.table_names.push(Some(table_name));
337        self.column_names.push(Some(&column_schema.name));
338        // Starts from 1
339        self.ordinal_positions.push(Some((index + 1) as i64));
340
341        if column_schema.data_type.is_string() {
342            self.character_maximum_lengths.push(Some(MAX_STRING_LENGTH));
343            self.character_octet_lengths.push(Some(MAX_STRING_LENGTH));
344            self.numeric_precisions.push(None);
345            self.numeric_scales.push(None);
346            self.datetime_precisions.push(None);
347            self.character_set_names.push(Some(UTF8_CHARSET_NAME));
348            self.collation_names.push(Some(UTF8_COLLATE_NAME));
349        } else if column_schema.data_type.is_numeric() || column_schema.data_type.is_decimal() {
350            self.character_maximum_lengths.push(None);
351            self.character_octet_lengths.push(None);
352
353            self.numeric_precisions.push(
354                column_schema
355                    .data_type
356                    .numeric_precision()
357                    .map(|x| x as i64),
358            );
359            self.numeric_scales
360                .push(column_schema.data_type.numeric_scale().map(|x| x as i64));
361
362            self.datetime_precisions.push(None);
363            self.character_set_names.push(None);
364            self.collation_names.push(None);
365        } else {
366            self.character_maximum_lengths.push(None);
367            self.character_octet_lengths.push(None);
368            self.numeric_precisions.push(None);
369            self.numeric_scales.push(None);
370
371            match &column_schema.data_type {
372                ConcreteDataType::Timestamp(ts_type) => {
373                    self.datetime_precisions
374                        .push(Some(ts_type.precision() as i64));
375                }
376                ConcreteDataType::Time(time_type) => {
377                    self.datetime_precisions
378                        .push(Some(time_type.precision() as i64));
379                }
380                _ => self.datetime_precisions.push(None),
381            }
382
383            self.character_set_names.push(None);
384            self.collation_names.push(None);
385        }
386
387        self.column_keys.push(Some(column_key));
388        self.greptime_data_types
389            .push(Some(&column_schema.data_type.name()));
390        self.data_types.push(Some(&data_type));
391        self.semantic_types.push(Some(semantic_type));
392        self.column_defaults.push(
393            column_schema
394                .default_constraint()
395                .map(|s| format!("{}", s))
396                .as_deref(),
397        );
398        if column_schema.is_nullable() {
399            self.is_nullables.push(Some(YES));
400        } else {
401            self.is_nullables.push(Some(NO));
402        }
403        self.column_types.push(Some(&data_type));
404        let column_comment = column_schema.column_comment().map(|x| x.as_ref());
405        self.column_comments.push(column_comment);
406    }
407
408    fn finish(&mut self) -> Result<RecordBatch> {
409        let rows_num = self.collation_names.len();
410
411        let privileges = Arc::new(ConstantVector::new(
412            Arc::new(StringVector::from(vec![DEFAULT_PRIVILEGES])),
413            rows_num,
414        ));
415        let empty_string = Arc::new(ConstantVector::new(
416            Arc::new(StringVector::from(vec![EMPTY_STR])),
417            rows_num,
418        ));
419        let srs_ids = Arc::new(ConstantVector::new(
420            Arc::new(Int64Vector::from(vec![None])),
421            rows_num,
422        ));
423
424        let columns: Vec<VectorRef> = vec![
425            Arc::new(self.catalog_names.finish()),
426            Arc::new(self.schema_names.finish()),
427            Arc::new(self.table_names.finish()),
428            Arc::new(self.column_names.finish()),
429            Arc::new(self.ordinal_positions.finish()),
430            Arc::new(self.character_maximum_lengths.finish()),
431            Arc::new(self.character_octet_lengths.finish()),
432            Arc::new(self.numeric_precisions.finish()),
433            Arc::new(self.numeric_scales.finish()),
434            Arc::new(self.datetime_precisions.finish()),
435            Arc::new(self.character_set_names.finish()),
436            Arc::new(self.collation_names.finish()),
437            Arc::new(self.column_keys.finish()),
438            empty_string.clone(),
439            privileges,
440            empty_string,
441            Arc::new(self.greptime_data_types.finish()),
442            Arc::new(self.data_types.finish()),
443            Arc::new(self.semantic_types.finish()),
444            Arc::new(self.column_defaults.finish()),
445            Arc::new(self.is_nullables.finish()),
446            Arc::new(self.column_types.finish()),
447            Arc::new(self.column_comments.finish()),
448            srs_ids,
449        ];
450
451        RecordBatch::new(self.schema.clone(), columns).context(CreateRecordBatchSnafu)
452    }
453}
454
455impl DfPartitionStream for InformationSchemaColumns {
456    fn schema(&self) -> &ArrowSchemaRef {
457        self.schema.arrow_schema()
458    }
459
460    fn execute(&self, _: Arc<TaskContext>) -> DfSendableRecordBatchStream {
461        let schema = self.schema.arrow_schema().clone();
462        let mut builder = self.builder();
463        Box::pin(DfRecordBatchStreamAdapter::new(
464            schema,
465            futures::stream::once(async move {
466                builder
467                    .make_columns(None)
468                    .await
469                    .map(|x| x.into_df_record_batch())
470                    .map_err(Into::into)
471            }),
472        ))
473    }
474}
catalog/system_schema/information_schema/columns.rs

catalog/system_schema/information_schema/
columns.rs
