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::{ProjectSchemaSnafu, Result};
32use crate::information_schema::{
33    DatanodeInspectKind, DatanodeInspectRequest, InformationTable, SSTS_INDEX_META, SSTS_MANIFEST,
34    SSTS_STORAGE,
35};
36use crate::system_schema::utils;
37
38fn projected_schema(schema: &SchemaRef, request: &ScanRequest) -> Result<SchemaRef> {
39    if let Some(p) = request.projection_indices() {
40        Ok(Arc::new(schema.try_project(p).context(ProjectSchemaSnafu)?))
41    } else {
42        Ok(schema.clone())
43    }
44}
45
46fn inspect_plan(
47    catalog_manager: &Weak<dyn CatalogManager>,
48    kind: DatanodeInspectKind,
49    schema: &SchemaRef,
50    request: ScanRequest,
51) -> Result<Option<Arc<dyn ExecutionPlan>>> {
52    let projected_schema = projected_schema(schema, &request)?;
53    let info_ext = utils::information_extension(catalog_manager)?;
54    let req = DatanodeInspectRequest {
55        kind,
56        scan: request,
57    };
58
59    info_ext.inspect_datanode_plan(req, projected_schema)
60}
61
62/// Information schema table for sst manifest.
63pub struct InformationSchemaSstsManifest {
64    schema: SchemaRef,
65    catalog_manager: Weak<dyn CatalogManager>,
66}
67
68impl InformationSchemaSstsManifest {
69    pub(super) fn new(catalog_manager: Weak<dyn CatalogManager>) -> Self {
70        Self {
71            schema: ManifestSstEntry::schema(),
72            catalog_manager,
73        }
74    }
75}
76
77impl InformationTable for InformationSchemaSstsManifest {
78    fn table_id(&self) -> TableId {
79        INFORMATION_SCHEMA_SSTS_MANIFEST_TABLE_ID
80    }
81
82    fn table_name(&self) -> &'static str {
83        SSTS_MANIFEST
84    }
85
86    fn schema(&self) -> SchemaRef {
87        self.schema.clone()
88    }
89
90    fn to_stream(&self, request: ScanRequest) -> Result<SendableRecordBatchStream> {
91        let schema = projected_schema(&self.schema, &request)?;
92        let info_ext = utils::information_extension(&self.catalog_manager)?;
93        let req = DatanodeInspectRequest {
94            kind: DatanodeInspectKind::SstManifest,
95            scan: request,
96        };
97
98        let future = async move {
99            info_ext
100                .inspect_datanode(req)
101                .await
102                .map_err(BoxedError::new)
103                .context(common_recordbatch::error::ExternalSnafu)
104        };
105        Ok(Box::pin(AsyncRecordBatchStreamAdapter::new(
106            schema,
107            Box::pin(future),
108        )))
109    }
110
111    fn scan_plan(&self, request: ScanRequest) -> Result<Option<Arc<dyn ExecutionPlan>>> {
112        inspect_plan(
113            &self.catalog_manager,
114            DatanodeInspectKind::SstManifest,
115            &self.schema,
116            request,
117        )
118    }
119}
120
121/// Information schema table for sst storage.
122pub struct InformationSchemaSstsStorage {
123    schema: SchemaRef,
124    catalog_manager: Weak<dyn CatalogManager>,
125}
126
127impl InformationSchemaSstsStorage {
128    pub(super) fn new(catalog_manager: Weak<dyn CatalogManager>) -> Self {
129        Self {
130            schema: StorageSstEntry::schema(),
131            catalog_manager,
132        }
133    }
134}
135
136impl InformationTable for InformationSchemaSstsStorage {
137    fn table_id(&self) -> TableId {
138        INFORMATION_SCHEMA_SSTS_STORAGE_TABLE_ID
139    }
140
141    fn table_name(&self) -> &'static str {
142        SSTS_STORAGE
143    }
144
145    fn schema(&self) -> SchemaRef {
146        self.schema.clone()
147    }
148
149    fn to_stream(&self, request: ScanRequest) -> Result<SendableRecordBatchStream> {
150        let schema = projected_schema(&self.schema, &request)?;
151
152        let info_ext = utils::information_extension(&self.catalog_manager)?;
153        let req = DatanodeInspectRequest {
154            kind: DatanodeInspectKind::SstStorage,
155            scan: request,
156        };
157
158        let future = async move {
159            info_ext
160                .inspect_datanode(req)
161                .await
162                .map_err(BoxedError::new)
163                .context(common_recordbatch::error::ExternalSnafu)
164        };
165        Ok(Box::pin(AsyncRecordBatchStreamAdapter::new(
166            schema,
167            Box::pin(future),
168        )))
169    }
170
171    fn scan_plan(&self, request: ScanRequest) -> Result<Option<Arc<dyn ExecutionPlan>>> {
172        inspect_plan(
173            &self.catalog_manager,
174            DatanodeInspectKind::SstStorage,
175            &self.schema,
176            request,
177        )
178    }
179}
180
181/// Information schema table for index metadata.
182pub struct InformationSchemaSstsIndexMeta {
183    schema: SchemaRef,
184    catalog_manager: Weak<dyn CatalogManager>,
185}
186
187impl InformationSchemaSstsIndexMeta {
188    pub(super) fn new(catalog_manager: Weak<dyn CatalogManager>) -> Self {
189        Self {
190            schema: PuffinIndexMetaEntry::schema(),
191            catalog_manager,
192        }
193    }
194}
195
196impl InformationTable for InformationSchemaSstsIndexMeta {
197    fn table_id(&self) -> TableId {
198        INFORMATION_SCHEMA_SSTS_INDEX_META_TABLE_ID
199    }
200
201    fn table_name(&self) -> &'static str {
202        SSTS_INDEX_META
203    }
204
205    fn schema(&self) -> SchemaRef {
206        self.schema.clone()
207    }
208
209    fn to_stream(&self, request: ScanRequest) -> Result<SendableRecordBatchStream> {
210        let schema = projected_schema(&self.schema, &request)?;
211
212        let info_ext = utils::information_extension(&self.catalog_manager)?;
213        let req = DatanodeInspectRequest {
214            kind: DatanodeInspectKind::SstIndexMeta,
215            scan: request,
216        };
217
218        let future = async move {
219            info_ext
220                .inspect_datanode(req)
221                .await
222                .map_err(BoxedError::new)
223                .context(common_recordbatch::error::ExternalSnafu)
224        };
225        Ok(Box::pin(AsyncRecordBatchStreamAdapter::new(
226            schema,
227            Box::pin(future),
228        )))
229    }
230
231    fn scan_plan(&self, request: ScanRequest) -> Result<Option<Arc<dyn ExecutionPlan>>> {
232        inspect_plan(
233            &self.catalog_manager,
234            DatanodeInspectKind::SstIndexMeta,
235            &self.schema,
236            request,
237        )
238    }
239}
240
241#[cfg(test)]
242mod tests {
243    use cache::{build_fundamental_cache_registry, with_default_composite_cache_registry};
244    use common_catalog::consts::{DEFAULT_CATALOG_NAME, INFORMATION_SCHEMA_NAME};
245    use common_meta::cache::{CacheRegistryBuilder, LayeredCacheRegistryBuilder};
246    use common_meta::cluster::NodeInfo;
247    use common_meta::datanode::RegionStat;
248    use common_meta::key::flow::flow_state::FlowStat;
249    use common_meta::kv_backend::memory::MemoryKvBackend;
250    use common_procedure::ProcedureInfo;
251    use common_recordbatch::{RecordBatches, SendableRecordBatchStream};
252    use datafusion::physical_plan::empty::EmptyExec;
253
254    use super::*;
255    use crate::error::Error;
256    use crate::information_schema::InformationExtension;
257    use crate::kvbackend::KvBackendCatalogManagerBuilder;
258
259    struct PlanInformationExtension;
260
261    #[async_trait::async_trait]
262    impl InformationExtension for PlanInformationExtension {
263        type Error = Error;
264
265        async fn nodes(&self) -> std::result::Result<Vec<NodeInfo>, Self::Error> {
266            Ok(vec![])
267        }
268
269        async fn procedures(
270            &self,
271        ) -> std::result::Result<Vec<(String, ProcedureInfo)>, Self::Error> {
272            Ok(vec![])
273        }
274
275        async fn region_stats(&self) -> std::result::Result<Vec<RegionStat>, Self::Error> {
276            Ok(vec![])
277        }
278
279        async fn flow_stats(&self) -> std::result::Result<Option<FlowStat>, Self::Error> {
280            Ok(None)
281        }
282
283        async fn inspect_datanode(
284            &self,
285            _request: DatanodeInspectRequest,
286        ) -> std::result::Result<SendableRecordBatchStream, Self::Error> {
287            Ok(RecordBatches::empty().as_stream())
288        }
289
290        fn inspect_datanode_plan(
291            &self,
292            _request: DatanodeInspectRequest,
293            schema: SchemaRef,
294        ) -> std::result::Result<Option<Arc<dyn ExecutionPlan>>, Self::Error> {
295            Ok(Some(Arc::new(EmptyExec::new(
296                schema.arrow_schema().clone(),
297            ))))
298        }
299    }
300
301    #[tokio::test]
302    async fn ssts_manifest_uses_information_extension_scan_plan() {
303        let backend = Arc::new(MemoryKvBackend::default());
304        let layered_cache_builder = LayeredCacheRegistryBuilder::default()
305            .add_cache_registry(CacheRegistryBuilder::default().build());
306        let fundamental_cache_registry = build_fundamental_cache_registry(backend.clone());
307        let layered_cache_registry = Arc::new(
308            with_default_composite_cache_registry(
309                layered_cache_builder.add_cache_registry(fundamental_cache_registry),
310            )
311            .unwrap()
312            .build(),
313        );
314
315        let catalog_manager = KvBackendCatalogManagerBuilder::new(
316            Arc::new(PlanInformationExtension),
317            backend,
318            layered_cache_registry,
319        )
320        .build();
321
322        let table = catalog_manager
323            .table(
324                DEFAULT_CATALOG_NAME,
325                INFORMATION_SCHEMA_NAME,
326                SSTS_MANIFEST,
327                None,
328            )
329            .await
330            .unwrap()
331            .unwrap();
332
333        let request = ScanRequest {
334            projection_input: Some(vec![0].into()),
335            ..Default::default()
336        };
337        let plan = table.scan_to_plan(request).unwrap().unwrap();
338
339        assert!(plan.as_any().is::<EmptyExec>());
340        assert_eq!(1, plan.schema().fields().len());
341    }
342}
catalog/system_schema/information_schema/ssts.rs

catalog/system_schema/information_schema/
ssts.rs
