feat(stream): add kafka backill executor

src/common/src/array/data_chunk.rs

@@ -121,6 +121,11 @@ impl DataChunk {
         self.visibility.next_set_bit(row_idx)
     }
 
+    /// Return the prev visible row index ~~on or~~ before `row_idx`.
+    pub fn prev_visible_row_idx(&self, row_idx: usize) -> Option<usize> {
+        self.visibility.next_set_bit(row_idx)
+    }
+
     pub fn into_parts(self) -> (Vec<ArrayRef>, Bitmap) {
         (self.columns.to_vec(), self.visibility)
     }

src/common/src/array/data_chunk_iter.rs

@@ -85,6 +85,24 @@ impl<'a> Iterator for DataChunkRefIter<'a> {
     }
 }
 
+impl<'a> DoubleEndedIterator for DataChunkRefIter<'a> {
+    fn next_back(&mut self) -> Option<Self::Item> {
+        if self.idx.start == self.idx.end {
+            return None;
+        }
+        match self.chunk.prev_visible_row_idx(self.idx.end) {
+            Some(idx) if idx >= self.idx.start => {
+                self.idx.end = idx;
+                Some(RowRef::new(self.chunk, idx))
+            }
+            _ => {
+                self.idx.end = self.idx.start;
+                None
+            }
+        }
+    }
+}
+
 impl<'a> FusedIterator for DataChunkRefIter<'a> {}
 
 pub struct DataChunkRefIterWithHoles<'a> {

src/common/src/array/stream_chunk_iter.rs

@@ -35,6 +35,12 @@ impl StreamChunk {
         self.rows_in(0..self.capacity())
     }
 
+    /// # Panics
+    /// If the chunk is empty.
+    pub fn last_row(&self) -> RowRef<'_> {
+        self.data_chunk().rows().next_back().unwrap()
+    }
+
     /// Return an iterator on rows of this stream chunk in a range.
     pub fn rows_in(&self, range: Range<usize>) -> impl Iterator<Item = (Op, RowRef<'_>)> {
         self.data_chunk().rows_in(range).map(|row| {

src/common/src/buffer/bitmap.rs

@@ -315,6 +315,13 @@ impl Bitmap {
         (bit_idx..self.len()).find(|&idx| unsafe { self.is_set_unchecked(idx) })
     }
 
+    /// Return the prev set bit index ~~on or~~ before `bit_idx`.
+    pub fn prev_set_bit(&self, bit_idx: usize) -> Option<usize> {
+        (0..bit_idx)
+            .rev()
+            .find(|&idx| unsafe { self.is_set_unchecked(idx) })
+    }
+
     /// Counts the number of bits set to 1.
     pub fn count_ones(&self) -> usize {
         self.count_ones

src/common/src/types/mod.rs

@@ -678,13 +678,17 @@ macro_rules! impl_convert {
 
             paste! {
                 impl ScalarImpl {
+                    /// # Panics
+                    /// If the scalar is not of the expected type.
                     pub fn [<as_ $suffix_name>](&self) -> &$scalar {
                         match self {
                             Self::$variant_name(ref scalar) => scalar,
                             other_scalar => panic!("cannot convert ScalarImpl::{} to concrete type {}", other_scalar.get_ident(), stringify!($variant_name))
                         }
                     }
 
+                    /// # Panics
+                    /// If the scalar is not of the expected type.
                     pub fn [<into_ $suffix_name>](self) -> $scalar {
                         match self {
                             Self::$variant_name(scalar) => scalar,
@@ -694,7 +698,8 @@ macro_rules! impl_convert {
                 }
 
                 impl <'scalar> ScalarRefImpl<'scalar> {
-                    // Note that this conversion consume self.
+                    /// # Panics
+                    /// If the scalar is not of the expected type.
                     pub fn [<into_ $suffix_name>](self) -> $scalar_ref {
                         match self {
                             Self::$variant_name(inner) => inner,

src/connector/src/source/base.rs

@@ -330,6 +330,7 @@ pub type BoxTryStream<M> = BoxStream<'static, Result<M, RwError>>;
 #[derive(Clone, Debug, PartialEq)]
 pub struct StreamChunkWithState {
     pub chunk: StreamChunk,
+    /// The latest offsets for each split in this chunk.
     pub split_offset_mapping: Option<HashMap<SplitId, String>>,
 }
 

src/source/src/connector_source.rs

@@ -107,6 +107,7 @@ impl ConnectorSource {
         }
     }
 
+    /// If `state` is `None`, returns a pending stream.
     pub async fn stream_reader(
         &self,
         state: ConnectorState,