feat(streaming): introduce sort operator based on watermark

proto/stream_plan.proto

@@ -395,6 +395,14 @@ message ProjectSetNode {
   repeated expr.ProjectSetSelectItem select_list = 1;
 }
 
+// Sorts inputs and outputs ordered data based on watermark.
+message SortNode {
+  // Persists data above watermark.
+  catalog.Table state_table = 1;
+  // Column index of watermark to perform sorting.
+  uint32 sort_column_index = 2;
+}
+
 message StreamNode {
   oneof node_body {
     SourceNode source = 100;
@@ -422,6 +430,7 @@ message StreamNode {
     DynamicFilterNode dynamic_filter = 122;
     ProjectSetNode project_set = 123;
     GroupTopNNode group_top_n = 124;
+    SortNode sort = 125;
   }
   // The id for the operator. This is local per mview.
   // TODO: should better be a uint32.

src/common/src/buffer/bitmap.rs

@@ -264,6 +264,21 @@ impl Bitmap {
             num_bits: self.num_bits,
         }
     }
+
+    /// Performs bitwise saturate subtract on two equal-length bitmaps.
+    ///
+    /// For example, lhs = [01110] and rhs = [00111], then
+    /// `bit_saturate_subtract(lhs, rhs)` results in [01000]
+    pub fn bit_saturate_subtract(lhs: &Bitmap, rhs: &Bitmap) -> Bitmap {
+        assert_eq!(lhs.num_bits, rhs.num_bits);
+        let bits = lhs
+            .bits
+            .iter()
+            .zip_eq(rhs.bits.iter())
+            .map(|(&a, &b)| (!(a & b)) & a)
+            .collect();
+        Bitmap::from_bytes_with_num_bits(bits, lhs.num_bits)
+    }
 }
 
 impl<'a, 'b> BitAnd<&'b Bitmap> for &'a Bitmap {
@@ -464,6 +479,16 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_bitwise_saturate_subtract() {
+        let bitmap1 = Bitmap::from_bytes(Bytes::from_static(&[0b01101010]));
+        let bitmap2 = Bitmap::from_bytes(Bytes::from_static(&[0b01001110]));
+        assert_eq!(
+            Bitmap::from_bytes(Bytes::from_static(&[0b00100000])),
+            Bitmap::bit_saturate_subtract(&bitmap1, &bitmap2)
+        );
+    }
+
     #[test]
     fn test_bitmap_is_set() {
         let bitmap = Bitmap::from_bytes(Bytes::from_static(&[0b01001010]));

src/storage/src/table/streaming_table/state_table.rs

@@ -350,7 +350,7 @@ impl<S: StateStore> StateTable<S> {
     }
 
     /// Get the vnode value with given (prefix of) primary key
-    fn compute_vnode(&self, pk_prefix: &Row) -> VirtualNode {
+    pub fn compute_vnode(&self, pk_prefix: &Row) -> VirtualNode {
         let prefix_len = pk_prefix.0.len();
         if let Some(vnode_col_idx_in_pk) = self.vnode_col_idx_in_pk {
             let vnode = pk_prefix.0.get(vnode_col_idx_in_pk).unwrap();
@@ -371,6 +371,10 @@ impl<S: StateStore> StateTable<S> {
         &self.pk_serde
     }
 
+    pub fn vnode_bitmap(&self) -> &Bitmap {
+        &self.vnodes
+    }
+
     pub fn is_dirty(&self) -> bool {
         self.mem_table.is_dirty()
     }

src/stream/src/executor/mod.rs

@@ -74,6 +74,7 @@ mod rearranged_chain;
 mod receiver;
 mod simple;
 mod sink;
+mod sort;
 pub mod source;
 pub mod subtask;
 mod top_n;
@@ -111,6 +112,7 @@ pub use receiver::ReceiverExecutor;
 use risingwave_pb::source::{ConnectorSplit, ConnectorSplits};
 use simple::{SimpleExecutor, SimpleExecutorWrapper};
 pub use sink::SinkExecutor;
+pub use sort::SortExecutor;
 pub use source::*;
 pub use top_n::{AppendOnlyTopNExecutor, GroupTopNExecutor, TopNExecutor};
 pub use union::UnionExecutor;