catalog/system_schema/information_schema/
region_info.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::INFORMATION_SCHEMA_REGION_INFO_TABLE_ID;
18use common_error::ext::BoxedError;
19use common_recordbatch::SendableRecordBatchStream;
20use common_recordbatch::adapter::AsyncRecordBatchStreamAdapter;
21use datafusion::physical_plan::ExecutionPlan;
22use datatypes::schema::SchemaRef;
23use snafu::ResultExt;
24use store_api::region_info::RegionInfoEntry;
25use store_api::storage::{ScanRequest, TableId};
26
27use crate::CatalogManager;
28use crate::error::{ProjectSchemaSnafu, Result};
29use crate::information_schema::{
30    DatanodeInspectKind, DatanodeInspectRequest, InformationTable, REGION_INFO,
31};
32use crate::system_schema::utils;
33
34/// Information schema table for region info.
35pub struct InformationSchemaRegionInfo {
36    schema: SchemaRef,
37    catalog_manager: Weak<dyn CatalogManager>,
38}
39
40impl InformationSchemaRegionInfo {
41    pub(super) fn new(catalog_manager: Weak<dyn CatalogManager>) -> Self {
42        Self {
43            schema: RegionInfoEntry::schema(),
44            catalog_manager,
45        }
46    }
47}
48
49impl InformationTable for InformationSchemaRegionInfo {
50    fn table_id(&self) -> TableId {
51        INFORMATION_SCHEMA_REGION_INFO_TABLE_ID
52    }
53
54    fn table_name(&self) -> &'static str {
55        REGION_INFO
56    }
57
58    fn schema(&self) -> SchemaRef {
59        self.schema.clone()
60    }
61
62    fn to_stream(&self, request: ScanRequest) -> Result<SendableRecordBatchStream> {
63        let schema = if let Some(p) = request.projection_indices() {
64            Arc::new(self.schema.try_project(p).context(ProjectSchemaSnafu)?)
65        } else {
66            self.schema.clone()
67        };
68
69        let info_ext = utils::information_extension(&self.catalog_manager)?;
70        let req = DatanodeInspectRequest {
71            kind: DatanodeInspectKind::RegionInfo,
72            scan: request,
73        };
74
75        let future = async move {
76            info_ext
77                .inspect_datanode(req)
78                .await
79                .map_err(BoxedError::new)
80                .context(common_recordbatch::error::ExternalSnafu)
81        };
82        Ok(Box::pin(AsyncRecordBatchStreamAdapter::new(
83            schema,
84            Box::pin(future),
85        )))
86    }
87
88    fn scan_plan(&self, request: ScanRequest) -> Result<Option<Arc<dyn ExecutionPlan>>> {
89        let schema = if let Some(p) = request.projection_indices() {
90            Arc::new(self.schema.try_project(p).context(ProjectSchemaSnafu)?)
91        } else {
92            self.schema.clone()
93        };
94
95        let info_ext = utils::information_extension(&self.catalog_manager)?;
96        let req = DatanodeInspectRequest {
97            kind: DatanodeInspectKind::RegionInfo,
98            scan: request,
99        };
100
101        info_ext.inspect_datanode_plan(req, schema)
102    }
103}
catalog/system_schema/information_schema/region_info.rs

catalog/system_schema/information_schema/
region_info.rs
