catalog/system_schema/information_schema/
tables.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 core::pin::pin;
16use std::sync::{Arc, Weak};
17
18use arrow_schema::SchemaRef as ArrowSchemaRef;
19use common_catalog::consts::{INFORMATION_SCHEMA_TABLES_TABLE_ID, MITO_ENGINE};
20use common_error::ext::BoxedError;
21use common_meta::datanode::RegionStat;
22use common_recordbatch::adapter::RecordBatchStreamAdapter;
23use common_recordbatch::{RecordBatch, SendableRecordBatchStream};
24use common_telemetry::error;
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, ScalarVectorBuilder, VectorRef};
30use datatypes::schema::{ColumnSchema, Schema, SchemaRef};
31use datatypes::value::Value;
32use datatypes::vectors::{
33    StringVectorBuilder, TimestampSecondVectorBuilder, UInt32VectorBuilder, UInt64VectorBuilder,
34};
35use futures::StreamExt;
36use snafu::{OptionExt, ResultExt};
37use store_api::storage::{ScanRequest, TableId};
38use table::metadata::{TableInfo, TableType};
39
40use crate::CatalogManager;
41use crate::error::{
42    CreateRecordBatchSnafu, FindRegionRoutesSnafu, InternalSnafu, Result,
43    UpgradeWeakCatalogManagerRefSnafu,
44};
45use crate::kvbackend::KvBackendCatalogManager;
46use crate::system_schema::SystemTable;
47use crate::system_schema::information_schema::{
48    InformationSchemaTableScope, InformationTable, Predicates, TABLES,
49};
50use crate::system_schema::utils;
51
52pub const TABLE_CATALOG: &str = "table_catalog";
53pub const TABLE_SCHEMA: &str = "table_schema";
54pub const TABLE_NAME: &str = "table_name";
55pub const TABLE_TYPE: &str = "table_type";
56pub const VERSION: &str = "version";
57pub const ROW_FORMAT: &str = "row_format";
58pub const TABLE_ROWS: &str = "table_rows";
59pub const DATA_LENGTH: &str = "data_length";
60pub const INDEX_LENGTH: &str = "index_length";
61pub const MAX_DATA_LENGTH: &str = "max_data_length";
62pub const AVG_ROW_LENGTH: &str = "avg_row_length";
63pub const DATA_FREE: &str = "data_free";
64pub const AUTO_INCREMENT: &str = "auto_increment";
65pub const CREATE_TIME: &str = "create_time";
66pub const UPDATE_TIME: &str = "update_time";
67pub const CHECK_TIME: &str = "check_time";
68pub const TABLE_COLLATION: &str = "table_collation";
69pub const CHECKSUM: &str = "checksum";
70pub const CREATE_OPTIONS: &str = "create_options";
71pub const TABLE_COMMENT: &str = "table_comment";
72pub const MAX_INDEX_LENGTH: &str = "max_index_length";
73pub const TEMPORARY: &str = "temporary";
74const TABLE_ID: &str = "table_id";
75pub const ENGINE: &str = "engine";
76const INIT_CAPACITY: usize = 42;
77
78#[derive(Debug)]
79pub(super) struct InformationSchemaTables {
80    schema: SchemaRef,
81    catalog_name: String,
82    catalog_manager: Weak<dyn CatalogManager>,
83}
84
85impl InformationSchemaTables {
86    pub(super) fn new(catalog_name: String, catalog_manager: Weak<dyn CatalogManager>) -> Self {
87        Self {
88            schema: Self::schema(),
89            catalog_name,
90            catalog_manager,
91        }
92    }
93
94    pub(crate) fn schema() -> SchemaRef {
95        Arc::new(Schema::new(vec![
96            ColumnSchema::new(TABLE_CATALOG, ConcreteDataType::string_datatype(), false),
97            ColumnSchema::new(TABLE_SCHEMA, ConcreteDataType::string_datatype(), false),
98            ColumnSchema::new(TABLE_NAME, ConcreteDataType::string_datatype(), false),
99            ColumnSchema::new(TABLE_TYPE, ConcreteDataType::string_datatype(), false),
100            ColumnSchema::new(TABLE_ID, ConcreteDataType::uint32_datatype(), true),
101            ColumnSchema::new(DATA_LENGTH, ConcreteDataType::uint64_datatype(), true),
102            ColumnSchema::new(MAX_DATA_LENGTH, ConcreteDataType::uint64_datatype(), true),
103            ColumnSchema::new(INDEX_LENGTH, ConcreteDataType::uint64_datatype(), true),
104            ColumnSchema::new(MAX_INDEX_LENGTH, ConcreteDataType::uint64_datatype(), true),
105            ColumnSchema::new(AVG_ROW_LENGTH, ConcreteDataType::uint64_datatype(), true),
106            ColumnSchema::new(ENGINE, ConcreteDataType::string_datatype(), true),
107            ColumnSchema::new(VERSION, ConcreteDataType::uint64_datatype(), true),
108            ColumnSchema::new(ROW_FORMAT, ConcreteDataType::string_datatype(), true),
109            ColumnSchema::new(TABLE_ROWS, ConcreteDataType::uint64_datatype(), true),
110            ColumnSchema::new(DATA_FREE, ConcreteDataType::uint64_datatype(), true),
111            ColumnSchema::new(AUTO_INCREMENT, ConcreteDataType::uint64_datatype(), true),
112            ColumnSchema::new(
113                CREATE_TIME,
114                ConcreteDataType::timestamp_second_datatype(),
115                true,
116            ),
117            ColumnSchema::new(
118                UPDATE_TIME,
119                ConcreteDataType::timestamp_second_datatype(),
120                true,
121            ),
122            ColumnSchema::new(
123                CHECK_TIME,
124                ConcreteDataType::timestamp_second_datatype(),
125                true,
126            ),
127            ColumnSchema::new(TABLE_COLLATION, ConcreteDataType::string_datatype(), true),
128            ColumnSchema::new(CHECKSUM, ConcreteDataType::uint64_datatype(), true),
129            ColumnSchema::new(CREATE_OPTIONS, ConcreteDataType::string_datatype(), true),
130            ColumnSchema::new(TABLE_COMMENT, ConcreteDataType::string_datatype(), true),
131            ColumnSchema::new(TEMPORARY, ConcreteDataType::string_datatype(), true),
132        ]))
133    }
134
135    fn builder(&self) -> InformationSchemaTablesBuilder {
136        InformationSchemaTablesBuilder::new(
137            self.schema.clone(),
138            self.catalog_name.clone(),
139            self.catalog_manager.clone(),
140        )
141    }
142}
143
144impl InformationTable for InformationSchemaTables {
145    fn scope(&self) -> InformationSchemaTableScope {
146        InformationSchemaTableScope::Catalog
147    }
148}
149
150impl SystemTable for InformationSchemaTables {
151    fn table_id(&self) -> TableId {
152        INFORMATION_SCHEMA_TABLES_TABLE_ID
153    }
154
155    fn table_name(&self) -> &'static str {
156        TABLES
157    }
158
159    fn schema(&self) -> SchemaRef {
160        self.schema.clone()
161    }
162
163    fn to_stream(&self, request: ScanRequest) -> Result<SendableRecordBatchStream> {
164        let schema = self.schema.arrow_schema().clone();
165        let mut builder = self.builder();
166        let stream = Box::pin(DfRecordBatchStreamAdapter::new(
167            schema,
168            futures::stream::once(async move {
169                builder
170                    .make_tables(Some(request))
171                    .await
172                    .map(|x| x.into_df_record_batch())
173                    .map_err(|err| datafusion::error::DataFusionError::External(Box::new(err)))
174            }),
175        ));
176        Ok(Box::pin(
177            RecordBatchStreamAdapter::try_new(stream)
178                .map_err(BoxedError::new)
179                .context(InternalSnafu)?,
180        ))
181    }
182}
183
184/// Builds the `information_schema.TABLE` table row by row
185///
186/// Columns are based on <https://www.postgresql.org/docs/current/infoschema-columns.html>
187struct InformationSchemaTablesBuilder {
188    schema: SchemaRef,
189    catalog_name: String,
190    catalog_manager: Weak<dyn CatalogManager>,
191
192    catalog_names: StringVectorBuilder,
193    schema_names: StringVectorBuilder,
194    table_names: StringVectorBuilder,
195    table_types: StringVectorBuilder,
196    table_ids: UInt32VectorBuilder,
197    version: UInt64VectorBuilder,
198    row_format: StringVectorBuilder,
199    table_rows: UInt64VectorBuilder,
200    data_length: UInt64VectorBuilder,
201    max_data_length: UInt64VectorBuilder,
202    index_length: UInt64VectorBuilder,
203    avg_row_length: UInt64VectorBuilder,
204    max_index_length: UInt64VectorBuilder,
205    data_free: UInt64VectorBuilder,
206    auto_increment: UInt64VectorBuilder,
207    create_time: TimestampSecondVectorBuilder,
208    update_time: TimestampSecondVectorBuilder,
209    check_time: TimestampSecondVectorBuilder,
210    table_collation: StringVectorBuilder,
211    checksum: UInt64VectorBuilder,
212    create_options: StringVectorBuilder,
213    table_comment: StringVectorBuilder,
214    engines: StringVectorBuilder,
215    temporary: StringVectorBuilder,
216}
217
218impl InformationSchemaTablesBuilder {
219    fn new(
220        schema: SchemaRef,
221        catalog_name: String,
222        catalog_manager: Weak<dyn CatalogManager>,
223    ) -> Self {
224        Self {
225            schema,
226            catalog_name,
227            catalog_manager,
228            catalog_names: StringVectorBuilder::with_capacity(INIT_CAPACITY),
229            schema_names: StringVectorBuilder::with_capacity(INIT_CAPACITY),
230            table_names: StringVectorBuilder::with_capacity(INIT_CAPACITY),
231            table_types: StringVectorBuilder::with_capacity(INIT_CAPACITY),
232            table_ids: UInt32VectorBuilder::with_capacity(INIT_CAPACITY),
233            data_length: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
234            max_data_length: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
235            index_length: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
236            avg_row_length: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
237            engines: StringVectorBuilder::with_capacity(INIT_CAPACITY),
238            version: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
239            row_format: StringVectorBuilder::with_capacity(INIT_CAPACITY),
240            table_rows: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
241            max_index_length: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
242            data_free: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
243            auto_increment: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
244            create_time: TimestampSecondVectorBuilder::with_capacity(INIT_CAPACITY),
245            update_time: TimestampSecondVectorBuilder::with_capacity(INIT_CAPACITY),
246            check_time: TimestampSecondVectorBuilder::with_capacity(INIT_CAPACITY),
247            table_collation: StringVectorBuilder::with_capacity(INIT_CAPACITY),
248            checksum: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
249            create_options: StringVectorBuilder::with_capacity(INIT_CAPACITY),
250            table_comment: StringVectorBuilder::with_capacity(INIT_CAPACITY),
251            temporary: StringVectorBuilder::with_capacity(INIT_CAPACITY),
252        }
253    }
254
255    /// Construct the `information_schema.tables` virtual table
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 partition_manager = catalog_manager
263            .as_any()
264            .downcast_ref::<KvBackendCatalogManager>()
265            .map(|catalog_manager| catalog_manager.partition_manager());
266        let predicates = Predicates::from_scan_request(&request);
267
268        let information_extension = utils::information_extension(&self.catalog_manager)?;
269
270        // TODO(dennis): `region_stats` API is not stable in distributed cluster because of network issue etc.
271        // But we don't want the statements such as `show tables` fail,
272        // so using `unwrap_or_else` here instead of `?` operator.
273        let region_stats = {
274            let mut x = information_extension
275                .region_stats()
276                .await
277                .unwrap_or_else(|e| {
278                    error!(e; "Failed to find region stats in information_schema, fallback to all empty");
279                    vec![]
280                });
281            x.sort_unstable_by_key(|x| x.id);
282            x
283        };
284
285        for schema_name in catalog_manager.schema_names(&catalog_name, None).await? {
286            let table_stream = catalog_manager.tables(&catalog_name, &schema_name, None);
287
288            const BATCH_SIZE: usize = 128;
289            // Split tables into chunks
290            let mut table_chunks = pin!(table_stream.ready_chunks(BATCH_SIZE));
291
292            while let Some(tables) = table_chunks.next().await {
293                let tables = tables.into_iter().collect::<Result<Vec<_>>>()?;
294                let mito_or_physical_table_ids = tables
295                    .iter()
296                    .filter(|table| {
297                        table.table_info().meta.engine == MITO_ENGINE
298                            || table.table_info().is_physical_table()
299                    })
300                    .map(|table| table.table_info().ident.table_id)
301                    .collect::<Vec<_>>();
302
303                let table_routes = if let Some(partition_manager) = &partition_manager {
304                    partition_manager
305                        .batch_find_region_routes(&mito_or_physical_table_ids)
306                        .await
307                        .context(FindRegionRoutesSnafu)?
308                } else {
309                    mito_or_physical_table_ids
310                        .into_iter()
311                        .map(|id| (id, vec![]))
312                        .collect()
313                };
314
315                for table in tables {
316                    let table_region_stats =
317                        match table_routes.get(&table.table_info().ident.table_id) {
318                            Some(routes) => routes
319                                .iter()
320                                .flat_map(|route| {
321                                    let region_id = route.region.id;
322                                    region_stats
323                                        .binary_search_by_key(&region_id, |x| x.id)
324                                        .map(|i| &region_stats[i])
325                                })
326                                .collect::<Vec<_>>(),
327                            None => vec![],
328                        };
329
330                    self.add_table(
331                        &predicates,
332                        &catalog_name,
333                        &schema_name,
334                        table.table_info(),
335                        table.table_type(),
336                        &table_region_stats,
337                    );
338                }
339            }
340        }
341
342        self.finish()
343    }
344
345    #[allow(clippy::too_many_arguments)]
346    fn add_table(
347        &mut self,
348        predicates: &Predicates,
349        catalog_name: &str,
350        schema_name: &str,
351        table_info: Arc<TableInfo>,
352        table_type: TableType,
353        region_stats: &[&RegionStat],
354    ) {
355        let table_name = table_info.name.as_ref();
356        let table_id = table_info.table_id();
357        let engine = table_info.meta.engine.as_ref();
358
359        let table_type_text = match table_type {
360            TableType::Base => "BASE TABLE",
361            TableType::View => "VIEW",
362            TableType::Temporary => "LOCAL TEMPORARY",
363        };
364
365        let row = [
366            (TABLE_CATALOG, &Value::from(catalog_name)),
367            (TABLE_ID, &Value::from(table_id)),
368            (TABLE_SCHEMA, &Value::from(schema_name)),
369            (ENGINE, &Value::from(engine)),
370            (TABLE_NAME, &Value::from(table_name)),
371            (TABLE_TYPE, &Value::from(table_type_text)),
372        ];
373
374        if !predicates.eval(&row) {
375            return;
376        }
377
378        self.catalog_names.push(Some(catalog_name));
379        self.schema_names.push(Some(schema_name));
380        self.table_names.push(Some(table_name));
381        self.table_types.push(Some(table_type_text));
382        self.table_ids.push(Some(table_id));
383
384        let data_length: u64 = region_stats.iter().map(|stat| stat.sst_size).sum();
385        let table_rows: u64 = region_stats.iter().map(|stat| stat.num_rows).sum();
386        let index_length: u64 = region_stats.iter().map(|stat| stat.index_size).sum();
387
388        let total_data_length: u64 = data_length
389            + region_stats
390                .iter()
391                .map(|stat| stat.memtable_size)
392                .sum::<u64>();
393        let avg_row_length = total_data_length.checked_div(table_rows).unwrap_or(0);
394
395        self.data_length.push(Some(data_length));
396        self.index_length.push(Some(index_length));
397        self.table_rows.push(Some(table_rows));
398        self.avg_row_length.push(Some(avg_row_length));
399
400        // TODO(sunng87): use real data for these fields
401        self.max_data_length.push(Some(0));
402        self.checksum.push(Some(0));
403        self.max_index_length.push(Some(0));
404        self.data_free.push(Some(0));
405        self.auto_increment.push(Some(0));
406        self.row_format.push(Some("Fixed"));
407        self.table_collation.push(Some("utf8_bin"));
408        self.update_time
409            .push(Some(table_info.meta.updated_on.timestamp().into()));
410        self.check_time.push(None);
411        // use mariadb default table version number here
412        self.version.push(Some(11));
413        self.table_comment.push(table_info.desc.as_deref());
414        self.create_options
415            .push(Some(table_info.meta.options.to_string().as_ref()));
416        self.create_time
417            .push(Some(table_info.meta.created_on.timestamp().into()));
418
419        self.temporary
420            .push(if matches!(table_type, TableType::Temporary) {
421                Some("Y")
422            } else {
423                Some("N")
424            });
425        self.engines.push(Some(engine));
426    }
427
428    fn finish(&mut self) -> Result<RecordBatch> {
429        let columns: Vec<VectorRef> = vec![
430            Arc::new(self.catalog_names.finish()),
431            Arc::new(self.schema_names.finish()),
432            Arc::new(self.table_names.finish()),
433            Arc::new(self.table_types.finish()),
434            Arc::new(self.table_ids.finish()),
435            Arc::new(self.data_length.finish()),
436            Arc::new(self.max_data_length.finish()),
437            Arc::new(self.index_length.finish()),
438            Arc::new(self.max_index_length.finish()),
439            Arc::new(self.avg_row_length.finish()),
440            Arc::new(self.engines.finish()),
441            Arc::new(self.version.finish()),
442            Arc::new(self.row_format.finish()),
443            Arc::new(self.table_rows.finish()),
444            Arc::new(self.data_free.finish()),
445            Arc::new(self.auto_increment.finish()),
446            Arc::new(self.create_time.finish()),
447            Arc::new(self.update_time.finish()),
448            Arc::new(self.check_time.finish()),
449            Arc::new(self.table_collation.finish()),
450            Arc::new(self.checksum.finish()),
451            Arc::new(self.create_options.finish()),
452            Arc::new(self.table_comment.finish()),
453            Arc::new(self.temporary.finish()),
454        ];
455        RecordBatch::new(self.schema.clone(), columns).context(CreateRecordBatchSnafu)
456    }
457}
458
459impl DfPartitionStream for InformationSchemaTables {
460    fn schema(&self) -> &ArrowSchemaRef {
461        self.schema.arrow_schema()
462    }
463
464    fn execute(&self, _: Arc<TaskContext>) -> DfSendableRecordBatchStream {
465        let schema = self.schema.arrow_schema().clone();
466        let mut builder = self.builder();
467        Box::pin(DfRecordBatchStreamAdapter::new(
468            schema,
469            futures::stream::once(async move {
470                builder
471                    .make_tables(None)
472                    .await
473                    .map(|x| x.into_df_record_batch())
474                    .map_err(Into::into)
475            }),
476        ))
477    }
478}
catalog/system_schema/information_schema/tables.rs

catalog/system_schema/information_schema/
tables.rs
