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

catalog/system_schema/information_schema/
ssts.rs
