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

catalog/system_schema/information_schema/
statistics.rs
