catalog/system_schema/information_schema/
views.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_VIEW_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, Schema, SchemaRef};
28use datatypes::value::Value;
29use datatypes::vectors::StringVectorBuilder;
30use futures::TryStreamExt;
31use snafu::{OptionExt, ResultExt};
32use store_api::storage::{ScanRequest, TableId};
33use table::metadata::TableType;
34
35use crate::CatalogManager;
36use crate::error::{
37    CastManagerSnafu, CreateRecordBatchSnafu, GetViewCacheSnafu, InternalSnafu, Result,
38    UpgradeWeakCatalogManagerRefSnafu, ViewInfoNotFoundSnafu,
39};
40use crate::kvbackend::KvBackendCatalogManager;
41use crate::system_schema::SystemTable;
42use crate::system_schema::information_schema::{
43    InformationSchemaTableScope, InformationTable, Predicates, VIEWS,
44};
45const INIT_CAPACITY: usize = 42;
46
47pub const TABLE_CATALOG: &str = "table_catalog";
48pub const TABLE_SCHEMA: &str = "table_schema";
49pub const TABLE_NAME: &str = "table_name";
50pub const VIEW_DEFINITION: &str = "view_definition";
51pub const CHECK_OPTION: &str = "check_option";
52pub const IS_UPDATABLE: &str = "is_updatable";
53pub const DEFINER: &str = "definer";
54pub const SECURITY_TYPE: &str = "security_type";
55pub const CHARACTER_SET_CLIENT: &str = "character_set_client";
56pub const COLLATION_CONNECTION: &str = "collation_connection";
57
58/// The `information_schema.views` to provides information about views in databases.
59#[derive(Debug)]
60pub(super) struct InformationSchemaViews {
61    schema: SchemaRef,
62    catalog_name: String,
63    catalog_manager: Weak<dyn CatalogManager>,
64}
65
66impl InformationSchemaViews {
67    pub(super) fn new(catalog_name: String, catalog_manager: Weak<dyn CatalogManager>) -> Self {
68        Self {
69            schema: Self::schema(),
70            catalog_name,
71            catalog_manager,
72        }
73    }
74
75    pub(crate) fn schema() -> SchemaRef {
76        Arc::new(Schema::new(vec![
77            ColumnSchema::new(TABLE_CATALOG, ConcreteDataType::string_datatype(), false),
78            ColumnSchema::new(TABLE_SCHEMA, ConcreteDataType::string_datatype(), false),
79            ColumnSchema::new(TABLE_NAME, ConcreteDataType::string_datatype(), false),
80            ColumnSchema::new(VIEW_DEFINITION, ConcreteDataType::string_datatype(), false),
81            ColumnSchema::new(CHECK_OPTION, ConcreteDataType::string_datatype(), true),
82            ColumnSchema::new(IS_UPDATABLE, ConcreteDataType::string_datatype(), true),
83            ColumnSchema::new(DEFINER, ConcreteDataType::string_datatype(), true),
84            ColumnSchema::new(SECURITY_TYPE, ConcreteDataType::string_datatype(), true),
85            ColumnSchema::new(
86                CHARACTER_SET_CLIENT,
87                ConcreteDataType::string_datatype(),
88                true,
89            ),
90            ColumnSchema::new(
91                COLLATION_CONNECTION,
92                ConcreteDataType::string_datatype(),
93                true,
94            ),
95        ]))
96    }
97
98    fn builder(&self) -> InformationSchemaViewsBuilder {
99        InformationSchemaViewsBuilder::new(
100            self.schema.clone(),
101            self.catalog_name.clone(),
102            self.catalog_manager.clone(),
103        )
104    }
105}
106
107impl InformationTable for InformationSchemaViews {
108    fn scope(&self) -> InformationSchemaTableScope {
109        InformationSchemaTableScope::Catalog
110    }
111}
112
113impl SystemTable for InformationSchemaViews {
114    fn table_id(&self) -> TableId {
115        INFORMATION_SCHEMA_VIEW_TABLE_ID
116    }
117
118    fn table_name(&self) -> &'static str {
119        VIEWS
120    }
121
122    fn schema(&self) -> SchemaRef {
123        self.schema.clone()
124    }
125
126    fn to_stream(&self, request: ScanRequest) -> Result<SendableRecordBatchStream> {
127        let schema = self.schema.arrow_schema().clone();
128        let mut builder = self.builder();
129        let stream = Box::pin(DfRecordBatchStreamAdapter::new(
130            schema,
131            futures::stream::once(async move {
132                builder
133                    .make_views(Some(request))
134                    .await
135                    .map(|x| x.into_df_record_batch())
136                    .map_err(|err| datafusion::error::DataFusionError::External(Box::new(err)))
137            }),
138        ));
139        Ok(Box::pin(
140            RecordBatchStreamAdapter::try_new(stream)
141                .map_err(BoxedError::new)
142                .context(InternalSnafu)?,
143        ))
144    }
145}
146
147/// Builds the `information_schema.VIEWS` table row by row
148///
149/// Columns are based on <https://dev.mysql.com/doc/refman/8.4/en/information-schema-views-table.html>
150struct InformationSchemaViewsBuilder {
151    schema: SchemaRef,
152    catalog_name: String,
153    catalog_manager: Weak<dyn CatalogManager>,
154
155    catalog_names: StringVectorBuilder,
156    schema_names: StringVectorBuilder,
157    table_names: StringVectorBuilder,
158    view_definitions: StringVectorBuilder,
159    check_options: StringVectorBuilder,
160    is_updatable: StringVectorBuilder,
161    definer: StringVectorBuilder,
162    security_type: StringVectorBuilder,
163    character_set_client: StringVectorBuilder,
164    collation_connection: StringVectorBuilder,
165}
166
167impl InformationSchemaViewsBuilder {
168    fn new(
169        schema: SchemaRef,
170        catalog_name: String,
171        catalog_manager: Weak<dyn CatalogManager>,
172    ) -> Self {
173        Self {
174            schema,
175            catalog_name,
176            catalog_manager,
177            catalog_names: StringVectorBuilder::with_capacity(INIT_CAPACITY),
178            schema_names: StringVectorBuilder::with_capacity(INIT_CAPACITY),
179            table_names: StringVectorBuilder::with_capacity(INIT_CAPACITY),
180            view_definitions: StringVectorBuilder::with_capacity(INIT_CAPACITY),
181            check_options: StringVectorBuilder::with_capacity(INIT_CAPACITY),
182            is_updatable: StringVectorBuilder::with_capacity(INIT_CAPACITY),
183            definer: StringVectorBuilder::with_capacity(INIT_CAPACITY),
184            security_type: StringVectorBuilder::with_capacity(INIT_CAPACITY),
185            character_set_client: StringVectorBuilder::with_capacity(INIT_CAPACITY),
186            collation_connection: StringVectorBuilder::with_capacity(INIT_CAPACITY),
187        }
188    }
189
190    /// Construct the `information_schema.views` virtual table
191    async fn make_views(&mut self, request: Option<ScanRequest>) -> Result<RecordBatch> {
192        let catalog_name = self.catalog_name.clone();
193        let catalog_manager = self
194            .catalog_manager
195            .upgrade()
196            .context(UpgradeWeakCatalogManagerRefSnafu)?;
197        let predicates = Predicates::from_scan_request(&request);
198        let view_info_cache = catalog_manager
199            .as_any()
200            .downcast_ref::<KvBackendCatalogManager>()
201            .context(CastManagerSnafu)?
202            .view_info_cache()?;
203
204        for schema_name in catalog_manager.schema_names(&catalog_name, None).await? {
205            let mut stream = catalog_manager.tables(&catalog_name, &schema_name, None);
206
207            while let Some(table) = stream.try_next().await? {
208                let table_info = table.table_info();
209                if table_info.table_type == TableType::View {
210                    let view_info = view_info_cache
211                        .get(table_info.ident.table_id)
212                        .await
213                        .context(GetViewCacheSnafu)?
214                        .context(ViewInfoNotFoundSnafu {
215                            name: &table_info.name,
216                        })?;
217                    self.add_view(
218                        &predicates,
219                        &catalog_name,
220                        &schema_name,
221                        &table_info.name,
222                        &view_info.definition,
223                    )
224                }
225            }
226        }
227
228        self.finish()
229    }
230
231    fn add_view(
232        &mut self,
233        predicates: &Predicates,
234        catalog_name: &str,
235        schema_name: &str,
236        table_name: &str,
237        definition: &str,
238    ) {
239        let row = [
240            (TABLE_CATALOG, &Value::from(catalog_name)),
241            (TABLE_SCHEMA, &Value::from(schema_name)),
242            (TABLE_NAME, &Value::from(table_name)),
243        ];
244
245        if !predicates.eval(&row) {
246            return;
247        }
248        self.catalog_names.push(Some(catalog_name));
249        self.schema_names.push(Some(schema_name));
250        self.table_names.push(Some(table_name));
251        self.view_definitions.push(Some(definition));
252        self.check_options.push(None);
253        // View is not updatable, statements such UPDATE , DELETE , and INSERT are illegal and are rejected.
254        self.is_updatable.push(Some("NO"));
255        self.definer.push(None);
256        self.security_type.push(None);
257        self.character_set_client.push(Some("utf8"));
258        self.collation_connection.push(Some("utf8_bin"));
259    }
260
261    fn finish(&mut self) -> Result<RecordBatch> {
262        let columns: Vec<VectorRef> = vec![
263            Arc::new(self.catalog_names.finish()),
264            Arc::new(self.schema_names.finish()),
265            Arc::new(self.table_names.finish()),
266            Arc::new(self.view_definitions.finish()),
267            Arc::new(self.check_options.finish()),
268            Arc::new(self.is_updatable.finish()),
269            Arc::new(self.definer.finish()),
270            Arc::new(self.security_type.finish()),
271            Arc::new(self.character_set_client.finish()),
272            Arc::new(self.collation_connection.finish()),
273        ];
274        RecordBatch::new(self.schema.clone(), columns).context(CreateRecordBatchSnafu)
275    }
276}
277
278impl DfPartitionStream for InformationSchemaViews {
279    fn schema(&self) -> &ArrowSchemaRef {
280        self.schema.arrow_schema()
281    }
282
283    fn execute(&self, _: Arc<TaskContext>) -> DfSendableRecordBatchStream {
284        let schema = self.schema.arrow_schema().clone();
285        let mut builder = self.builder();
286        Box::pin(DfRecordBatchStreamAdapter::new(
287            schema,
288            futures::stream::once(async move {
289                builder
290                    .make_views(None)
291                    .await
292                    .map(|x| x.into_df_record_batch())
293                    .map_err(Into::into)
294            }),
295        ))
296    }
297}
catalog/system_schema/information_schema/views.rs

catalog/system_schema/information_schema/
views.rs
