feat(streaming): migrate TopNExecutor to ExecutorV2

risingwavelabs · Apr 10, 2022 · ed71798 · ed71798
1 parent 3f5f703
commit ed71798
Show file tree

Hide file tree

Showing 7 changed files with 1,075 additions and 846 deletions.
diff --git a/src/stream/src/executor/top_n.rs b/src/stream/src/executor/top_n.rs
diff --git a/src/stream/src/executor/top_n_appendonly.rs b/src/stream/src/executor/top_n_appendonly.rs
@@ -390,7 +390,7 @@ impl<S: StateStore> StatefulExecutor for AppendOnlyTopNExecutor<S> {
     }
 }
 
-pub(super) fn generate_output(
+pub fn generate_output(
     new_rows: Vec<Row>,
     new_ops: Vec<Op>,
     schema: &Schema,

diff --git a/src/stream/src/executor_v2/error.rs b/src/stream/src/executor_v2/error.rs
@@ -36,6 +36,9 @@ pub enum StreamExecutorError {
     #[error("aggregate state error {0}")]
     AggStateError(RwError),
 
+    #[error("top n state error {0}")]
+    TopNStateError(RwError),
+
     #[error("channel `{0}` closed")]
     ChannelClosed(String),
 }
@@ -57,6 +60,10 @@ impl StreamExecutorError {
         Self::AggStateError(error.into()).into()
     }
 
+    pub fn top_n_state_error(error: impl Into<RwError>) -> TracedStreamExecutorError {
+        Self::TopNStateError(error.into()).into()
+    }
+
     pub fn channel_closed(name: impl Into<String>) -> TracedStreamExecutorError {
         Self::ChannelClosed(name.into()).into()
     }

diff --git a/src/stream/src/executor_v2/mod.rs b/src/stream/src/executor_v2/mod.rs
@@ -39,6 +39,8 @@ pub mod receiver;
 mod simple;
 #[cfg(test)]
 mod test_utils;
+mod top_n;
+mod top_n_executor;
 mod v1_compat;
 
 pub use batch_query::BatchQueryExecutor;
@@ -50,6 +52,7 @@ pub use local_simple_agg::LocalSimpleAggExecutor;
 pub use merge::MergeExecutor;
 pub use mview::*;
 pub(crate) use simple::{SimpleExecutor, SimpleExecutorWrapper};
+pub use top_n::TopNExecutor;
 pub use v1_compat::StreamExecutorV1;
 
 pub type BoxedExecutor = Box<dyn Executor>;