feat(streaming): migrate TopNExecutor to ExecutorV2 (#1737)

risingwavelabs · Apr 11, 2022 · fc93222 · fc93222
1 parent 01947e4
commit fc93222
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
@@ -39,6 +39,9 @@ pub enum StreamExecutorError {
     #[error("input error")]
     InputError(RwError),
 
+    #[error("top n state error {0}")]
+    TopNStateError(RwError),
+
     #[error("channel `{0}` closed")]
     ChannelClosed(String),
 }
@@ -64,6 +67,10 @@ impl StreamExecutorError {
         Self::InputError(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
@@ -41,6 +41,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;
@@ -53,6 +55,7 @@ pub use lookup::*;
 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>;