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

catalog/system_schema/information_schema/
region_info.rs
