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 common_time::timestamp::TimeUnit;
24use datafusion::execution::TaskContext;
25use datafusion::physical_plan::stream::RecordBatchStreamAdapter as DfRecordBatchStreamAdapter;
26use datafusion::physical_plan::streaming::PartitionStream as DfPartitionStream;
27use datatypes::prelude::{ConcreteDataType, ScalarVectorBuilder, VectorRef};
28use datatypes::schema::{ColumnSchema, Schema, SchemaRef};
29use datatypes::timestamp::TimestampMillisecond;
30use datatypes::value::Value;
31use datatypes::vectors::{
32    StringVectorBuilder, TimestampMillisecondVectorBuilder, UInt32VectorBuilder,
33    UInt64VectorBuilder,
34};
35use snafu::ResultExt;
36use store_api::storage::{ScanRequest, TableId};
37
38use crate::CatalogManager;
39use crate::error::{CreateRecordBatchSnafu, InternalSnafu, Result};
40use crate::information_schema::Predicates;
41use crate::system_schema::SystemTable;
42use crate::system_schema::information_schema::{
43    InformationSchemaTableScope, InformationTable, REGION_STATISTICS,
44};
45use crate::system_schema::utils;
46
47const REGION_ID: &str = "region_id";
48const TABLE_ID: &str = "table_id";
49const REGION_NUMBER: &str = "region_number";
50const REGION_ROWS: &str = "region_rows";
51const WRITTEN_BYTES: &str = "written_bytes_since_open";
52const QUERY_CPU_TIME_MILLIS: &str = "query_cpu_time_millis";
53const QUERY_SCANNED_BYTES: &str = "query_scanned_bytes";
54const DISK_SIZE: &str = "disk_size";
55const MEMTABLE_SIZE: &str = "memtable_size";
56const MANIFEST_SIZE: &str = "manifest_size";
57const SST_SIZE: &str = "sst_size";
58const SST_NUM: &str = "sst_num";
59const INDEX_SIZE: &str = "index_size";
60const ENGINE: &str = "engine";
61const REGION_ROLE: &str = "region_role";
62const MIN_TIMESTAMP: &str = "min_timestamp";
63const MAX_TIMESTAMP: &str = "max_timestamp";
64
65const INIT_CAPACITY: usize = 42;
66
67/// The `REGION_STATISTICS` table provides information about the region statistics. Including fields:
68///
69/// - `region_id`: The region id.
70/// - `table_id`: The table id.
71/// - `region_number`: The region number.
72/// - `region_rows`: The number of rows in region.
73/// - `written_bytes_since_open`: The total bytes written of the region since region opened.
74/// - `query_cpu_time_millis`: The total query CPU time of the region since region opened, in milliseconds.
75/// - `query_scanned_bytes`: The total bytes scanned by queries since region opened.
76/// - `memtable_size`: The memtable size in bytes.
77/// - `disk_size`: The approximate disk size in bytes.
78/// - `manifest_size`: Stored bytes of the latest observed checkpoint and subsequent
79///   deltas (compressed when enabled), excluding staging and `_last_checkpoint`.
80///   Older files retained on disk, including failed deletions, are not counted.
81///   Followers update this estimate on sync; metadata read failures can leave it stale.
82///   Promotion does not refresh it, so an incomplete estimate can persist without
83///   further syncs or checkpoints.
84/// - `sst_size`: The sst data files size in bytes.
85/// - `index_size`: The sst index files size in bytes.
86/// - `engine`: The engine type.
87/// - `region_role`: The region role.
88#[derive(Debug)]
89pub(super) struct InformationSchemaRegionStatistics {
90    schema: SchemaRef,
91    catalog_manager: Weak<dyn CatalogManager>,
92}
93
94impl InformationSchemaRegionStatistics {
95    pub(super) fn new(catalog_manager: Weak<dyn CatalogManager>) -> Self {
96        Self {
97            schema: Self::schema(),
98            catalog_manager,
99        }
100    }
101
102    pub(crate) fn schema() -> SchemaRef {
103        Arc::new(Schema::new(vec![
104            ColumnSchema::new(REGION_ID, ConcreteDataType::uint64_datatype(), false),
105            ColumnSchema::new(TABLE_ID, ConcreteDataType::uint32_datatype(), false),
106            ColumnSchema::new(REGION_NUMBER, ConcreteDataType::uint32_datatype(), false),
107            ColumnSchema::new(REGION_ROWS, ConcreteDataType::uint64_datatype(), true),
108            ColumnSchema::new(WRITTEN_BYTES, ConcreteDataType::uint64_datatype(), true),
109            ColumnSchema::new(
110                QUERY_CPU_TIME_MILLIS,
111                ConcreteDataType::uint64_datatype(),
112                true,
113            ),
114            ColumnSchema::new(
115                QUERY_SCANNED_BYTES,
116                ConcreteDataType::uint64_datatype(),
117                true,
118            ),
119            ColumnSchema::new(DISK_SIZE, ConcreteDataType::uint64_datatype(), true),
120            ColumnSchema::new(MEMTABLE_SIZE, ConcreteDataType::uint64_datatype(), true),
121            ColumnSchema::new(MANIFEST_SIZE, ConcreteDataType::uint64_datatype(), true),
122            ColumnSchema::new(SST_SIZE, ConcreteDataType::uint64_datatype(), true),
123            ColumnSchema::new(SST_NUM, ConcreteDataType::uint64_datatype(), true),
124            ColumnSchema::new(INDEX_SIZE, ConcreteDataType::uint64_datatype(), true),
125            ColumnSchema::new(ENGINE, ConcreteDataType::string_datatype(), true),
126            ColumnSchema::new(REGION_ROLE, ConcreteDataType::string_datatype(), true),
127            ColumnSchema::new(
128                MIN_TIMESTAMP,
129                ConcreteDataType::timestamp_millisecond_datatype(),
130                true,
131            ),
132            ColumnSchema::new(
133                MAX_TIMESTAMP,
134                ConcreteDataType::timestamp_millisecond_datatype(),
135                true,
136            ),
137        ]))
138    }
139
140    fn builder(&self) -> InformationSchemaRegionStatisticsBuilder {
141        InformationSchemaRegionStatisticsBuilder::new(
142            self.schema.clone(),
143            self.catalog_manager.clone(),
144        )
145    }
146}
147
148impl InformationTable for InformationSchemaRegionStatistics {
149    fn scope(&self) -> InformationSchemaTableScope {
150        InformationSchemaTableScope::Global
151    }
152}
153
154impl SystemTable for InformationSchemaRegionStatistics {
155    fn table_id(&self) -> TableId {
156        INFORMATION_SCHEMA_REGION_STATISTICS_TABLE_ID
157    }
158
159    fn table_name(&self) -> &'static str {
160        REGION_STATISTICS
161    }
162
163    fn schema(&self) -> SchemaRef {
164        self.schema.clone()
165    }
166
167    fn to_stream(&self, request: ScanRequest) -> Result<SendableRecordBatchStream> {
168        let schema = self.schema.arrow_schema().clone();
169        let mut builder = self.builder();
170
171        let stream = Box::pin(DfRecordBatchStreamAdapter::new(
172            schema,
173            futures::stream::once(async move {
174                builder
175                    .make_region_statistics(Some(request))
176                    .await
177                    .map(|x| x.into_df_record_batch())
178                    .map_err(Into::into)
179            }),
180        ));
181
182        Ok(Box::pin(
183            RecordBatchStreamAdapter::try_new(stream)
184                .map_err(BoxedError::new)
185                .context(InternalSnafu)?,
186        ))
187    }
188}
189
190struct InformationSchemaRegionStatisticsBuilder {
191    schema: SchemaRef,
192    catalog_manager: Weak<dyn CatalogManager>,
193
194    region_ids: UInt64VectorBuilder,
195    table_ids: UInt32VectorBuilder,
196    region_numbers: UInt32VectorBuilder,
197    region_rows: UInt64VectorBuilder,
198    written_bytes: UInt64VectorBuilder,
199    query_cpu_time_millis: UInt64VectorBuilder,
200    query_scanned_bytes: UInt64VectorBuilder,
201    disk_sizes: UInt64VectorBuilder,
202    memtable_sizes: UInt64VectorBuilder,
203    manifest_sizes: UInt64VectorBuilder,
204    sst_sizes: UInt64VectorBuilder,
205    sst_nums: UInt64VectorBuilder,
206    index_sizes: UInt64VectorBuilder,
207    engines: StringVectorBuilder,
208    region_roles: StringVectorBuilder,
209    min_timestamps: TimestampMillisecondVectorBuilder,
210    max_timestamps: TimestampMillisecondVectorBuilder,
211}
212
213impl InformationSchemaRegionStatisticsBuilder {
214    fn new(schema: SchemaRef, catalog_manager: Weak<dyn CatalogManager>) -> Self {
215        Self {
216            schema,
217            catalog_manager,
218            region_ids: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
219            table_ids: UInt32VectorBuilder::with_capacity(INIT_CAPACITY),
220            region_numbers: UInt32VectorBuilder::with_capacity(INIT_CAPACITY),
221            region_rows: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
222            written_bytes: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
223            query_cpu_time_millis: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
224            query_scanned_bytes: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
225            disk_sizes: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
226            memtable_sizes: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
227            manifest_sizes: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
228            sst_sizes: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
229            sst_nums: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
230            index_sizes: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
231            engines: StringVectorBuilder::with_capacity(INIT_CAPACITY),
232            region_roles: StringVectorBuilder::with_capacity(INIT_CAPACITY),
233            min_timestamps: TimestampMillisecondVectorBuilder::with_capacity(INIT_CAPACITY),
234            max_timestamps: TimestampMillisecondVectorBuilder::with_capacity(INIT_CAPACITY),
235        }
236    }
237
238    /// Construct a new `InformationSchemaRegionStatistics` from the collected data.
239    async fn make_region_statistics(
240        &mut self,
241        request: Option<ScanRequest>,
242    ) -> Result<RecordBatch> {
243        let predicates = Predicates::from_scan_request(&request);
244        let information_extension = utils::information_extension(&self.catalog_manager)?;
245        let region_stats = information_extension.region_stats().await?;
246        for region_stat in region_stats {
247            self.add_region_statistic(&predicates, region_stat);
248        }
249        self.finish()
250    }
251
252    fn add_region_statistic(&mut self, predicate: &Predicates, region_stat: RegionStat) {
253        let row = [
254            (REGION_ID, &Value::from(region_stat.id.as_u64())),
255            (TABLE_ID, &Value::from(region_stat.id.table_id())),
256            (REGION_NUMBER, &Value::from(region_stat.id.region_number())),
257            (REGION_ROWS, &Value::from(region_stat.num_rows)),
258            (WRITTEN_BYTES, &Value::from(region_stat.written_bytes)),
259            (
260                QUERY_CPU_TIME_MILLIS,
261                &Value::from(region_stat.query_cpu_time / 1_000_000),
262            ),
263            (
264                QUERY_SCANNED_BYTES,
265                &Value::from(region_stat.query_scanned_bytes),
266            ),
267            (DISK_SIZE, &Value::from(region_stat.approximate_bytes)),
268            (MEMTABLE_SIZE, &Value::from(region_stat.memtable_size)),
269            (MANIFEST_SIZE, &Value::from(region_stat.manifest_size)),
270            (SST_SIZE, &Value::from(region_stat.sst_size)),
271            (SST_NUM, &Value::from(region_stat.sst_num)),
272            (INDEX_SIZE, &Value::from(region_stat.index_size)),
273            (ENGINE, &Value::from(region_stat.engine.as_str())),
274            (REGION_ROLE, &Value::from(region_stat.role.to_string())),
275        ];
276
277        if !predicate.eval(&row) {
278            return;
279        }
280
281        self.region_ids.push(Some(region_stat.id.as_u64()));
282        self.table_ids.push(Some(region_stat.id.table_id()));
283        self.region_numbers
284            .push(Some(region_stat.id.region_number()));
285        self.region_rows.push(Some(region_stat.num_rows));
286        self.written_bytes.push(Some(region_stat.written_bytes));
287        self.query_cpu_time_millis
288            .push(Some(region_stat.query_cpu_time / 1_000_000));
289        self.query_scanned_bytes
290            .push(Some(region_stat.query_scanned_bytes));
291        self.disk_sizes.push(Some(region_stat.approximate_bytes));
292        self.memtable_sizes.push(Some(region_stat.memtable_size));
293        self.manifest_sizes.push(Some(region_stat.manifest_size));
294        self.sst_sizes.push(Some(region_stat.sst_size));
295        self.sst_nums.push(Some(region_stat.sst_num));
296        self.index_sizes.push(Some(region_stat.index_size));
297        self.engines.push(Some(&region_stat.engine));
298        self.region_roles.push(Some(&region_stat.role.to_string()));
299        // Floor the min and ceil the max so the window stays a superset: a narrower
300        // one would hide regions from a time-bounded lookup.
301        self.min_timestamps.push(
302            region_stat
303                .min_timestamp
304                .and_then(|ts| ts.convert_to(TimeUnit::Millisecond))
305                .map(TimestampMillisecond),
306        );
307        self.max_timestamps.push(
308            region_stat
309                .max_timestamp
310                .and_then(|ts| ts.convert_to_ceil(TimeUnit::Millisecond))
311                .map(TimestampMillisecond),
312        );
313    }
314
315    fn finish(&mut self) -> Result<RecordBatch> {
316        let columns: Vec<VectorRef> = vec![
317            Arc::new(self.region_ids.finish()),
318            Arc::new(self.table_ids.finish()),
319            Arc::new(self.region_numbers.finish()),
320            Arc::new(self.region_rows.finish()),
321            Arc::new(self.written_bytes.finish()),
322            Arc::new(self.query_cpu_time_millis.finish()),
323            Arc::new(self.query_scanned_bytes.finish()),
324            Arc::new(self.disk_sizes.finish()),
325            Arc::new(self.memtable_sizes.finish()),
326            Arc::new(self.manifest_sizes.finish()),
327            Arc::new(self.sst_sizes.finish()),
328            Arc::new(self.sst_nums.finish()),
329            Arc::new(self.index_sizes.finish()),
330            Arc::new(self.engines.finish()),
331            Arc::new(self.region_roles.finish()),
332            Arc::new(self.min_timestamps.finish()),
333            Arc::new(self.max_timestamps.finish()),
334        ];
335
336        RecordBatch::new(self.schema.clone(), columns).context(CreateRecordBatchSnafu)
337    }
338}
339
340impl DfPartitionStream for InformationSchemaRegionStatistics {
341    fn schema(&self) -> &ArrowSchemaRef {
342        self.schema.arrow_schema()
343    }
344
345    fn execute(&self, _: Arc<TaskContext>) -> DfSendableRecordBatchStream {
346        let schema = self.schema.arrow_schema().clone();
347        let mut builder = self.builder();
348        Box::pin(DfRecordBatchStreamAdapter::new(
349            schema,
350            futures::stream::once(async move {
351                builder
352                    .make_region_statistics(None)
353                    .await
354                    .map(|x| x.into_df_record_batch())
355                    .map_err(Into::into)
356            }),
357        ))
358    }
359}
catalog/system_schema/information_schema/region_statistics.rs

catalog/system_schema/information_schema/
region_statistics.rs
