Revert "feat(backfill): Persist backfill operator state"

ci/scripts/gen-flamegraph.sh

@@ -77,7 +77,6 @@ install_all() {
   # faster addr2line to speed up heap flamegraph analysis by jeprof
   echo ">>> Installing addr2line"
   git clone https://github.com/gimli-rs/addr2line
-  cp risingwave/rust-toolchain addr2line/rust-toolchain
   pushd addr2line
   cargo b --examples -r
   mv ./target/release/examples/addr2line $(which addr2line)

proto/stream_plan.proto

@@ -419,9 +419,6 @@ message ChainNode {
   // ChainType is used to decide which implementation for the ChainNode.
   ChainType chain_type = 4;
 
-  /// The state table used by Backfill operator for persisting internal state
-  catalog.Table state_table = 5;
-
   // The upstream materialized view info used by backfill.
   plan_common.StorageTableDesc table_desc = 7;
 }

src/common/src/hash/consistent_hash/bitmap.rs

@@ -25,11 +25,6 @@ impl Bitmap {
         self.iter_ones().map(VirtualNode::from_index)
     }
 
-    /// Enumerates the virtual nodes set to 1 in the bitmap.
-    pub fn iter_vnodes_scalar(&self) -> impl Iterator<Item = i16> + '_ {
-        self.iter_vnodes().map(|vnode| vnode.to_scalar())
-    }
-
     /// Returns an iterator which yields the position ranges of continuous virtual nodes set to 1 in
     /// the bitmap.
     pub fn vnode_ranges(&self) -> impl Iterator<Item = RangeInclusive<VirtualNode>> + '_ {

src/common/src/hash/consistent_hash/vnode.rs

@@ -18,7 +18,7 @@ use parse_display::Display;
 use crate::array::{Array, ArrayImpl, DataChunk};
 use crate::hash::Crc32HashCode;
 use crate::row::{Row, RowExt};
-use crate::types::{DataType, ScalarRefImpl};
+use crate::types::ScalarRefImpl;
 use crate::util::hash_util::Crc32FastBuilder;
 use crate::util::row_id::extract_vnode_id_from_row_id;
 
@@ -64,9 +64,6 @@ pub type AllVirtualNodeIter = std::iter::Map<std::ops::Range<usize>, fn(usize) -
 impl VirtualNode {
     /// The maximum value of the virtual node.
     pub const MAX: VirtualNode = VirtualNode::from_index(Self::COUNT - 1);
-    /// We may use `VirtualNode` as a datum in a stream, or store it as a column.
-    /// Hence this reifies it as a RW datatype.
-    pub const RW_TYPE: DataType = DataType::Int16;
     /// The minimum (zero) value of the virtual node.
     pub const ZERO: VirtualNode = VirtualNode::from_index(0);
 
@@ -132,8 +129,8 @@ impl VirtualNode {
             .collect()
     }
 
-    // `compute_row` is used to calculate the `VirtualNode` for the corresponding columns in a
-    // `Row`. Similar to `compute_chunk`, it also contains special handling for serial columns.
+    // `compute_row` is used to calculate the `VirtualNode` for the corresponding column in a `Row`.
+    // Similar to `compute_chunk`, it also contains special handling for serial columns.
     pub fn compute_row(row: impl Row, indices: &[usize]) -> VirtualNode {
         let project = row.project(indices);
         if let Ok(Some(ScalarRefImpl::Serial(s))) = project.iter().exactly_one().as_ref() {

src/common/src/util/stream_graph_visitor.rs

@@ -178,11 +178,6 @@ fn visit_stream_node_tables_inner<F>(
                 always!(node.state_table, "Sort");
             }
 
-            // Chain
-            NodeBody::Chain(node) => {
-                always!(node.state_table, "Chain")
-            }
-
             // Note: add internal tables for new nodes here.
             NodeBody::Materialize(node) if !internal_tables_only => {
                 always!(node.table, "Materialize")