catalog/system_schema/information_schema/
statistics.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::INFORMATION_SCHEMA_STATISTICS_TABLE_ID;
19use common_error::ext::BoxedError;
20use common_recordbatch::adapter::RecordBatchStreamAdapter;
21use common_recordbatch::{RecordBatch, SendableRecordBatchStream};
22use datafusion::execution::TaskContext;
23use datafusion::physical_plan::SendableRecordBatchStream as DfSendableRecordBatchStream;
24use datafusion::physical_plan::stream::RecordBatchStreamAdapter as DfRecordBatchStreamAdapter;
25use datafusion::physical_plan::streaming::PartitionStream as DfPartitionStream;
26use datatypes::prelude::{ConcreteDataType, ScalarVectorBuilder, VectorRef};
27use datatypes::schema::{ColumnSchema, FulltextBackend, Schema, SchemaRef};
28use datatypes::value::Value;
29use datatypes::vectors::{Int64VectorBuilder, StringVectorBuilder};
30use futures::TryStreamExt;
31use snafu::{OptionExt, ResultExt};
32use store_api::sst_entry::{
33    PUFFIN_INDEX_TYPE_BLOOM_FILTER, PUFFIN_INDEX_TYPE_FULLTEXT_BLOOM,
34    PUFFIN_INDEX_TYPE_FULLTEXT_TANTIVY, PUFFIN_INDEX_TYPE_INVERTED,
35};
36use store_api::storage::{ScanRequest, TableId};
37
38use crate::CatalogManager;
39use crate::error::{
40    CreateRecordBatchSnafu, InternalSnafu, Result, UpgradeWeakCatalogManagerRefSnafu,
41};
42use crate::system_schema::SystemTable;
43use crate::system_schema::information_schema::key_column_usage::{
44    CONSTRAINT_NAME_FULLTEXT_INDEX, CONSTRAINT_NAME_INVERTED_INDEX, CONSTRAINT_NAME_PRI,
45    CONSTRAINT_NAME_SKIPPING_INDEX, CONSTRAINT_NAME_TIME_INDEX,
46};
47use crate::system_schema::information_schema::{
48    InformationSchemaTableScope, InformationTable, Predicates, STATISTICS,
49    primary_key_encoding_index_type,
50};
51
52pub const TABLE_CATALOG: &str = "table_catalog";
53pub const TABLE_SCHEMA: &str = "table_schema";
54pub const TABLE_NAME: &str = "table_name";
55pub const NON_UNIQUE: &str = "non_unique";
56pub const INDEX_SCHEMA: &str = "index_schema";
57pub const INDEX_NAME: &str = "index_name";
58pub const SEQ_IN_INDEX: &str = "seq_in_index";
59pub const COLUMN_NAME: &str = "column_name";
60pub const COLLATION: &str = "collation";
61pub const CARDINALITY: &str = "cardinality";
62pub const SUB_PART: &str = "sub_part";
63pub const PACKED: &str = "packed";
64pub const NULLABLE: &str = "nullable";
65pub const INDEX_TYPE: &str = "index_type";
66pub const COMMENT: &str = "comment";
67pub const INDEX_COMMENT: &str = "index_comment";
68pub const IS_VISIBLE: &str = "is_visible";
69pub const EXPRESSION: &str = "expression";
70pub const GREPTIME_INDEX_TYPE: &str = "greptime_index_type";
71
72const INIT_CAPACITY: usize = 42;
73const MYSQL_DEFAULT_CATALOG: &str = "def";
74const ASCENDING_COLLATION: &str = "A";
75const YES: &str = "YES";
76const EMPTY: &str = "";
77const PRIMARY: &str = "PRIMARY";
78const TIME: &str = "TIME";
79const FULLTEXT: &str = "FULLTEXT";
80const INVERTED: &str = "INVERTED";
81const SKIPPING: &str = "SKIPPING";
82const SECONDARY_INDEX_SEQ: i64 = 1;
83
84/// The `information_schema.statistics` table provides index metadata in a
85/// MySQL-compatible shape.
86#[derive(Debug)]
87pub(super) struct InformationSchemaStatistics {
88    schema: SchemaRef,
89    catalog_name: String,
90    catalog_manager: Weak<dyn CatalogManager>,
91}
92
93impl InformationSchemaStatistics {
94    pub(super) fn new(catalog_name: String, catalog_manager: Weak<dyn CatalogManager>) -> Self {
95        Self {
96            schema: Self::schema(),
97            catalog_name,
98            catalog_manager,
99        }
100    }
101
102    pub(crate) fn schema() -> SchemaRef {
103        Arc::new(Schema::new(vec![
104            ColumnSchema::new(TABLE_CATALOG, ConcreteDataType::string_datatype(), false),
105            ColumnSchema::new(TABLE_SCHEMA, ConcreteDataType::string_datatype(), false),
106            ColumnSchema::new(TABLE_NAME, ConcreteDataType::string_datatype(), false),
107            ColumnSchema::new(NON_UNIQUE, ConcreteDataType::int64_datatype(), false),
108            ColumnSchema::new(INDEX_SCHEMA, ConcreteDataType::string_datatype(), false),
109            ColumnSchema::new(INDEX_NAME, ConcreteDataType::string_datatype(), false),
110            ColumnSchema::new(SEQ_IN_INDEX, ConcreteDataType::int64_datatype(), false),
111            ColumnSchema::new(COLUMN_NAME, ConcreteDataType::string_datatype(), false),
112            ColumnSchema::new(COLLATION, ConcreteDataType::string_datatype(), true),
113            ColumnSchema::new(CARDINALITY, ConcreteDataType::int64_datatype(), true),
114            ColumnSchema::new(SUB_PART, ConcreteDataType::int64_datatype(), true),
115            ColumnSchema::new(PACKED, ConcreteDataType::string_datatype(), true),
116            ColumnSchema::new(NULLABLE, ConcreteDataType::string_datatype(), false),
117            ColumnSchema::new(INDEX_TYPE, ConcreteDataType::string_datatype(), false),
118            ColumnSchema::new(COMMENT, ConcreteDataType::string_datatype(), false),
119            ColumnSchema::new(INDEX_COMMENT, ConcreteDataType::string_datatype(), false),
120            ColumnSchema::new(IS_VISIBLE, ConcreteDataType::string_datatype(), false),
121            ColumnSchema::new(EXPRESSION, ConcreteDataType::string_datatype(), true),
122            ColumnSchema::new(
123                GREPTIME_INDEX_TYPE,
124                ConcreteDataType::string_datatype(),
125                true,
126            ),
127        ]))
128    }
129
130    fn builder(&self) -> InformationSchemaStatisticsBuilder {
131        InformationSchemaStatisticsBuilder::new(
132            self.schema.clone(),
133            self.catalog_name.clone(),
134            self.catalog_manager.clone(),
135        )
136    }
137}
138
139impl InformationTable for InformationSchemaStatistics {
140    fn scope(&self) -> InformationSchemaTableScope {
141        InformationSchemaTableScope::Catalog
142    }
143}
144
145impl SystemTable for InformationSchemaStatistics {
146    fn table_id(&self) -> TableId {
147        INFORMATION_SCHEMA_STATISTICS_TABLE_ID
148    }
149
150    fn table_name(&self) -> &'static str {
151        STATISTICS
152    }
153
154    fn schema(&self) -> SchemaRef {
155        self.schema.clone()
156    }
157
158    fn to_stream(&self, request: ScanRequest) -> Result<SendableRecordBatchStream> {
159        let schema = self.schema.arrow_schema().clone();
160        let mut builder = self.builder();
161        let stream = Box::pin(DfRecordBatchStreamAdapter::new(
162            schema,
163            futures::stream::once(async move {
164                builder
165                    .make_statistics(Some(request))
166                    .await
167                    .map(|x| x.into_df_record_batch())
168                    .map_err(Into::into)
169            }),
170        ));
171        Ok(Box::pin(
172            RecordBatchStreamAdapter::try_new(stream)
173                .map_err(BoxedError::new)
174                .context(InternalSnafu)?,
175        ))
176    }
177}
178
179struct InformationSchemaStatisticsBuilder {
180    schema: SchemaRef,
181    catalog_name: String,
182    catalog_manager: Weak<dyn CatalogManager>,
183
184    table_catalogs: StringVectorBuilder,
185    table_schemas: StringVectorBuilder,
186    table_names: StringVectorBuilder,
187    non_unique: Int64VectorBuilder,
188    index_schemas: StringVectorBuilder,
189    index_names: StringVectorBuilder,
190    seq_in_index: Int64VectorBuilder,
191    column_names: StringVectorBuilder,
192    collations: StringVectorBuilder,
193    cardinalities: Int64VectorBuilder,
194    sub_parts: Int64VectorBuilder,
195    packed: StringVectorBuilder,
196    nullable: StringVectorBuilder,
197    index_types: StringVectorBuilder,
198    comments: StringVectorBuilder,
199    index_comments: StringVectorBuilder,
200    is_visible: StringVectorBuilder,
201    expressions: StringVectorBuilder,
202    greptime_index_types: StringVectorBuilder,
203}
204
205impl InformationSchemaStatisticsBuilder {
206    fn new(
207        schema: SchemaRef,
208        catalog_name: String,
209        catalog_manager: Weak<dyn CatalogManager>,
210    ) -> Self {
211        Self {
212            schema,
213            catalog_name,
214            catalog_manager,
215            table_catalogs: StringVectorBuilder::with_capacity(INIT_CAPACITY),
216            table_schemas: StringVectorBuilder::with_capacity(INIT_CAPACITY),
217            table_names: StringVectorBuilder::with_capacity(INIT_CAPACITY),
218            non_unique: Int64VectorBuilder::with_capacity(INIT_CAPACITY),
219            index_schemas: StringVectorBuilder::with_capacity(INIT_CAPACITY),
220            index_names: StringVectorBuilder::with_capacity(INIT_CAPACITY),
221            seq_in_index: Int64VectorBuilder::with_capacity(INIT_CAPACITY),
222            column_names: StringVectorBuilder::with_capacity(INIT_CAPACITY),
223            collations: StringVectorBuilder::with_capacity(INIT_CAPACITY),
224            cardinalities: Int64VectorBuilder::with_capacity(INIT_CAPACITY),
225            sub_parts: Int64VectorBuilder::with_capacity(INIT_CAPACITY),
226            packed: StringVectorBuilder::with_capacity(INIT_CAPACITY),
227            nullable: StringVectorBuilder::with_capacity(INIT_CAPACITY),
228            index_types: StringVectorBuilder::with_capacity(INIT_CAPACITY),
229            comments: StringVectorBuilder::with_capacity(INIT_CAPACITY),
230            index_comments: StringVectorBuilder::with_capacity(INIT_CAPACITY),
231            is_visible: StringVectorBuilder::with_capacity(INIT_CAPACITY),
232            expressions: StringVectorBuilder::with_capacity(INIT_CAPACITY),
233            greptime_index_types: StringVectorBuilder::with_capacity(INIT_CAPACITY),
234        }
235    }
236
237    async fn make_statistics(&mut self, request: Option<ScanRequest>) -> Result<RecordBatch> {
238        let catalog_name = self.catalog_name.clone();
239        let catalog_manager = self
240            .catalog_manager
241            .upgrade()
242            .context(UpgradeWeakCatalogManagerRefSnafu)?;
243        let predicates = Predicates::from_scan_request(&request);
244
245        for schema_name in catalog_manager.schema_names(&catalog_name, None).await? {
246            let mut stream = catalog_manager.tables(&catalog_name, &schema_name, None);
247
248            while let Some(table) = stream.try_next().await? {
249                let table_info = table.table_info();
250                let table_name = &table_info.name;
251                let keys = &table_info.meta.primary_key_indices;
252                let schema = table.schema();
253                let primary_key_encoding =
254                    primary_key_encoding_index_type(&table_info.meta.options.extra_options);
255
256                for (primary_key_seq, idx) in keys.iter().enumerate() {
257                    if let Some(column) = schema.column_schemas().get(*idx) {
258                        self.add_statistics(
259                            &predicates,
260                            &schema_name,
261                            table_name,
262                            0,
263                            CONSTRAINT_NAME_PRI,
264                            primary_key_seq as i64 + 1,
265                            column,
266                            PRIMARY,
267                            primary_key_encoding,
268                        );
269                    }
270                }
271
272                for column in schema.column_schemas() {
273                    if column.is_time_index() {
274                        self.add_statistics(
275                            &predicates,
276                            &schema_name,
277                            table_name,
278                            1,
279                            CONSTRAINT_NAME_TIME_INDEX,
280                            1,
281                            column,
282                            TIME,
283                            EMPTY,
284                        );
285                    }
286                    if column.is_inverted_indexed() {
287                        let index_name =
288                            column_index_name(CONSTRAINT_NAME_INVERTED_INDEX, &column.name);
289                        self.add_statistics(
290                            &predicates,
291                            &schema_name,
292                            table_name,
293                            1,
294                            &index_name,
295                            SECONDARY_INDEX_SEQ,
296                            column,
297                            INVERTED,
298                            PUFFIN_INDEX_TYPE_INVERTED,
299                        );
300                    }
301                    if let Ok(Some(options)) = column.fulltext_options()
302                        && options.enable
303                    {
304                        let greptime_index_type = match options.backend {
305                            FulltextBackend::Bloom => PUFFIN_INDEX_TYPE_FULLTEXT_BLOOM,
306                            FulltextBackend::Tantivy => PUFFIN_INDEX_TYPE_FULLTEXT_TANTIVY,
307                        };
308                        let index_name =
309                            column_index_name(CONSTRAINT_NAME_FULLTEXT_INDEX, &column.name);
310                        self.add_statistics(
311                            &predicates,
312                            &schema_name,
313                            table_name,
314                            1,
315                            &index_name,
316                            SECONDARY_INDEX_SEQ,
317                            column,
318                            FULLTEXT,
319                            greptime_index_type,
320                        );
321                    }
322                    if column.is_skipping_indexed() {
323                        let index_name =
324                            column_index_name(CONSTRAINT_NAME_SKIPPING_INDEX, &column.name);
325                        self.add_statistics(
326                            &predicates,
327                            &schema_name,
328                            table_name,
329                            1,
330                            &index_name,
331                            SECONDARY_INDEX_SEQ,
332                            column,
333                            SKIPPING,
334                            PUFFIN_INDEX_TYPE_BLOOM_FILTER,
335                        );
336                    }
337                }
338            }
339        }
340
341        self.finish()
342    }
343
344    #[allow(clippy::too_many_arguments)]
345    fn add_statistics(
346        &mut self,
347        predicates: &Predicates,
348        table_schema: &str,
349        table_name: &str,
350        non_unique: i64,
351        index_name: &str,
352        seq_in_index: i64,
353        column: &ColumnSchema,
354        index_type: &str,
355        greptime_index_type: &str,
356    ) {
357        // MySQL's `information_schema.statistics.nullable` is `YES` or `''`,
358        // unlike `columns.is_nullable` which is `YES`/`NO`.
359        let nullable = if column.is_nullable() { YES } else { EMPTY };
360        let row = [
361            (TABLE_CATALOG, &Value::from(MYSQL_DEFAULT_CATALOG)),
362            (TABLE_SCHEMA, &Value::from(table_schema)),
363            (TABLE_NAME, &Value::from(table_name)),
364            (NON_UNIQUE, &Value::from(non_unique)),
365            (INDEX_SCHEMA, &Value::from(table_schema)),
366            (INDEX_NAME, &Value::from(index_name)),
367            (SEQ_IN_INDEX, &Value::from(seq_in_index)),
368            (COLUMN_NAME, &Value::from(column.name.as_str())),
369            (NULLABLE, &Value::from(nullable)),
370            (INDEX_TYPE, &Value::from(index_type)),
371            (GREPTIME_INDEX_TYPE, &Value::from(greptime_index_type)),
372        ];
373
374        if !predicates.eval(&row) {
375            return;
376        }
377
378        self.table_catalogs.push(Some(MYSQL_DEFAULT_CATALOG));
379        self.table_schemas.push(Some(table_schema));
380        self.table_names.push(Some(table_name));
381        self.non_unique.push(Some(non_unique));
382        self.index_schemas.push(Some(table_schema));
383        self.index_names.push(Some(index_name));
384        self.seq_in_index.push(Some(seq_in_index));
385        self.column_names.push(Some(&column.name));
386        self.collations.push(Some(ASCENDING_COLLATION));
387        self.cardinalities.push(None);
388        self.sub_parts.push(None);
389        self.packed.push(None);
390        self.nullable.push(Some(nullable));
391        self.index_types.push(Some(index_type));
392        self.comments.push(Some(EMPTY));
393        self.index_comments.push(Some(EMPTY));
394        self.is_visible.push(Some(YES));
395        self.expressions.push(None);
396        self.greptime_index_types.push(Some(greptime_index_type));
397    }
398
399    fn finish(&mut self) -> Result<RecordBatch> {
400        let columns: Vec<VectorRef> = vec![
401            Arc::new(self.table_catalogs.finish()),
402            Arc::new(self.table_schemas.finish()),
403            Arc::new(self.table_names.finish()),
404            Arc::new(self.non_unique.finish()),
405            Arc::new(self.index_schemas.finish()),
406            Arc::new(self.index_names.finish()),
407            Arc::new(self.seq_in_index.finish()),
408            Arc::new(self.column_names.finish()),
409            Arc::new(self.collations.finish()),
410            Arc::new(self.cardinalities.finish()),
411            Arc::new(self.sub_parts.finish()),
412            Arc::new(self.packed.finish()),
413            Arc::new(self.nullable.finish()),
414            Arc::new(self.index_types.finish()),
415            Arc::new(self.comments.finish()),
416            Arc::new(self.index_comments.finish()),
417            Arc::new(self.is_visible.finish()),
418            Arc::new(self.expressions.finish()),
419            Arc::new(self.greptime_index_types.finish()),
420        ];
421        RecordBatch::new(self.schema.clone(), columns).context(CreateRecordBatchSnafu)
422    }
423}
424
425fn column_index_name(index_name: &str, column_name: &str) -> String {
426    format!("{}_{}", index_name.replace(' ', "_"), column_name)
427}
428
429impl DfPartitionStream for InformationSchemaStatistics {
430    fn schema(&self) -> &ArrowSchemaRef {
431        self.schema.arrow_schema()
432    }
433
434    fn execute(&self, _: Arc<TaskContext>) -> DfSendableRecordBatchStream {
435        let schema = self.schema.arrow_schema().clone();
436        let mut builder = self.builder();
437        Box::pin(DfRecordBatchStreamAdapter::new(
438            schema,
439            futures::stream::once(async move {
440                builder
441                    .make_statistics(None)
442                    .await
443                    .map(|x| x.into_df_record_batch())
444                    .map_err(Into::into)
445            }),
446        ))
447    }
448}
catalog/system_schema/information_schema/statistics.rs

catalog/system_schema/information_schema/
statistics.rs
