Repository: risingwavelabs/risingwave
File: proto/stream_plan.proto (Protocol Buffers)

Code before:
  // Consistent with the semantics in AddMutation.
  map<uint32, Dispatchers> actor_new_dispatchers = 6;
  // CDC table snapshot splits
  source.CdcTableSnapshotSplitsWithGeneration actor_cdc_table_snapshot_splits = 7;
  map<uint32, SinkAddColumns> sink_add_columns = 8;
}

message SourceChangeSplitMutation {
  // `Source` and `SourceBackfill` are handled together here.
  map<uint32, source.ConnectorSplits> actor_splits = 2;
}

message PauseMutation {}

message ResumeMutation {}

message ThrottleMutation {
  // Throttle type to indicate which executors should apply the rate limit.
  // Keep the enum values aligned with meta.ThrottleType for semantic consistency.
  enum ThrottleType {
    THROTTLE_TYPE_UNSPECIFIED = 0;
    THROTTLE_TYPE_DML = 1;
    THROTTLE_TYPE_BACKFILL = 2;
    THROTTLE_TYPE_SOURCE = 3;
    THROTTLE_TYPE_SINK = 4;
  }
  message RateLimit {
    optional uint32 rate_limit = 1;
  }

  // Actor -> RateLimit
  map<uint32, RateLimit> actor_throttle = 1;
  // Which kind of executors should honor this rate limit.
  ThrottleType throttle_type = 2;
}

message SubscriptionUpstreamInfo {
  // can either be subscription_id or table_id of creating TableFragments
  uint32 subscriber_id = 1;
  uint32 upstream_mv_table_id = 2;
}

message DropSubscriptionsMutation {
  repeated SubscriptionUpstreamInfo info = 1;
}

message ConnectorPropsChangeMutation {
  message ConnectorPropsInfo {
    map<string, string> connector_props_info = 1;
  }
  map<uint32, ConnectorPropsInfo> connector_props_infos = 1;

Review comment (refactor):
The `ThrottleType` enum is duplicated in both `meta.proto` and as a nested enum in `stream_plan.proto`. While the comment indicates they should be kept aligned, this duplication creates maintainability risk. Consider using a single shared enum definition or importing the enum from `meta.proto` to avoid potential misalignment between the two definitions.

Code after:
  map<uint32, SinkSchemaChange> sink_schema_change = 8;
  repeated SubscriptionUpstreamInfo subscriptions_to_drop = 9;
}

message SourceChangeSplitMutation {
  // `Source` and `SourceBackfill` are handled together here.
  map<uint32, source.ConnectorSplits> actor_splits = 2;
}

message PauseMutation {}

message ResumeMutation {}

message ThrottleMutation {
  message ThrottleConfig {
    optional uint32 rate_limit = 1;
    common.ThrottleType throttle_type = 2;
  }

  map<uint32, ThrottleConfig> fragment_throttle = 1;
}

message SubscriptionUpstreamInfo {
  // can either be subscription_id or table_id of creating TableFragments
  uint32 subscriber_id = 1;
  uint32 upstream_mv_table_id = 2;
}

message DropSubscriptionsMutation {
  repeated SubscriptionUpstreamInfo info = 1;
}

message ConnectorPropsChangeMutation {
  message ConnectorPropsInfo {
    map<string, string> connector_props_info = 1;
  }
  map<uint32, ConnectorPropsInfo> connector_props_infos = 1;
}

message StartFragmentBackfillMutation {
  repeated uint32 fragment_ids = 1;
}

message RefreshStartMutation {
  // Table ID to start refresh operation.
  uint32 table_id = 1;
  // Associated source ID for this refresh operation.
  uint32 associated_source_id = 2;
}
message ListFinishMutation {
  // Associated source ID for this list operation.