risingwave_frontend/optimizer/plan_node/
batch_get_channel_delta_stats.rs

1// Copyright 2025 RisingWave Labs
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 risingwave_pb::batch_plan::plan_node::NodeBody;
16
17use super::batch::prelude::*;
18use super::utils::impl_distill_by_unit;
19use super::{
20    BatchPlanRef as PlanRef, ExprRewritable, PlanBase, PlanTreeNodeLeaf, ToBatchPb,
21    ToDistributedBatch, generic,
22};
23use crate::error::Result;
24use crate::optimizer::plan_node::ToLocalBatch;
25use crate::optimizer::property::{Distribution, Order};
26
27/// `BatchGetChannelDeltaStats` represents a batch plan node that retrieves channel statistics
28/// from the dashboard API. It has no inputs and returns channel stats data.
29#[derive(Debug, Clone, PartialEq, Eq, Hash)]
30pub struct BatchGetChannelDeltaStats {
31    pub base: PlanBase<Batch>,
32    core: generic::GetChannelDeltaStats,
33}
34
35impl PlanTreeNodeLeaf for BatchGetChannelDeltaStats {}
36impl_plan_tree_node_for_leaf! { Batch, BatchGetChannelDeltaStats }
37
38impl BatchGetChannelDeltaStats {
39    pub fn new(core: generic::GetChannelDeltaStats) -> Self {
40        Self::with_dist(core, Distribution::Single)
41    }
42
43    pub fn with_dist(core: generic::GetChannelDeltaStats, dist: Distribution) -> Self {
44        let base = PlanBase::new_batch_with_core(&core, dist, Order::any());
45        Self { base, core }
46    }
47
48    /// Get the `at_time` parameter
49    pub fn at_time(&self) -> Option<u64> {
50        self.core.at_time
51    }
52
53    /// Get the `time_offset` parameter
54    pub fn time_offset(&self) -> Option<u64> {
55        self.core.time_offset
56    }
57}
58
59impl_distill_by_unit!(BatchGetChannelDeltaStats, core, "BatchGetChannelDeltaStats");
60
61impl ToDistributedBatch for BatchGetChannelDeltaStats {
62    fn to_distributed(&self) -> Result<PlanRef> {
63        Ok(Self::with_dist(self.core.clone(), Distribution::Single).into())
64    }
65}
66
67impl ToBatchPb for BatchGetChannelDeltaStats {
68    fn to_batch_prost_body(&self) -> NodeBody {
69        use risingwave_pb::batch_plan::GetChannelDeltaStatsNode;
70
71        NodeBody::GetChannelDeltaStats(GetChannelDeltaStatsNode {
72            at_time: self.core.at_time,
73            time_offset: self.core.time_offset,
74        })
75    }
76}
77
78impl ToLocalBatch for BatchGetChannelDeltaStats {
79    fn to_local(&self) -> Result<PlanRef> {
80        Ok(Self::with_dist(self.core.clone(), Distribution::Single).into())
81    }
82}
83
84impl ExprRewritable<Batch> for BatchGetChannelDeltaStats {}
85
86impl crate::optimizer::plan_node::expr_visitable::ExprVisitable for BatchGetChannelDeltaStats {}
risingwave_frontend/optimizer/plan_node/batch_get_channel_delta_stats.rs

risingwave_frontend/optimizer/plan_node/
batch_get_channel_delta_stats.rs
