refactor(streaming): remove get_compacted_row from StateTable

src/stream/src/common/table/state_table.rs

@@ -30,7 +30,7 @@ use risingwave_common::catalog::{
     get_dist_key_in_pk_indices, ColumnDesc, ColumnId, TableId, TableOption,
 };
 use risingwave_common::hash::{VirtualNode, VnodeBitmapExt, VnodeCountCompat};
-use risingwave_common::row::{self, once, CompactedRow, Once, OwnedRow, Row, RowExt};
+use risingwave_common::row::{self, once, Once, OwnedRow, Row, RowExt};
 use risingwave_common::types::{DataType, Datum, DefaultOrd, DefaultOrdered, ScalarImpl};
 use risingwave_common::util::column_index_mapping::ColIndexMapping;
 use risingwave_common::util::epoch::EpochPair;
@@ -660,25 +660,6 @@ where
             .map_err(Into::into)
     }
 
-    /// Get a row in value-encoding format from state table.
-    pub async fn get_compacted_row(
-        &self,
-        pk: impl Row,
-    ) -> StreamExecutorResult<Option<CompactedRow>> {
-        if self.row_serde.kind().is_basic() {
-            // Basic serde is in value-encoding format, which is compatible with the compacted row.
-            self.get_encoded_row(pk)
-                .await
-                .map(|bytes| bytes.map(CompactedRow::new))
-        } else {
-            // For other encodings, we must first deserialize it into a `Row` first, then serialize
-            // it back into value-encoding format.
-            self.get_row(pk)
-                .await
-                .map(|row| row.map(CompactedRow::from))
-        }
-    }
-
     /// Update the vnode bitmap of the state table, returns the previous vnode bitmap.
     #[must_use = "the executor should decide whether to manipulate the cache based on the previous vnode bitmap"]
     pub fn update_vnode_bitmap(&mut self, new_vnodes: Arc<Bitmap>) -> (Arc<Bitmap>, bool) {

src/stream/src/executor/mview/materialize.rs

@@ -741,18 +741,19 @@ impl<SD: ValueRowSerde> MaterializeCache<SD> {
             }
             futures.push(async {
                 let key_row = table.pk_serde().deserialize(key).unwrap();
-                (key.to_vec(), table.get_compacted_row(&key_row).await)
+                let row = table.get_row(key_row).await?.map(CompactedRow::from);
+                StreamExecutorResult::Ok((key.to_vec(), row))
             });
         }
 
         let mut buffered = stream::iter(futures).buffer_unordered(10).fuse();
         while let Some(result) = buffered.next().await {
-            let (key, value) = result;
+            let (key, value) = result?;
             match conflict_behavior {
                 ConflictBehavior::Overwrite | ConflictBehavior::DoUpdateIfNotNull => {
-                    self.data.push(key, value?)
+                    self.data.push(key, value)
                 }
-                ConflictBehavior::IgnoreConflict => self.data.push(key, value?),
+                ConflictBehavior::IgnoreConflict => self.data.push(key, value),
                 _ => unreachable!(),
             };
         }