catalog/system_schema/information_schema/
region_statistics.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 arrow_schema::SchemaRef as ArrowSchemaRef;
18use common_catalog::consts::INFORMATION_SCHEMA_REGION_STATISTICS_TABLE_ID;
19use common_error::ext::BoxedError;
20use common_meta::datanode::RegionStat;
21use common_recordbatch::adapter::RecordBatchStreamAdapter;
22use common_recordbatch::{DfSendableRecordBatchStream, RecordBatch, SendableRecordBatchStream};
23use datafusion::execution::TaskContext;
24use datafusion::physical_plan::stream::RecordBatchStreamAdapter as DfRecordBatchStreamAdapter;
25use datafusion::physical_plan::streaming::PartitionStream as DfPartitionStream;
26use datatypes::prelude::{ConcreteDataType, ScalarVectorBuilder, VectorRef};
27use datatypes::schema::{ColumnSchema, Schema, SchemaRef};
28use datatypes::value::Value;
29use datatypes::vectors::{StringVectorBuilder, UInt32VectorBuilder, UInt64VectorBuilder};
30use snafu::ResultExt;
31use store_api::storage::{ScanRequest, TableId};
32
33use crate::CatalogManager;
34use crate::error::{CreateRecordBatchSnafu, InternalSnafu, Result};
35use crate::information_schema::Predicates;
36use crate::system_schema::information_schema::{InformationTable, REGION_STATISTICS};
37use crate::system_schema::utils;
38
39const REGION_ID: &str = "region_id";
40const TABLE_ID: &str = "table_id";
41const REGION_NUMBER: &str = "region_number";
42const REGION_ROWS: &str = "region_rows";
43const WRITTEN_BYTES: &str = "written_bytes_since_open";
44const QUERY_CPU_TIME_MILLIS: &str = "query_cpu_time_millis";
45const QUERY_SCANNED_BYTES: &str = "query_scanned_bytes";
46const DISK_SIZE: &str = "disk_size";
47const MEMTABLE_SIZE: &str = "memtable_size";
48const MANIFEST_SIZE: &str = "manifest_size";
49const SST_SIZE: &str = "sst_size";
50const SST_NUM: &str = "sst_num";
51const INDEX_SIZE: &str = "index_size";
52const ENGINE: &str = "engine";
53const REGION_ROLE: &str = "region_role";
54
55const INIT_CAPACITY: usize = 42;
56
57/// The `REGION_STATISTICS` table provides information about the region statistics. Including fields:
58///
59/// - `region_id`: The region id.
60/// - `table_id`: The table id.
61/// - `region_number`: The region number.
62/// - `region_rows`: The number of rows in region.
63/// - `written_bytes_since_open`: The total bytes written of the region since region opened.
64/// - `query_cpu_time_millis`: The total query CPU time of the region since region opened, in milliseconds.
65/// - `query_scanned_bytes`: The total bytes scanned by queries since region opened.
66/// - `memtable_size`: The memtable size in bytes.
67/// - `disk_size`: The approximate disk size in bytes.
68/// - `manifest_size`: The manifest size in bytes.
69/// - `sst_size`: The sst data files size in bytes.
70/// - `index_size`: The sst index files size in bytes.
71/// - `engine`: The engine type.
72/// - `region_role`: The region role.
73#[derive(Debug)]
74pub(super) struct InformationSchemaRegionStatistics {
75    schema: SchemaRef,
76    catalog_manager: Weak<dyn CatalogManager>,
77}
78
79impl InformationSchemaRegionStatistics {
80    pub(super) fn new(catalog_manager: Weak<dyn CatalogManager>) -> Self {
81        Self {
82            schema: Self::schema(),
83            catalog_manager,
84        }
85    }
86
87    pub(crate) fn schema() -> SchemaRef {
88        Arc::new(Schema::new(vec![
89            ColumnSchema::new(REGION_ID, ConcreteDataType::uint64_datatype(), false),
90            ColumnSchema::new(TABLE_ID, ConcreteDataType::uint32_datatype(), false),
91            ColumnSchema::new(REGION_NUMBER, ConcreteDataType::uint32_datatype(), false),
92            ColumnSchema::new(REGION_ROWS, ConcreteDataType::uint64_datatype(), true),
93            ColumnSchema::new(WRITTEN_BYTES, ConcreteDataType::uint64_datatype(), true),
94            ColumnSchema::new(
95                QUERY_CPU_TIME_MILLIS,
96                ConcreteDataType::uint64_datatype(),
97                true,
98            ),
99            ColumnSchema::new(
100                QUERY_SCANNED_BYTES,
101                ConcreteDataType::uint64_datatype(),
102                true,
103            ),
104            ColumnSchema::new(DISK_SIZE, ConcreteDataType::uint64_datatype(), true),
105            ColumnSchema::new(MEMTABLE_SIZE, ConcreteDataType::uint64_datatype(), true),
106            ColumnSchema::new(MANIFEST_SIZE, ConcreteDataType::uint64_datatype(), true),
107            ColumnSchema::new(SST_SIZE, ConcreteDataType::uint64_datatype(), true),
108            ColumnSchema::new(SST_NUM, ConcreteDataType::uint64_datatype(), true),
109            ColumnSchema::new(INDEX_SIZE, ConcreteDataType::uint64_datatype(), true),
110            ColumnSchema::new(ENGINE, ConcreteDataType::string_datatype(), true),
111            ColumnSchema::new(REGION_ROLE, ConcreteDataType::string_datatype(), true),
112        ]))
113    }
114
115    fn builder(&self) -> InformationSchemaRegionStatisticsBuilder {
116        InformationSchemaRegionStatisticsBuilder::new(
117            self.schema.clone(),
118            self.catalog_manager.clone(),
119        )
120    }
121}
122
123impl InformationTable for InformationSchemaRegionStatistics {
124    fn table_id(&self) -> TableId {
125        INFORMATION_SCHEMA_REGION_STATISTICS_TABLE_ID
126    }
127
128    fn table_name(&self) -> &'static str {
129        REGION_STATISTICS
130    }
131
132    fn schema(&self) -> SchemaRef {
133        self.schema.clone()
134    }
135
136    fn to_stream(&self, request: ScanRequest) -> Result<SendableRecordBatchStream> {
137        let schema = self.schema.arrow_schema().clone();
138        let mut builder = self.builder();
139
140        let stream = Box::pin(DfRecordBatchStreamAdapter::new(
141            schema,
142            futures::stream::once(async move {
143                builder
144                    .make_region_statistics(Some(request))
145                    .await
146                    .map(|x| x.into_df_record_batch())
147                    .map_err(Into::into)
148            }),
149        ));
150
151        Ok(Box::pin(
152            RecordBatchStreamAdapter::try_new(stream)
153                .map_err(BoxedError::new)
154                .context(InternalSnafu)?,
155        ))
156    }
157}
158
159struct InformationSchemaRegionStatisticsBuilder {
160    schema: SchemaRef,
161    catalog_manager: Weak<dyn CatalogManager>,
162
163    region_ids: UInt64VectorBuilder,
164    table_ids: UInt32VectorBuilder,
165    region_numbers: UInt32VectorBuilder,
166    region_rows: UInt64VectorBuilder,
167    written_bytes: UInt64VectorBuilder,
168    query_cpu_time_millis: UInt64VectorBuilder,
169    query_scanned_bytes: UInt64VectorBuilder,
170    disk_sizes: UInt64VectorBuilder,
171    memtable_sizes: UInt64VectorBuilder,
172    manifest_sizes: UInt64VectorBuilder,
173    sst_sizes: UInt64VectorBuilder,
174    sst_nums: UInt64VectorBuilder,
175    index_sizes: UInt64VectorBuilder,
176    engines: StringVectorBuilder,
177    region_roles: StringVectorBuilder,
178}
179
180impl InformationSchemaRegionStatisticsBuilder {
181    fn new(schema: SchemaRef, catalog_manager: Weak<dyn CatalogManager>) -> Self {
182        Self {
183            schema,
184            catalog_manager,
185            region_ids: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
186            table_ids: UInt32VectorBuilder::with_capacity(INIT_CAPACITY),
187            region_numbers: UInt32VectorBuilder::with_capacity(INIT_CAPACITY),
188            region_rows: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
189            written_bytes: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
190            query_cpu_time_millis: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
191            query_scanned_bytes: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
192            disk_sizes: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
193            memtable_sizes: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
194            manifest_sizes: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
195            sst_sizes: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
196            sst_nums: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
197            index_sizes: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
198            engines: StringVectorBuilder::with_capacity(INIT_CAPACITY),
199            region_roles: StringVectorBuilder::with_capacity(INIT_CAPACITY),
200        }
201    }
202
203    /// Construct a new `InformationSchemaRegionStatistics` from the collected data.
204    async fn make_region_statistics(
205        &mut self,
206        request: Option<ScanRequest>,
207    ) -> Result<RecordBatch> {
208        let predicates = Predicates::from_scan_request(&request);
209        let information_extension = utils::information_extension(&self.catalog_manager)?;
210        let region_stats = information_extension.region_stats().await?;
211        for region_stat in region_stats {
212            self.add_region_statistic(&predicates, region_stat);
213        }
214        self.finish()
215    }
216
217    fn add_region_statistic(&mut self, predicate: &Predicates, region_stat: RegionStat) {
218        let row = [
219            (REGION_ID, &Value::from(region_stat.id.as_u64())),
220            (TABLE_ID, &Value::from(region_stat.id.table_id())),
221            (REGION_NUMBER, &Value::from(region_stat.id.region_number())),
222            (REGION_ROWS, &Value::from(region_stat.num_rows)),
223            (WRITTEN_BYTES, &Value::from(region_stat.written_bytes)),
224            (
225                QUERY_CPU_TIME_MILLIS,
226                &Value::from(region_stat.query_cpu_time / 1_000_000),
227            ),
228            (
229                QUERY_SCANNED_BYTES,
230                &Value::from(region_stat.query_scanned_bytes),
231            ),
232            (DISK_SIZE, &Value::from(region_stat.approximate_bytes)),
233            (MEMTABLE_SIZE, &Value::from(region_stat.memtable_size)),
234            (MANIFEST_SIZE, &Value::from(region_stat.manifest_size)),
235            (SST_SIZE, &Value::from(region_stat.sst_size)),
236            (SST_NUM, &Value::from(region_stat.sst_num)),
237            (INDEX_SIZE, &Value::from(region_stat.index_size)),
238            (ENGINE, &Value::from(region_stat.engine.as_str())),
239            (REGION_ROLE, &Value::from(region_stat.role.to_string())),
240        ];
241
242        if !predicate.eval(&row) {
243            return;
244        }
245
246        self.region_ids.push(Some(region_stat.id.as_u64()));
247        self.table_ids.push(Some(region_stat.id.table_id()));
248        self.region_numbers
249            .push(Some(region_stat.id.region_number()));
250        self.region_rows.push(Some(region_stat.num_rows));
251        self.written_bytes.push(Some(region_stat.written_bytes));
252        self.query_cpu_time_millis
253            .push(Some(region_stat.query_cpu_time / 1_000_000));
254        self.query_scanned_bytes
255            .push(Some(region_stat.query_scanned_bytes));
256        self.disk_sizes.push(Some(region_stat.approximate_bytes));
257        self.memtable_sizes.push(Some(region_stat.memtable_size));
258        self.manifest_sizes.push(Some(region_stat.manifest_size));
259        self.sst_sizes.push(Some(region_stat.sst_size));
260        self.sst_nums.push(Some(region_stat.sst_num));
261        self.index_sizes.push(Some(region_stat.index_size));
262        self.engines.push(Some(&region_stat.engine));
263        self.region_roles.push(Some(&region_stat.role.to_string()));
264    }
265
266    fn finish(&mut self) -> Result<RecordBatch> {
267        let columns: Vec<VectorRef> = vec![
268            Arc::new(self.region_ids.finish()),
269            Arc::new(self.table_ids.finish()),
270            Arc::new(self.region_numbers.finish()),
271            Arc::new(self.region_rows.finish()),
272            Arc::new(self.written_bytes.finish()),
273            Arc::new(self.query_cpu_time_millis.finish()),
274            Arc::new(self.query_scanned_bytes.finish()),
275            Arc::new(self.disk_sizes.finish()),
276            Arc::new(self.memtable_sizes.finish()),
277            Arc::new(self.manifest_sizes.finish()),
278            Arc::new(self.sst_sizes.finish()),
279            Arc::new(self.sst_nums.finish()),
280            Arc::new(self.index_sizes.finish()),
281            Arc::new(self.engines.finish()),
282            Arc::new(self.region_roles.finish()),
283        ];
284
285        RecordBatch::new(self.schema.clone(), columns).context(CreateRecordBatchSnafu)
286    }
287}
288
289impl DfPartitionStream for InformationSchemaRegionStatistics {
290    fn schema(&self) -> &ArrowSchemaRef {
291        self.schema.arrow_schema()
292    }
293
294    fn execute(&self, _: Arc<TaskContext>) -> DfSendableRecordBatchStream {
295        let schema = self.schema.arrow_schema().clone();
296        let mut builder = self.builder();
297        Box::pin(DfRecordBatchStreamAdapter::new(
298            schema,
299            futures::stream::once(async move {
300                builder
301                    .make_region_statistics(None)
302                    .await
303                    .map(|x| x.into_df_record_batch())
304                    .map_err(Into::into)
305            }),
306        ))
307    }
308}
catalog/system_schema/information_schema/region_statistics.rs

catalog/system_schema/information_schema/
region_statistics.rs
