catalog/system_schema/information_schema/
compaction_tasks.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_COMPACTION_TASKS_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::compaction_task::CompactionTaskEntry;
25use store_api::storage::{ScanRequest, TableId};
26
27use crate::CatalogManager;
28use crate::error::Result;
29use crate::information_schema::{
30    COMPACTION_TASKS, DatanodeInspectKind, InformationSchemaTableScope, InformationTable,
31};
32use crate::system_schema::SystemTable;
33use crate::system_schema::utils;
34
35/// Information schema table for live compaction tasks.
36pub struct InformationSchemaCompactionTasks {
37    schema: SchemaRef,
38    catalog_manager: Weak<dyn CatalogManager>,
39}
40
41impl InformationSchemaCompactionTasks {
42    pub(crate) fn new(catalog_manager: Weak<dyn CatalogManager>) -> Self {
43        Self {
44            schema: CompactionTaskEntry::schema(),
45            catalog_manager,
46        }
47    }
48}
49
50impl InformationTable for InformationSchemaCompactionTasks {
51    fn scope(&self) -> InformationSchemaTableScope {
52        InformationSchemaTableScope::Global
53    }
54}
55
56impl SystemTable for InformationSchemaCompactionTasks {
57    fn table_id(&self) -> TableId {
58        INFORMATION_SCHEMA_COMPACTION_TASKS_TABLE_ID
59    }
60
61    fn table_name(&self) -> &'static str {
62        COMPACTION_TASKS
63    }
64
65    fn schema(&self) -> SchemaRef {
66        self.schema.clone()
67    }
68
69    fn to_stream(&self, request: ScanRequest) -> Result<SendableRecordBatchStream> {
70        let schema = utils::projected_schema(&self.schema, &request)?;
71
72        let info_ext = utils::information_extension(&self.catalog_manager)?;
73        let req =
74            utils::inspect_request(DatanodeInspectKind::CompactionTasks, &self.schema, request)?;
75
76        let future = async move {
77            info_ext
78                .inspect_datanode(req)
79                .await
80                .map_err(BoxedError::new)
81                .context(common_recordbatch::error::ExternalSnafu)
82        };
83        Ok(Box::pin(AsyncRecordBatchStreamAdapter::new(
84            schema,
85            Box::pin(future),
86        )))
87    }
88
89    fn scan_plan(&self, request: ScanRequest) -> Result<Option<Arc<dyn ExecutionPlan>>> {
90        utils::inspect_plan(
91            &self.catalog_manager,
92            DatanodeInspectKind::CompactionTasks,
93            &self.schema,
94            request,
95        )
96    }
97}
catalog/system_schema/information_schema/compaction_tasks.rs

catalog/system_schema/information_schema/
compaction_tasks.rs
