catalog/system_schema/information_schema/
ssts.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 common_catalog::consts::{
18    INFORMATION_SCHEMA_SSTS_INDEX_META_TABLE_ID, INFORMATION_SCHEMA_SSTS_MANIFEST_TABLE_ID,
19    INFORMATION_SCHEMA_SSTS_STORAGE_TABLE_ID,
20};
21use common_error::ext::BoxedError;
22use common_recordbatch::SendableRecordBatchStream;
23use common_recordbatch::adapter::AsyncRecordBatchStreamAdapter;
24use datafusion::physical_plan::ExecutionPlan;
25use datatypes::schema::SchemaRef;
26use snafu::ResultExt;
27use store_api::sst_entry::{ManifestSstEntry, PuffinIndexMetaEntry, StorageSstEntry};
28use store_api::storage::{ScanRequest, TableId};
29
30use crate::CatalogManager;
31use crate::error::Result;
32use crate::information_schema::{
33    DatanodeInspectKind, InformationSchemaTableScope, InformationTable, SSTS_INDEX_META,
34    SSTS_MANIFEST, SSTS_STORAGE,
35};
36use crate::system_schema::SystemTable;
37use crate::system_schema::utils;
38
39/// Information schema table for sst manifest.
40pub struct InformationSchemaSstsManifest {
41    schema: SchemaRef,
42    catalog_manager: Weak<dyn CatalogManager>,
43}
44
45impl InformationSchemaSstsManifest {
46    pub(super) fn new(catalog_manager: Weak<dyn CatalogManager>) -> Self {
47        Self {
48            schema: ManifestSstEntry::schema(),
49            catalog_manager,
50        }
51    }
52}
53
54impl InformationTable for InformationSchemaSstsManifest {
55    fn scope(&self) -> InformationSchemaTableScope {
56        InformationSchemaTableScope::Global
57    }
58}
59
60impl SystemTable for InformationSchemaSstsManifest {
61    fn table_id(&self) -> TableId {
62        INFORMATION_SCHEMA_SSTS_MANIFEST_TABLE_ID
63    }
64
65    fn table_name(&self) -> &'static str {
66        SSTS_MANIFEST
67    }
68
69    fn schema(&self) -> SchemaRef {
70        self.schema.clone()
71    }
72
73    fn to_stream(&self, request: ScanRequest) -> Result<SendableRecordBatchStream> {
74        let schema = utils::projected_schema(&self.schema, &request)?;
75        let info_ext = utils::information_extension(&self.catalog_manager)?;
76        let req = utils::inspect_request(DatanodeInspectKind::SstManifest, &self.schema, request)?;
77
78        let future = async move {
79            info_ext
80                .inspect_datanode(req)
81                .await
82                .map_err(BoxedError::new)
83                .context(common_recordbatch::error::ExternalSnafu)
84        };
85        Ok(Box::pin(AsyncRecordBatchStreamAdapter::new(
86            schema,
87            Box::pin(future),
88        )))
89    }
90
91    fn scan_plan(&self, request: ScanRequest) -> Result<Option<Arc<dyn ExecutionPlan>>> {
92        utils::inspect_plan(
93            &self.catalog_manager,
94            DatanodeInspectKind::SstManifest,
95            &self.schema,
96            request,
97        )
98    }
99}
100
101/// Information schema table for sst storage.
102pub struct InformationSchemaSstsStorage {
103    schema: SchemaRef,
104    catalog_manager: Weak<dyn CatalogManager>,
105}
106
107impl InformationSchemaSstsStorage {
108    pub(super) fn new(catalog_manager: Weak<dyn CatalogManager>) -> Self {
109        Self {
110            schema: StorageSstEntry::schema(),
111            catalog_manager,
112        }
113    }
114}
115
116impl InformationTable for InformationSchemaSstsStorage {
117    fn scope(&self) -> InformationSchemaTableScope {
118        InformationSchemaTableScope::Global
119    }
120}
121
122impl SystemTable for InformationSchemaSstsStorage {
123    fn table_id(&self) -> TableId {
124        INFORMATION_SCHEMA_SSTS_STORAGE_TABLE_ID
125    }
126
127    fn table_name(&self) -> &'static str {
128        SSTS_STORAGE
129    }
130
131    fn schema(&self) -> SchemaRef {
132        self.schema.clone()
133    }
134
135    fn to_stream(&self, request: ScanRequest) -> Result<SendableRecordBatchStream> {
136        let schema = utils::projected_schema(&self.schema, &request)?;
137
138        let info_ext = utils::information_extension(&self.catalog_manager)?;
139        let req = utils::inspect_request(DatanodeInspectKind::SstStorage, &self.schema, request)?;
140
141        let future = async move {
142            info_ext
143                .inspect_datanode(req)
144                .await
145                .map_err(BoxedError::new)
146                .context(common_recordbatch::error::ExternalSnafu)
147        };
148        Ok(Box::pin(AsyncRecordBatchStreamAdapter::new(
149            schema,
150            Box::pin(future),
151        )))
152    }
153
154    fn scan_plan(&self, request: ScanRequest) -> Result<Option<Arc<dyn ExecutionPlan>>> {
155        utils::inspect_plan(
156            &self.catalog_manager,
157            DatanodeInspectKind::SstStorage,
158            &self.schema,
159            request,
160        )
161    }
162}
163
164/// Information schema table for index metadata.
165pub struct InformationSchemaSstsIndexMeta {
166    schema: SchemaRef,
167    catalog_manager: Weak<dyn CatalogManager>,
168}
169
170impl InformationSchemaSstsIndexMeta {
171    pub(super) fn new(catalog_manager: Weak<dyn CatalogManager>) -> Self {
172        Self {
173            schema: PuffinIndexMetaEntry::schema(),
174            catalog_manager,
175        }
176    }
177}
178
179impl InformationTable for InformationSchemaSstsIndexMeta {
180    fn scope(&self) -> InformationSchemaTableScope {
181        InformationSchemaTableScope::Global
182    }
183}
184
185impl SystemTable for InformationSchemaSstsIndexMeta {
186    fn table_id(&self) -> TableId {
187        INFORMATION_SCHEMA_SSTS_INDEX_META_TABLE_ID
188    }
189
190    fn table_name(&self) -> &'static str {
191        SSTS_INDEX_META
192    }
193
194    fn schema(&self) -> SchemaRef {
195        self.schema.clone()
196    }
197
198    fn to_stream(&self, request: ScanRequest) -> Result<SendableRecordBatchStream> {
199        let schema = utils::projected_schema(&self.schema, &request)?;
200
201        let info_ext = utils::information_extension(&self.catalog_manager)?;
202        let req = utils::inspect_request(DatanodeInspectKind::SstIndexMeta, &self.schema, request)?;
203
204        let future = async move {
205            info_ext
206                .inspect_datanode(req)
207                .await
208                .map_err(BoxedError::new)
209                .context(common_recordbatch::error::ExternalSnafu)
210        };
211        Ok(Box::pin(AsyncRecordBatchStreamAdapter::new(
212            schema,
213            Box::pin(future),
214        )))
215    }
216
217    fn scan_plan(&self, request: ScanRequest) -> Result<Option<Arc<dyn ExecutionPlan>>> {
218        utils::inspect_plan(
219            &self.catalog_manager,
220            DatanodeInspectKind::SstIndexMeta,
221            &self.schema,
222            request,
223        )
224    }
225}
226
227#[cfg(test)]
228mod tests {
229    use crate::information_schema::DatanodeInspectRequest;
230    use cache::{build_fundamental_cache_registry, with_default_composite_cache_registry};
231    use common_catalog::consts::{DEFAULT_CATALOG_NAME, INFORMATION_SCHEMA_NAME};
232    use common_meta::cache::{CacheRegistryBuilder, LayeredCacheRegistryBuilder};
233    use common_meta::cluster::NodeInfo;
234    use common_meta::datanode::RegionStat;
235    use common_meta::key::flow::flow_state::FlowStat;
236    use common_meta::kv_backend::memory::MemoryKvBackend;
237    use common_procedure::ProcedureInfo;
238    use common_recordbatch::{RecordBatches, SendableRecordBatchStream};
239    use datafusion::physical_plan::empty::EmptyExec;
240
241    use super::*;
242    use crate::error::Error;
243    use crate::information_schema::InformationExtension;
244    use crate::kvbackend::KvBackendCatalogManagerBuilder;
245
246    struct PlanInformationExtension;
247
248    #[async_trait::async_trait]
249    impl InformationExtension for PlanInformationExtension {
250        type Error = Error;
251
252        async fn nodes(&self) -> std::result::Result<Vec<NodeInfo>, Self::Error> {
253            Ok(vec![])
254        }
255
256        async fn procedures(
257            &self,
258        ) -> std::result::Result<Vec<(String, ProcedureInfo)>, Self::Error> {
259            Ok(vec![])
260        }
261
262        async fn region_stats(&self) -> std::result::Result<Vec<RegionStat>, Self::Error> {
263            Ok(vec![])
264        }
265
266        async fn flow_stats(&self) -> std::result::Result<Option<FlowStat>, Self::Error> {
267            Ok(None)
268        }
269
270        async fn inspect_datanode(
271            &self,
272            _request: DatanodeInspectRequest,
273        ) -> std::result::Result<SendableRecordBatchStream, Self::Error> {
274            Ok(RecordBatches::empty().as_stream())
275        }
276
277        fn inspect_datanode_plan(
278            &self,
279            _request: DatanodeInspectRequest,
280            schema: SchemaRef,
281        ) -> std::result::Result<Option<Arc<dyn ExecutionPlan>>, Self::Error> {
282            Ok(Some(Arc::new(EmptyExec::new(
283                schema.arrow_schema().clone(),
284            ))))
285        }
286    }
287
288    #[test]
289    fn test_inspect_request_keeps_unsafe_filters_on_frontend() {
290        use datafusion::arrow::datatypes::{DataType, TimeUnit};
291        use datafusion::logical_expr::expr::Cast;
292        use datafusion::logical_expr::{Expr, col, lit};
293
294        for (kind, schema) in [
295            (
296                DatanodeInspectKind::CompactionTasks,
297                store_api::compaction_task::CompactionTaskEntry::schema(),
298            ),
299            (DatanodeInspectKind::SstManifest, ManifestSstEntry::schema()),
300            (DatanodeInspectKind::SstStorage, StorageSstEntry::schema()),
301            (
302                DatanodeInspectKind::SstIndexMeta,
303                PuffinIndexMetaEntry::schema(),
304            ),
305        ] {
306            let safe = col(&schema.column_schemas()[0].name).is_not_null();
307            let scan = ScanRequest {
308                projection: Some(vec![0]),
309                filters: vec![safe.clone()],
310                limit: Some(1),
311                ..Default::default()
312            };
313            assert_eq!(
314                utils::inspect_request(kind, &schema, scan.clone()).unwrap(),
315                DatanodeInspectRequest {
316                    kind,
317                    scan: scan.clone()
318                }
319            );
320            for unsafe_filter in [
321                Expr::Cast(Cast::new(
322                    Box::new(lit(0_i64)),
323                    DataType::Timestamp(TimeUnit::Nanosecond, Some("+08:00".into())),
324                ))
325                .is_not_null(),
326                Expr::Cast(Cast::new(Box::new(lit(1_i64)), DataType::Decimal32(9, 0)))
327                    .is_not_null(),
328                col("missing_column").is_null(),
329            ] {
330                let mut input = scan.clone();
331                input.filters.push(unsafe_filter);
332                let mut expected = scan.clone();
333                expected.limit = None;
334                assert_eq!(
335                    utils::inspect_request(kind, &schema, input).unwrap(),
336                    DatanodeInspectRequest {
337                        kind,
338                        scan: expected
339                    }
340                );
341            }
342        }
343    }
344
345    #[tokio::test]
346    async fn ssts_manifest_uses_information_extension_scan_plan() {
347        let backend = Arc::new(MemoryKvBackend::default());
348        let layered_cache_builder = LayeredCacheRegistryBuilder::default()
349            .add_cache_registry(CacheRegistryBuilder::default().build());
350        let fundamental_cache_registry = build_fundamental_cache_registry(backend.clone());
351        let layered_cache_registry = Arc::new(
352            with_default_composite_cache_registry(
353                layered_cache_builder.add_cache_registry(fundamental_cache_registry),
354            )
355            .unwrap()
356            .build(),
357        );
358
359        let catalog_manager = KvBackendCatalogManagerBuilder::new(
360            Arc::new(PlanInformationExtension),
361            backend,
362            layered_cache_registry,
363        )
364        .build();
365
366        let table = catalog_manager
367            .table(
368                DEFAULT_CATALOG_NAME,
369                INFORMATION_SCHEMA_NAME,
370                SSTS_MANIFEST,
371                None,
372            )
373            .await
374            .unwrap()
375            .unwrap();
376
377        let filter =
378            datafusion::logical_expr::col("region_id").eq(datafusion::logical_expr::lit(42_u64));
379        assert_eq!(
380            table.supports_filters_pushdown(&[&filter]).unwrap(),
381            vec![table::metadata::FilterPushDownType::Inexact],
382        );
383
384        let request = ScanRequest {
385            projection: Some(vec![0]),
386            ..Default::default()
387        };
388        let plan = table.scan_to_plan(request).unwrap().unwrap();
389
390        assert!(plan.as_ref().is::<EmptyExec>());
391        assert_eq!(1, plan.schema().fields().len());
392    }
393}
catalog/system_schema/information_schema/ssts.rs

catalog/system_schema/information_schema/
ssts.rs
