Repository: risingwavelabs/risingwave (language: rust)

Issue:
cherry-pick fix(hash agg): split `AggGroup` inter states and outputs change building, fixing old groups not deleted bug (#20435) to branch release-2.0
Cherry picking \#20435 onto branch release-2.0,

This PR/issue was created by cherry-pick action from commit 6f25c0386af269adf0799c16a62ef65633e21218.,


Discussion / hints:

Fix patch:
diff --git a/src/stream/src/common/table/state_table.rs b/src/stream/src/common/table/state_table.rs
index 2dac8d1dc09ef..13f4125552faf 100644
--- a/src/stream/src/common/table/state_table.rs
+++ b/src/stream/src/common/table/state_table.rs
@@ -1026,18 +1026,6 @@ where
         self.update_inner(new_key_bytes, Some(old_value_bytes), new_value_bytes);
     }
 
-    /// Update a row without giving old value.
-    ///
-    /// `op_consistency_level` should be set to `Inconsistent`.
-    pub fn update_without_old_value(&mut self, new_value: impl Row) {
-        let new_pk = (&new_value).project(self.pk_indices());
-        let new_key_bytes =
-            serialize_pk_with_vnode(new_pk, &self.pk_serde, self.compute_vnode_by_pk(new_pk));
-        let new_value_bytes = self.serialize_value(new_value);
-
-        self.update_inner(new_key_bytes, None, new_value_bytes);
-    }
-
     /// Write a record into state table. Must have the same schema with the table.
     pub fn write_record(&mut self, record: Record<impl Row>) {
         match record {
diff --git a/src/stream/src/executor/aggregation/agg_group.rs b/src/stream/src/executor/aggregation/agg_group.rs
index 9a37507370289..a1a5531710b3c 100644
--- a/src/stream/src/executor/aggregation/agg_group.rs
+++ b/src/stream/src/executor/aggregation/agg_group.rs
@@ -34,14 +34,58 @@ use crate::consistency::consistency_panic;
 use crate::executor::error::StreamExecutorResult;
 use crate::executor::PkIndices;
 
+#[derive(Debug)]
+pub struct Context {
+    group_key: Option<GroupKey>,
+}
+
+impl Context {
+    pub fn group_key(&self) -> Option<&GroupKey> {
+        self.group_key.as_ref()
+    }
+
+    pub fn group_key_row(&self) -> OwnedRow {
+        self.group_key()
+            .map(GroupKey::table_row)
+            .cloned()
+            .unwrap_or_default()
+    }
+}
+
+fn row_count_of(ctx: &Context, row: Option<impl Row>, row_count_col: usize) -> usize {
+    match row {
+        Some(row) => {
+            let mut row_count = row
+                .datum_at(row_count_col)
+                .expect("row count field should not be NULL")
+                .into_int64();
+
+            if row_count < 0 {
+                consistency_panic!(group = ?ctx.group_key_row(), row_count, "row count should be non-negative");
+
+                // NOTE: Here is the case that an inconsistent `DELETE` arrives at HashAgg executor, and there's no
+                // corresponding group existing before (or has been deleted). In this case, previous row count should
+                // be `0` and current row count be `-1` after handling the `DELETE`. To ignore the inconsistency, we
+                // reset `row_count` to `0` here, so that `OnlyOutputIfHasInput` will return no change, so that the
+                // inconsistent will be hidden from downstream. This won't prevent from incorrect results of existing
+                // groups, but at least can prevent from downstream panicking due to non-existing keys.
+                // See https://github.com/risingwavelabs/risingwave/issues/14031 for more information.
+                row_count = 0;
+            }
+            row_count.try_into().unwrap()
+        }
+        None => 0,
+    }
+}
+
 pub trait Strategy {
     /// Infer the change type of the aggregation result. Don't need to take the ownership of
-    /// `prev_outputs` and `curr_outputs`.
+    /// `prev_row` and `curr_row`.
     fn infer_change_type(
-        prev_row_count: usize,
-        curr_row_count: usize,
-        prev_outputs: Option<&OwnedRow>,
-        curr_outputs: &OwnedRow,
+        ctx: &Context,
+        prev_row: Option<&OwnedRow>,
+        curr_row: &OwnedRow,
+        row_count_col: usize,
     ) -> Option<RecordType>;
 }
 
@@ -53,12 +97,13 @@ pub struct OnlyOutputIfHasInput;
 
 impl Strategy for AlwaysOutput {
     fn infer_change_type(
-        prev_row_count: usize,
-        _curr_row_count: usize,
-        prev_outputs: Option<&OwnedRow>,
-        _curr_outputs: &OwnedRow,
+        ctx: &Context,
+        prev_row: Option<&OwnedRow>,
+        _curr_row: &OwnedRow,
+        row_count_col: usize,
     ) -> Option<RecordType> {
-        match prev_outputs {
+        let prev_row_count = row_count_of(ctx, prev_row, row_count_col);
+        match prev_row {
             None => {
                 // First time to build changes, assert to ensure correctness.
                 // Note that it's not true vice versa, i.e. `prev_row_count == 0` doesn't imply
@@ -85,11 +130,14 @@ impl Strategy for AlwaysOutput {
 
 impl Strategy for OnlyOutputIfHasInput {
     fn infer_change_type(
-        prev_row_count: usize,
-        curr_row_count: usize,
-        prev_outputs: Option<&OwnedRow>,
-        curr_outputs: &OwnedRow,
+        ctx: &Context,
+        prev_row: Option<&OwnedRow>,
+        curr_row: &OwnedRow,
+        row_count_col: usize,
     ) -> Option<RecordType> {
+        let prev_row_count = row_count_of(ctx, prev_row, row_count_col);
+        let curr_row_count = row_count_of(ctx, Some(curr_row), row_count_col);
+
         match (prev_row_count, curr_row_count) {
             (0, 0) => {
                 // No rows of current group exist.
@@ -105,7 +153,7 @@ impl Strategy for OnlyOutputIfHasInput {
             }
             (_, _) => {
                 // Update output row.
-                if prev_outputs.expect("must exist previous outputs") == curr_outputs {
+                if prev_row.expect("must exist previous row") == curr_row {
                     // No output change.
                     None
                 } else {
@@ -159,32 +207,43 @@ impl GroupKey {
 
 /// [`AggGroup`] manages agg states of all agg calls for one `group_key`.
 pub struct AggGroup<S: StateStore, Strtg: Strategy> {
-    /// Group key.
-    group_key: Option<GroupKey>,
+    /// Agg group context, containing the group key.
+    ctx: Context,
 
     /// Current managed states for all [`AggCall`]s.
     states: Vec<AggState>,
 
-    /// Previous outputs of aggregate functions. Initializing with `None`.
+    /// Previous intermediate states, stored in the intermediate state table.
+    prev_inter_states: Option<OwnedRow>,
+
+    /// Previous outputs, yielded to downstream.
+    /// If `EOWC` is true, this field is not used.
     prev_outputs: Option<OwnedRow>,
 
     /// Index of row count agg call (`count(*)`) in the call list.
     row_count_index: usize,
 
+    /// Whether the emit policy is EOWC.
+    emit_on_window_close: bool,
+
     _phantom: PhantomData<(S, Strtg)>,
 }
 
 impl<S: StateStore, Strtg: Strategy> Debug for AggGroup<S, Strtg> {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         f.debug_struct("AggGroup")
-            .field("group_key", &self.group_key)
+            .field("group_key", &self.ctx.group_key)
+            .field("prev_inter_states", &self.prev_inter_states)
             .field("prev_outputs", &self.prev_outputs)
+            .field("row_count_index", &self.row_count_index)
+            .field("emit_on_window_close", &self.emit_on_window_close)
             .finish()
     }
 }
 
 impl<S: StateStore, Strtg: Strategy> EstimateSize for AggGroup<S, Strtg> {
     fn estimated_heap_size(&self) -> usize {
+        // TODO(rc): should include the size of `prev_inter_states` and `prev_outputs`
         self.states
             .iter()
             .map(|state| state.estimated_heap_size())
@@ -205,14 +264,15 @@ impl<S: StateStore, Strtg: Strategy> AggGroup<S, Strtg> {
         intermediate_state_table: &StateTable<S>,
         pk_indices: &PkIndices,
         row_count_index: usize,
+        emit_on_window_close: bool,
         extreme_cache_size: usize,
         input_schema: &Schema,
     ) -> StreamExecutorResult<Self> {
-        let encoded_states = intermediate_state_table
+        let inter_states = intermediate_state_table
             .get_row(group_key.as_ref().map(GroupKey::table_pk))
             .await?;
-        if let Some(encoded_states) = &encoded_states {
-            assert_eq!(encoded_states.len(), agg_calls.len());
+        if let Some(inter_states) = &inter_states {
+            assert_eq!(inter_states.len(), agg_calls.len());
         }
 
         let mut states = Vec::with_capacity(agg_calls.len());
@@ -222,7 +282,7 @@ impl<S: StateStore, Strtg: Strategy> AggGroup<S, Strtg> {
                 agg_call,
                 agg_func,
                 &storages[idx],
-                encoded_states.as_ref().map(|outputs| &outputs[idx]),
+                inter_states.as_ref().map(|s| &s[idx]),
                 pk_indices,
                 extreme_cache_size,
                 input_schema,
@@ -231,32 +291,36 @@ impl<S: StateStore, Strtg: Strategy> AggGroup<S, Strtg> {
         }
 
         let mut this = Self {
-            group_key,
+            ctx: Context { group_key },
             states,
-            prev_outputs: None, // will be initialized later
+            prev_inter_states: inter_states,
+            prev_outputs: None, // will be set below
             row_count_index,
+            emit_on_window_close,
             _phantom: PhantomData,
         };
 
-        if encoded_states.is_some() {
-            let (_, outputs) = this.get_outputs(storages, agg_funcs).await?;
+        if !this.emit_on_window_close && this.prev_inter_states.is_some() {
+            let outputs = this.get_outputs(storages, agg_funcs).await?;
             this.prev_outputs = Some(outputs);
         }
 
         Ok(this)
     }
 
-    /// Create a group from encoded states for EOWC. The previous output is set to `None`.
+    /// Create a group from intermediate states for EOWC output.
+    /// Will always produce `Insert` when building change.
     #[allow(clippy::too_many_arguments)]
-    pub fn create_eowc(
+    pub fn for_eowc_output(
         version: PbAggNodeVersion,
         group_key: Option<GroupKey>,
         agg_calls: &[AggCall],
         agg_funcs: &[BoxedAggregateFunction],
         storages: &[AggStateStorage<S>],
-        encoded_states: &OwnedRow,
+        inter_states: &OwnedRow,
         pk_indices: &PkIndices,
         row_count_index: usize,
+        emit_on_window_close: bool,
         extreme_cache_size: usize,
         input_schema: &Schema,
     ) -> StreamExecutorResult<Self> {
@@ -267,7 +331,7 @@ impl<S: StateStore, Strtg: Strategy> AggGroup<S, Strtg> {
                 agg_call,
                 agg_func,
                 &storages[idx],
-                Some(&encoded_states[idx]),
+                Some(&inter_states[idx]),
                 pk_indices,
                 extreme_cache_size,
                 input_schema,
@@ -276,36 +340,18 @@ impl<S: StateStore, Strtg: Strategy> AggGroup<S, Strtg> {
         }
 
         Ok(Self {
-            group_key,
+            ctx: Context { group_key },
             states,
-            prev_outputs: None,
+            prev_inter_states: None, // this doesn't matter
+            prev_outputs: None,      // this will make sure the outputs change to be `Insert`
             row_count_index,
+            emit_on_window_close,
             _phantom: PhantomData,
         })
     }
 
     pub fn group_key(&self) -> Option<&GroupKey> {
-        self.group_key.as_ref()
-    }
-
-    pub fn group_key_row(&self) -> OwnedRow {
-        self.group_key
-            .as_ref()
-            .map(GroupKey::table_row)
-            .cloned()
-            .unwrap_or_default()
-    }
-
-    fn prev_row_count(&self) -> usize {
-        match &self.prev_outputs {
-            Some(states) => states[self.row_count_index]
-                .as_ref()
-                .map(|x| {
-                    TryInto::try_into(*x.as_int64()).expect("row count should be non-negative")
-                })
-                .unwrap_or(0),
-            None => 0,
-        }
+        self.ctx.group_key()
     }
 
     /// Get current row count of this group.
@@ -314,27 +360,11 @@ impl<S: StateStore, Strtg: Strategy> AggGroup<S, Strtg> {
             self.states[self.row_count_index],
             AggState::Value(ref state) => state
         );
-        let mut row_count = *row_count_state
-            .as_datum()
-            .as_ref()
-            .expect("row count state should not be NULL")
-            .as_int64();
-        if row_count < 0 {
-            consistency_panic!(group = ?self.group_key_row(), row_count, "row count should be non-negative");
-
-            // NOTE: Here is the case that an inconsistent `DELETE` arrives at HashAgg executor, and there's no
-            // corresponding group existing before (or has been deleted). In this case, `prev_row_count()` will
-            // report `0`. To ignore the inconsistent, we set `curr_row_count` to `0` here, so that `OnlyOutputIfHasInput`
-            // will return no change, so that the inconsistent will be hidden from downstream. This won't prevent from
-            // incorrect results of existing groups, but at least can prevent from downstream panicking due to non-existing
-            // keys. See https://github.com/risingwavelabs/risingwave/issues/14031 for more information.
-            row_count = 0;
-        }
-        row_count.try_into().unwrap()
+        row_count_of(&self.ctx, Some([row_count_state.as_datum().clone()]), 0)
     }
 
     pub(crate) fn is_uninitialized(&self) -> bool {
-        self.prev_outputs.is_none()
+        self.prev_inter_states.is_none()
     }
 
     /// Apply input chunk to all managed agg states.
@@ -349,7 +379,7 @@ impl<S: StateStore, Strtg: Strategy> AggGroup<S, Strtg> {
         visibilities: Vec<Bitmap>,
     ) -> StreamExecutorResult<()> {
         if self.curr_row_count() == 0 {
-            tracing::trace!(group = ?self.group_key_row(), "first time see this group");
+            tracing::trace!(group = ?self.ctx.group_key_row(), "first time see this group");
         }
         for (((state, call), func), visibility) in (self.states.iter_mut())
             .zip_eq_fast(calls)
@@ -360,7 +390,7 @@ impl<S: StateStore, Strtg: Strategy> AggGroup<S, Strtg> {
         }
 
         if self.curr_row_count() == 0 {
-            tracing::trace!(group = ?self.group_key_row(), "last time see this group");
+            tracing::trace!(group = ?self.ctx.group_key_row(), "last time see this group");
         }
 
         Ok(())
@@ -375,26 +405,18 @@ impl<S: StateStore, Strtg: Strategy> AggGroup<S, Strtg> {
         Ok(())
     }
 
-    /// Encode intermediate states.
-    pub fn encode_states(
-        &self,
-        funcs: &[BoxedAggregateFunction],
-    ) -> StreamExecutorResult<OwnedRow> {
-        let mut encoded_states = Vec::with_capacity(self.states.len());
+    /// Get the encoded intermediate states of all managed agg states.
+    fn get_inter_states(&self, funcs: &[BoxedAggregateFunction]) -> StreamExecutorResult<OwnedRow> {
+        let mut inter_states = Vec::with_capacity(self.states.len());
         for (state, func) in self.states.iter().zip_eq_fast(funcs) {
             let encoded = match state {
                 AggState::Value(s) => func.encode_state(s)?,
                 // For minput state, we don't need to store it in state table.
                 AggState::MaterializedInput(_) => None,
             };
-            encoded_states.push(encoded);
+            inter_states.push(encoded);
         }
-        let states = self
-            .group_key()
-            .map(GroupKey::table_row)
-            .chain(OwnedRow::new(encoded_states))
-            .into_owned_row();
-        Ok(states)
+        Ok(OwnedRow::new(inter_states))
     }
 
     /// Get the outputs of all managed agg states, without group key prefix.
@@ -405,12 +427,13 @@ impl<S: StateStore, Strtg: Strategy> AggGroup<S, Strtg> {
         &mut self,
         storages: &[AggStateStorage<S>],
         funcs: &[BoxedAggregateFunction],
-    ) -> StreamExecutorResult<(usize, OwnedRow)> {
+    ) -> StreamExecutorResult<OwnedRow> {
         let row_count = self.curr_row_count();
         if row_count == 0 {
             // Reset all states (in fact only value states will be reset).
             // This is important because for some agg calls (e.g. `sum`), if no row is applied,
             // they should output NULL, for some other calls (e.g. `sum0`), they should output 0.
+            // This actually also prevents inconsistent negative row count from being worse.
             // FIXME(rc): Deciding whether to reset states according to `row_count` is not precisely
             // correct, see https://github.com/risingwavelabs/risingwave/issues/7412 for bug description.
             self.reset(funcs)?;
@@ -421,45 +444,124 @@ impl<S: StateStore, Strtg: Strategy> AggGroup<S, Strtg> {
                 .zip_eq_fast(storages)
                 .zip_eq_fast(funcs)
                 .map(|((state, storage), func)| {
-                    state.get_output(storage, func, self.group_key.as_ref())
+                    state.get_output(storage, func, self.ctx.group_key())
                 }),
         )
         .await
-        .map(|row| (row_count, OwnedRow::new(row)))
+        .map(OwnedRow::new)
+    }
+
+    /// Build change for aggregation intermediate states, according to previous and current agg states.
+    /// The change should be applied to the intermediate state table.
+    ///
+    /// The saved previous inter states will be updated to the latest states after calling this method.
+    pub fn build_states_change(
+        &mut self,
+        funcs: &[BoxedAggregateFunction],
+    ) -> StreamExecutorResult<Option<Record<OwnedRow>>> {
+        let curr_inter_states = self.get_inter_states(funcs)?;
+        let change_type = Strtg::infer_change_type(
+            &self.ctx,
+            self.prev_inter_states.as_ref(),
+            &curr_inter_states,
+            self.row_count_index,
+        );
+
+        tracing::trace!(
+            group = ?self.ctx.group_key_row(),
+            prev_inter_states = ?self.prev_inter_states,
+            curr_inter_states = ?curr_inter_states,
+            change_type = ?change_type,
+            "build intermediate states change"
+        );
+
+        let Some(change_type) = change_type else {
+            return Ok(None);
+        };
+        Ok(Some(match change_type {
+            RecordType::Insert => {
+                let new_row = self
+                    .group_key()
+                    .map(GroupKey::table_row)
+                    .chain(&curr_inter_states)
+                    .into_owned_row();
+                self.prev_inter_states = Some(curr_inter_states);
+                Record::Insert { new_row }
+            }
+            RecordType::Delete => {
+                let prev_inter_states = self
+                    .prev_inter_states
+                    .take()
+                    .expect("must exist previous intermediate states");
+                let old_row = self
+                    .group_key()
+                    .map(GroupKey::table_row)
+                    .chain(prev_inter_states)
+                    .into_owned_row();
+                Record::Delete { old_row }
+            }
+            RecordType::Update => {
+                let new_row = self
+                    .group_key()
+                    .map(GroupKey::table_row)
+                    .chain(&curr_inter_states)
+                    .into_owned_row();
+                let prev_inter_states = self
+                    .prev_inter_states
+                    .replace(curr_inter_states)
+                    .expect("must exist previous intermediate states");
+                let old_row = self
+                    .group_key()
+                    .map(GroupKey::table_row)
+                    .chain(prev_inter_states)
+                    .into_owned_row();
+                Record::Update { old_row, new_row }
+            }
+        }))
     }
 
     /// Build aggregation result change, according to previous and current agg outputs.
+    /// The change should be yielded to downstream.
+    ///
     /// The saved previous outputs will be updated to the latest outputs after this method.
-    pub async fn build_change(
+    ///
+    /// Note that this method is very likely to cost more than `build_states_change`, because it
+    /// needs to produce output for materialized input states which may involve state table read.
+    pub async fn build_outputs_change(
         &mut self,
         storages: &[AggStateStorage<S>],
         funcs: &[BoxedAggregateFunction],
     ) -> StreamExecutorResult<Option<Record<OwnedRow>>> {
-        let prev_row_count = self.prev_row_count();
-        let (curr_row_count, curr_outputs) = self.get_outputs(storages, funcs).await?;
+        let curr_outputs = self.get_outputs(storages, funcs).await?;
 
         let change_type = Strtg::infer_change_type(
-            prev_row_count,
-            curr_row_count,
+            &self.ctx,
             self.prev_outputs.as_ref(),
             &curr_outputs,
+            self.row_count_index,
         );
 
         tracing::trace!(
-            group = ?self.group_key_row(),
-            prev_row_count,
-            curr_row_count,
+            group = ?self.ctx.group_key_row(),
+            prev_outputs = ?self.prev_outputs,
+            curr_outputs = ?curr_outputs,
             change_type = ?change_type,
-            "build change"
+            "build outputs change"
         );
 
-        Ok(change_type.map(|change_type| match change_type {
+        let Some(change_type) = change_type else {
+            return Ok(None);
+        };
+        Ok(Some(match change_type {
             RecordType::Insert => {
                 let new_row = self
                     .group_key()
                     .map(GroupKey::table_row)
                     .chain(&curr_outputs)
                     .into_owned_row();
+                // Although we say the `prev_outputs` field is not used in EOWC mode, we still
+                // do the same here to keep the code simple. When it's actually running in EOWC
+                // mode, `build_outputs_change` will be called only once for each group.
                 self.prev_outputs = Some(curr_outputs);
                 Record::Insert { new_row }
             }
diff --git a/src/stream/src/executor/hash_agg.rs b/src/stream/src/executor/hash_agg.rs
index 6c6a99f4c2421..3e9a0080a251c 100644
--- a/src/stream/src/executor/hash_agg.rs
+++ b/src/stream/src/executor/hash_agg.rs
@@ -298,6 +298,7 @@ impl<K: HashKey, S: StateStore> HashAggExecutor<K, S> {
                                 &this.intermediate_state_table,
                                 &this.input_pk_indices,
                                 this.row_count_index,
+                                this.emit_on_window_close,
                                 this.extreme_cache_size,
                                 &this.input_schema,
                             )
@@ -407,14 +408,17 @@ impl<K: HashKey, S: StateStore> HashAggExecutor<K, S> {
         let window_watermark = vars.window_watermark.take();
 
         // flush changed states into intermediate state table
-        for agg_group in vars.dirty_groups.values() {
-            let encoded_states = agg_group.encode_states(&this.agg_funcs)?;
+        for mut agg_group in vars.dirty_groups.values_mut() {
+            let Some(inter_states_change) = agg_group.build_states_change(&this.agg_funcs)? else {
+                continue;
+            };
+
             if this.emit_on_window_close {
                 vars.buffer
-                    .update_without_old_value(encoded_states, &mut this.intermediate_state_table);
+                    .apply_change(inter_states_change, &mut this.intermediate_state_table);
             } else {
                 this.intermediate_state_table
-                    .update_without_old_value(encoded_states);
+                    .write_record(inter_states_change);
             }
         }
 
@@ -432,9 +436,9 @@ impl<K: HashKey, S: StateStore> HashAggExecutor<K, S> {
                         .into_iter()
                         .take(this.group_key_indices.len())
                         .collect();
-                    let states = row.into_iter().skip(this.group_key_indices.len()).collect();
+                    let inter_states = row.into_iter().skip(this.group_key_indices.len()).collect();
 
-                    let mut agg_group = AggGroup::create_eowc(
+                    let mut agg_group = AggGroup::<S>::for_eowc_output(
                         this.version,
                         Some(GroupKey::new(
                             group_key,
@@ -443,15 +447,16 @@ impl<K: HashKey, S: StateStore> HashAggExecutor<K, S> {
                         &this.agg_calls,
                         &this.agg_funcs,
                         &this.storages,
-                        &states,
+                        &inter_states,
                         &this.input_pk_indices,
                         this.row_count_index,
+                        this.emit_on_window_close,
                         this.extreme_cache_size,
                         &this.input_schema,
                     )?;
 
                     let change = agg_group
-                        .build_change(&this.storages, &this.agg_funcs)
+                        .build_outputs_change(&this.storages, &this.agg_funcs)
                         .await?;
                     if let Some(change) = change {
                         if let Some(chunk) = vars.chunk_builder.append_record(change) {
@@ -466,7 +471,7 @@ impl<K: HashKey, S: StateStore> HashAggExecutor<K, S> {
             for mut agg_group in vars.dirty_groups.values_mut() {
                 let agg_group = agg_group.as_mut();
                 let change = agg_group
-                    .build_change(&this.storages, &this.agg_funcs)
+                    .build_outputs_change(&this.storages, &this.agg_funcs)
                     .await?;
                 if let Some(change) = change {
                     if let Some(chunk) = vars.chunk_builder.append_record(change) {
diff --git a/src/stream/src/executor/simple_agg.rs b/src/stream/src/executor/simple_agg.rs
index fdecd5b7a4502..b1f2d5dedf515 100644
--- a/src/stream/src/executor/simple_agg.rs
+++ b/src/stream/src/executor/simple_agg.rs
@@ -196,33 +196,39 @@ impl<S: StateStore> SimpleAggExecutor<S> {
             // Flush distinct dedup state.
             vars.distinct_dedup.flush(&mut this.distinct_dedup_tables)?;
 
-            // Flush states into intermediate state table.
-            let encoded_states = vars.agg_group.encode_states(&this.agg_funcs)?;
-            this.intermediate_state_table
-                .update_without_old_value(encoded_states);
+            // Build and apply change for intermediate states.
+            if let Some(inter_states_change) =
+                vars.agg_group.build_states_change(&this.agg_funcs)?
+            {
+                this.intermediate_state_table
+                    .write_record(inter_states_change);
+            }
         }
+        vars.state_changed = false;
 
-        // Retrieve modified states and put the changes into the builders.
-        let chunk = vars
+        // Build and apply change for the final outputs.
+        let outputs_change = vars
             .agg_group
-            .build_change(&this.storages, &this.agg_funcs)
-            .await?
-            .and_then(|change| {
-                if !this.must_output_per_barrier {
-                    if let Record::Update { old_row, new_row } = &change {
-                        if old_row == new_row {
-                            return None;
-                        }
-                    };
-                }
-                Some(change.to_stream_chunk(&this.info.schema.data_types()))
-            });
+            .build_outputs_change(&this.storages, &this.agg_funcs)
+            .await?;
+
+        let change =
+            outputs_change.expect("`AlwaysOutput` strategy will output a change in any case");
+
+        let chunk = if !this.must_output_per_barrier
+            && let Record::Update { old_row, new_row } = &change
+            && old_row == new_row
+        {
+            // for cases without approx percentile, we don't need to output the change if it's noop
+            None
+        } else {
+            Some(change.to_stream_chunk(&this.info.schema.data_types()))
+        };
 
         // Commit all state tables.
         futures::future::try_join_all(this.all_state_tables_mut().map(|table| table.commit(epoch)))
             .await?;
 
-        vars.state_changed = false;
         Ok(chunk)
     }
 
@@ -265,6 +271,7 @@ impl<S: StateStore> SimpleAggExecutor<S> {
                 &this.intermediate_state_table,
                 &this.input_pk_indices,
                 this.row_count_index,
+                false, // emit on window close
                 this.extreme_cache_size,
                 &this.input_schema,
             )
diff --git a/src/stream/src/executor/sort_buffer.rs b/src/stream/src/executor/sort_buffer.rs
index b3d658f3af4a5..25d27b14f0469 100644
--- a/src/stream/src/executor/sort_buffer.rs
+++ b/src/stream/src/executor/sort_buffer.rs
@@ -118,18 +118,6 @@ impl<S: StateStore> SortBuffer<S> {
         self.cache.insert(key, new_row.into_owned_row());
     }
 
-    /// Update a row in the buffer without giving the old value.
-    pub fn update_without_old_value(
-        &mut self,
-        new_row: impl Row,
-        buffer_table: &mut StateTable<S>,
-    ) {
-        buffer_table.update_without_old_value(&new_row);
-        let key = row_to_cache_key(self.sort_column_index, &new_row, buffer_table);
-        self.cache.delete(&key);
-        self.cache.insert(key, new_row.into_owned_row());
-    }
-
     /// Apply a change to the buffer, insert/delete/update.
     pub fn apply_change(&mut self, change: Record<impl Row>, buffer_table: &mut StateTable<S>) {
         match change {
diff --git a/src/stream/src/from_proto/hash_agg.rs b/src/stream/src/from_proto/hash_agg.rs
index 3331640519617..d47e17bd9f244 100644
--- a/src/stream/src/from_proto/hash_agg.rs
+++ b/src/stream/src/from_proto/hash_agg.rs
@@ -83,7 +83,7 @@ impl ExecutorBuilder for HashAggExecutorBuilder {
         )
         .await;
         // disable sanity check so that old value is not required when updating states
-        let intermediate_state_table = StateTable::from_table_catalog_inconsistent_op(
+        let intermediate_state_table = StateTable::from_table_catalog(
             node.get_intermediate_state_table().unwrap(),
             store.clone(),
             vnodes.clone(),
diff --git a/src/stream/src/from_proto/simple_agg.rs b/src/stream/src/from_proto/simple_agg.rs
index 689acc7d16a9c..86e8af9119ad0 100644
--- a/src/stream/src/from_proto/simple_agg.rs
+++ b/src/stream/src/from_proto/simple_agg.rs
@@ -45,7 +45,7 @@ impl ExecutorBuilder for SimpleAggExecutorBuilder {
             build_agg_state_storages_from_proto(node.get_agg_call_states(), store.clone(), None)
                 .await;
         // disable sanity check so that old value is not required when updating states
-        let intermediate_state_table = StateTable::from_table_catalog_inconsistent_op(
+        let intermediate_state_table = StateTable::from_table_catalog(
             node.get_intermediate_state_table().unwrap(),
             store.clone(),
             None,


Test patch:
diff --git a/ci/scripts/run-e2e-test.sh b/ci/scripts/run-e2e-test.sh
index ff88dced8780e..e4002fc5f4e9f 100755
--- a/ci/scripts/run-e2e-test.sh
+++ b/ci/scripts/run-e2e-test.sh
@@ -81,6 +81,7 @@ RUST_LOG="info,risingwave_stream=info,risingwave_batch=info,risingwave_storage=i
 cluster_start
 # Please make sure the regression is expected before increasing the timeout.
 sqllogictest -p 4566 -d dev './e2e_test/streaming/**/*.slt' --junit "streaming-${profile}"
+risedev slt -p 4566 -d dev './e2e_test/queryable_internal_state/**/*.slt' --junit "queryable-internal-state-${profile}"
 sqllogictest -p 4566 -d dev './e2e_test/backfill/sink/different_pk_and_dist_key.slt'
 
 echo "--- Kill cluster"
diff --git a/e2e_test/commands/internal_table.mjs b/e2e_test/commands/internal_table.mjs
index b51cc0313dca2..1deb727e7c3fe 100755
--- a/e2e_test/commands/internal_table.mjs
+++ b/e2e_test/commands/internal_table.mjs
@@ -17,7 +17,7 @@ async function psql(query) {
   return (
     await $`
 psql -h $RISEDEV_RW_FRONTEND_LISTEN_ADDRESS -p $RISEDEV_RW_FRONTEND_PORT -U root -d dev \
---csv --tuples-only -c ${query}
+--csv --tuples-only --quiet -c ${query}
 `
   )
     .toString()
diff --git a/e2e_test/queryable_internal_state/group_agg.slt b/e2e_test/queryable_internal_state/group_agg.slt
new file mode 100644
index 0000000000000..6dd1ed2b338e0
--- /dev/null
+++ b/e2e_test/queryable_internal_state/group_agg.slt
@@ -0,0 +1,56 @@
+# See https://github.com/risingwavelabs/risingwave/pull/20435 for the bug.
+
+statement ok
+SET RW_IMPLICIT_FLUSH TO true;
+
+statement ok
+create table t (foo int, bar int);
+
+statement ok
+insert into t values (1, 2), (1, 5);
+
+statement ok
+create materialized view mv as select foo, sum(bar) as sum from t group by foo;
+
+query ok
+select * from mv;
+----
+1 7
+
+system ok
+internal_table.mjs --name mv --type hashaggstate --count
+----
+count: 1
+
+
+statement ok
+insert into t values (2, 3), (2, 4);
+
+query ok
+select * from mv order by foo;
+----
+1 7
+2 7
+
+system ok
+internal_table.mjs --name mv --type hashaggstate --count
+----
+count: 2
+
+
+statement ok
+delete from t where foo = 1;
+
+query ok
+select * from mv;
+----
+2 7
+
+system ok
+internal_table.mjs --name mv --type hashaggstate --count
+----
+count: 1
+
+
+statement ok
+drop table t cascade;