feat(meta): support inject barrier to enable and disable writing old value for subscription

proto/stream_plan.proto

@@ -94,6 +94,16 @@ message CombinedMutation {
   repeated BarrierMutation mutations = 1;
 }
 
+message CreateSubscriptionMutation {
+  uint32 subscription_id = 1;
+  uint32 upstream_mv_table_id = 2;
+}
+
+message DropSubscriptionMutation {
+  uint32 subscription_id = 1;
+  uint32 upstream_mv_table_id = 2;
+}
+
 message BarrierMutation {
   oneof mutation {
     // Add new dispatchers to some actors, used for creating materialized views.
@@ -111,6 +121,10 @@ message BarrierMutation {
     ResumeMutation resume = 8;
     // Throttle specific source exec or chain exec.
     ThrottleMutation throttle = 10;
+    // Create subscription on mv
+    CreateSubscriptionMutation create_subscription = 11;
+    // Drop subscription on mv
+    DropSubscriptionMutation drop_subscription = 12;
     // Combined mutation.
     CombinedMutation combined = 100;
   }

proto/stream_service.proto

@@ -25,8 +25,12 @@ message BroadcastActorInfoTableRequest {
 
 // Create channels and gRPC connections for a fragment
 message BuildActorsRequest {
+  message SubscriptionIds {
+    repeated uint32 subscription_ids = 1;
+  }
   string request_id = 1;
   repeated uint32 actor_id = 2;
+  map<uint32, SubscriptionIds> related_subscriptions = 3;
 }
 
 message BuildActorsResponse {

src/compute/src/rpc/service/stream_service.rs

@@ -14,6 +14,7 @@
 
 use await_tree::InstrumentAwait;
 use futures::{Stream, StreamExt, TryStreamExt};
+use risingwave_common::catalog::TableId;
 use risingwave_pb::stream_service::stream_service_server::StreamService;
 use risingwave_pb::stream_service::*;
 use risingwave_storage::dispatch_state_store;
@@ -65,7 +66,21 @@ impl StreamService for StreamServiceImpl {
         let req = request.into_inner();
 
         let actor_id = req.actor_id;
-        let res = self.mgr.build_actors(actor_id).await;
+        let res = self
+            .mgr
+            .build_actors(
+                actor_id,
+                req.related_subscriptions
+                    .into_iter()
+                    .map(|(table_id, subscriptions)| {
+                        (
+                            TableId::new(table_id),
+                            subscriptions.subscription_ids.into_iter().collect(),
+                        )
+                    })
+                    .collect(),
+            )
+            .await;
         match res {
             Err(e) => {
                 error!(error = %e.as_report(), "failed to build actors");

src/meta/src/barrier/command.rs

@@ -21,6 +21,8 @@ use itertools::Itertools;
 use risingwave_common::buffer::Bitmap;
 use risingwave_common::catalog::TableId;
 use risingwave_common::hash::ActorMapping;
+use risingwave_common::types::Timestamptz;
+use risingwave_common::util::epoch::Epoch;
 use risingwave_connector::source::SplitImpl;
 use risingwave_hummock_sdk::HummockEpoch;
 use risingwave_pb::catalog::CreateType;
@@ -32,12 +34,13 @@ use risingwave_pb::stream_plan::barrier_mutation::Mutation;
 use risingwave_pb::stream_plan::throttle_mutation::RateLimit;
 use risingwave_pb::stream_plan::update_mutation::*;
 use risingwave_pb::stream_plan::{
-    AddMutation, BarrierMutation, CombinedMutation, Dispatcher, Dispatchers, PauseMutation,
-    ResumeMutation, SourceChangeSplitMutation, StopMutation, StreamActor, ThrottleMutation,
-    UpdateMutation,
+    AddMutation, BarrierMutation, CombinedMutation, CreateSubscriptionMutation, Dispatcher,
+    Dispatchers, DropSubscriptionMutation, PauseMutation, ResumeMutation,
+    SourceChangeSplitMutation, StopMutation, StreamActor, ThrottleMutation, UpdateMutation,
 };
 use risingwave_pb::stream_service::WaitEpochCommitRequest;
 use thiserror_ext::AsReport;
+use tracing::warn;
 
 use super::info::{ActorDesc, CommandActorChanges, InflightActorInfo};
 use super::trace::TracedEpoch;
@@ -201,6 +204,17 @@ pub enum Command {
     /// `Throttle` command generates a `Throttle` barrier with the given throttle config to change
     /// the `rate_limit` of `FlowControl` Executor after `StreamScan` or Source.
     Throttle(ThrottleConfig),
+
+    CreateSubscription {
+        subscription_id: u32,
+        upstream_mv_table_id: TableId,
+        retention_second: u64,
+    },
+
+    DropSubscription {
+        subscription_id: u32,
+        upstream_mv_table_id: TableId,
+    },
 }
 
 impl Command {
@@ -285,6 +299,8 @@ impl Command {
             Command::ReplaceTable(plan) => Some(plan.actor_changes()),
             Command::SourceSplitAssignment(_) => None,
             Command::Throttle(_) => None,
+            Command::CreateSubscription { .. } => None,
+            Command::DropSubscription { .. } => None,
         }
     }
 
@@ -663,6 +679,22 @@ impl CommandContext {
                     tracing::debug!("update mutation: {mutation:?}");
                     Some(mutation)
                 }
+
+                Command::CreateSubscription {
+                    upstream_mv_table_id,
+                    subscription_id,
+                    ..
+                } => Some(Mutation::CreateSubscription(CreateSubscriptionMutation {
+                    upstream_mv_table_id: upstream_mv_table_id.table_id,
+                    subscription_id: *subscription_id,
+                })),
+                Command::DropSubscription {
+                    upstream_mv_table_id,
+                    subscription_id,
+                } => Some(Mutation::DropSubscription(DropSubscriptionMutation {
+                    upstream_mv_table_id: upstream_mv_table_id.table_id,
+                    subscription_id: *subscription_id,
+                })),
             };
 
         mutation
@@ -1071,8 +1103,21 @@ impl CommandContext {
                     )
                     .await;
             }
+
+            Command::CreateSubscription { .. } => {}
+            Command::DropSubscription { .. } => {}
         }
 
         Ok(())
     }
+
+    pub fn get_truncate_epoch(&self, retention_second: u64) -> Epoch {
+        let Some(truncate_timestamptz) = Timestamptz::from_secs(
+            self.prev_epoch.value().as_timestamptz().timestamp() - retention_second as i64,
+        ) else {
+            warn!(retention_second, prev_epoch = ?self.prev_epoch.value(), "invalid retention second value");
+            return self.prev_epoch.value();
+        };
+        Epoch::from_unix_millis(truncate_timestamptz.timestamp_millis() as u64)
+    }
 }

src/meta/src/barrier/info.rs

@@ -14,9 +14,11 @@
 
 use std::collections::{HashMap, HashSet};
 
+use risingwave_common::catalog::TableId;
 use risingwave_pb::common::PbWorkerNode;
 use tracing::warn;
 
+use crate::barrier::Command;
 use crate::manager::{ActiveStreamingWorkerNodes, ActorInfos, WorkerId};
 use crate::model::ActorId;
 
@@ -48,11 +50,18 @@ pub struct InflightActorInfo {
 
     /// `actor_id` => `WorkerId`
     pub actor_location_map: HashMap<ActorId, WorkerId>,
+
+    /// mv_table_id => subscription_id => retention seconds
+    pub mv_depended_subscriptions: HashMap<TableId, HashMap<u32, u64>>,
 }
 
 impl InflightActorInfo {
     /// Resolve inflight actor info from given nodes and actors that are loaded from meta store. It will be used during recovery to rebuild all streaming actors.
-    pub fn resolve(active_nodes: &ActiveStreamingWorkerNodes, actor_infos: ActorInfos) -> Self {
+    pub fn resolve(
+        active_nodes: &ActiveStreamingWorkerNodes,
+        actor_infos: ActorInfos,
+        mv_depended_subscriptions: HashMap<TableId, HashMap<u32, u64>>,
+    ) -> Self {
         let node_map = active_nodes.current().clone();
 
         let actor_map = actor_infos
@@ -77,6 +86,7 @@ impl InflightActorInfo {
             actor_map,
             actor_map_to_send,
             actor_location_map,
+            mv_depended_subscriptions,
         }
     }
 
@@ -96,8 +106,8 @@ impl InflightActorInfo {
 
     /// Apply some actor changes before issuing a barrier command, if the command contains any new added actors, we should update
     /// the info correspondingly.
-    pub fn pre_apply(&mut self, changes: Option<CommandActorChanges>) {
-        if let Some(CommandActorChanges { to_add, .. }) = changes {
+    pub fn pre_apply(&mut self, command: &Command) {
+        if let Some(CommandActorChanges { to_add, .. }) = command.actor_changes() {
             for actor_desc in to_add {
                 assert!(self.node_map.contains_key(&actor_desc.node_id));
                 assert!(
@@ -124,12 +134,27 @@ impl InflightActorInfo {
                 );
             }
         };
+        if let Command::CreateSubscription {
+            subscription_id,
+            upstream_mv_table_id,
+            retention_second,
+        } = command
+        {
+            if let Some(prev_retiontion) = self
+                .mv_depended_subscriptions
+                .entry(*upstream_mv_table_id)
+                .or_default()
+                .insert(*subscription_id, *retention_second)
+            {
+                warn!(subscription_id, ?upstream_mv_table_id, mv_depended_subscriptions = ?self.mv_depended_subscriptions, prev_retiontion, "add an existing subscription id");
+            }
+        }
     }
 
     /// Apply some actor changes after the barrier command is collected, if the command contains any actors that are dropped, we should
     /// remove that from the snapshot correspondingly.
-    pub fn post_apply(&mut self, changes: Option<CommandActorChanges>) {
-        if let Some(CommandActorChanges { to_remove, .. }) = changes {
+    pub fn post_apply(&mut self, command: &Command) {
+        if let Some(CommandActorChanges { to_remove, .. }) = command.actor_changes() {
             for actor_id in to_remove {
                 let node_id = self
                     .actor_location_map
@@ -145,6 +170,25 @@ impl InflightActorInfo {
             self.actor_map_to_send
                 .retain(|_, actor_ids| !actor_ids.is_empty());
         }
+        if let Command::DropSubscription {
+            subscription_id,
+            upstream_mv_table_id,
+        } = command
+        {
+            let removed = match self.mv_depended_subscriptions.get_mut(upstream_mv_table_id) {
+                Some(subscriptions) => {
+                    let removed = subscriptions.remove(subscription_id).is_some();
+                    if removed && subscriptions.is_empty() {
+                        self.mv_depended_subscriptions.remove(upstream_mv_table_id);
+                    }
+                    removed
+                }
+                None => false,
+            };
+            if !removed {
+                warn!(subscription_id, ?upstream_mv_table_id, mv_depended_subscriptions = ?self.mv_depended_subscriptions, "remove a non-existing subscription id");
+            }
+        }
     }
 
     /// Returns actor list to collect in the target worker node.

src/meta/src/barrier/mod.rs

@@ -16,7 +16,6 @@ use std::assert_matches::assert_matches;
 use std::collections::hash_map::Entry;
 use std::collections::{BTreeMap, HashMap, HashSet};
 use std::future::pending;
-use std::iter::empty;
 use std::mem::{replace, take};
 use std::sync::Arc;
 use std::time::Duration;
@@ -1084,16 +1083,20 @@ impl GlobalBarrierManagerContext {
         &self,
         active_nodes: &ActiveStreamingWorkerNodes,
     ) -> MetaResult<InflightActorInfo> {
+        let subscriptions = self
+            .metadata_manager
+            .get_mv_depended_subscriptions()
+            .await?;
         let info = match &self.metadata_manager {
             MetadataManager::V1(mgr) => {
                 let all_actor_infos = mgr.fragment_manager.load_all_actors().await;
 
-                InflightActorInfo::resolve(active_nodes, all_actor_infos)
+                InflightActorInfo::resolve(active_nodes, all_actor_infos, subscriptions)
             }
             MetadataManager::V2(mgr) => {
                 let all_actor_infos = mgr.catalog_controller.load_all_actors().await?;
 
-                InflightActorInfo::resolve(active_nodes, all_actor_infos)
+                InflightActorInfo::resolve(active_nodes, all_actor_infos, subscriptions)
             }
         };
 
@@ -1187,8 +1190,18 @@ fn collect_commit_epoch_info(
         old_value_ssts.into_iter(),
         synced_ssts.iter().map(|sst| &sst.sst_info),
         epochs,
-        // TODO: pass log store table id and the corresponding truncate_epoch
-        empty(),
+        command_ctx
+            .info
+            .mv_depended_subscriptions
+            .iter()
+            .filter_map(|(mv_table_id, subscriptions)| {
+                subscriptions.values().max().map(|max_retention| {
+                    (
+                        mv_table_id.table_id,
+                        command_ctx.get_truncate_epoch(*max_retention).0,
+                    )
+                })
+            }),
     );
 
     CommitEpochInfo::new(

src/meta/src/barrier/recovery.rs

@@ -1171,6 +1171,7 @@ impl GlobalBarrierManagerContext {
                     let actors = actors.iter().cloned().collect();
                     (*node_id, actors)
                 }),
+                &info.mv_depended_subscriptions,
             )
             .await?;
 

src/meta/src/barrier/rpc.rs

@@ -24,10 +24,12 @@ use futures::future::try_join_all;
 use futures::stream::{BoxStream, FuturesUnordered};
 use futures::{pin_mut, FutureExt, StreamExt};
 use itertools::Itertools;
+use risingwave_common::catalog::TableId;
 use risingwave_common::hash::ActorId;
 use risingwave_common::util::tracing::TracingContext;
 use risingwave_pb::common::{ActorInfo, WorkerNode};
 use risingwave_pb::stream_plan::{Barrier, BarrierMutation, StreamActor};
+use risingwave_pb::stream_service::build_actors_request::SubscriptionIds;
 use risingwave_pb::stream_service::{
     streaming_control_stream_request, streaming_control_stream_response, BarrierCompleteResponse,
     BroadcastActorInfoTableRequest, BuildActorsRequest, DropActorsRequest, InjectBarrierRequest,
@@ -402,18 +404,36 @@ impl StreamRpcManager {
         &self,
         node_map: &HashMap<WorkerId, WorkerNode>,
         node_actors: impl Iterator<Item = (WorkerId, Vec<ActorId>)>,
+        related_subscriptions: &HashMap<TableId, HashMap<u32, u64>>,
     ) -> MetaResult<()> {
         self.make_request(
             node_actors.map(|(worker_id, actors)| (node_map.get(&worker_id).unwrap(), actors)),
-            |client, actors| async move {
-                let request_id = Self::new_request_id();
-                tracing::debug!(request_id = request_id.as_str(), actors = ?actors, "build actors");
-                client
-                    .build_actors(BuildActorsRequest {
-                        request_id,
-                        actor_id: actors,
-                    })
-                    .await
+            |client, actors| {
+                let related_subscriptions = related_subscriptions.clone();
+                async move {
+                    let request_id = Self::new_request_id();
+                    tracing::debug!(request_id = request_id.as_str(), actors = ?actors, "build actors");
+                    client
+                        .build_actors(BuildActorsRequest {
+                            request_id,
+                            actor_id: actors,
+                            related_subscriptions: related_subscriptions
+                                .iter()
+                                .map(|(table_id, subscriptions)| {
+                                    (
+                                        table_id.table_id,
+                                        SubscriptionIds {
+                                            subscription_ids: subscriptions
+                                                .keys()
+                                                .cloned()
+                                                .collect(),
+                                        },
+                                    )
+                                })
+                                .collect(),
+                        })
+                        .await
+                }
             },
         )
         .await?;