feat: create secret catalog

proto/catalog.proto

@@ -62,10 +62,12 @@ message StreamSourceInfo {
   SchemaRegistryNameStrategy name_strategy = 10;
   optional string key_message_name = 11;
   plan_common.ExternalTableDesc external_table = 12;
-  // **This field should now be called `is_shared`.** Not renamed for backwards compatibility.
+  // **This field should now be called `is_shared`.** Not renamed for backwards
+  // compatibility.
   //
   // Whether the stream source is a shared source (it has a streaming job).
-  // This is related with [RFC: Reusable Source Executor](https://github.com/risingwavelabs/rfcs/pull/72).
+  // This is related with [RFC: Reusable Source
+  // Executor](https://github.com/risingwavelabs/rfcs/pull/72).
   //
   // Currently, the following sources can be shared:
   //
@@ -80,6 +82,9 @@ message StreamSourceInfo {
   bool is_distributed = 15;
   // Options specified by user in the FORMAT ENCODE clause.
   map<string, string> format_encode_options = 14;
+
+  // Handle the source relies on any sceret. The key is the propertity name and the value is the secret id.
+  map<string, uint32> secret_ref = 16;
 }
 
 message Source {
@@ -173,6 +178,9 @@ message Sink {
 
   // Whether it should use background ddl or block until backfill finishes.
   CreateType create_type = 24;
+
+  // Handle the source relies on any sceret. The key is the propertity name and the value is the secret id.
+  map<string, uint32> secret_ref = 25;
 }
 
 message Subscription {
@@ -239,7 +247,8 @@ message Index {
   optional uint64 created_at_epoch = 11;
   StreamJobStatus stream_job_status = 12;
 
-  // Use to record the prefix len of the index_item to reconstruct index columns provided by users.
+  // Use to record the prefix len of the index_item to reconstruct index columns
+  // provided by users.
   uint32 index_columns_len = 13;
   // Cluster version (tracked by git commit) when initialized/created
   optional string initialized_at_cluster_version = 14;
@@ -304,9 +313,7 @@ message Table {
   repeated plan_common.ColumnCatalog columns = 5;
   repeated common.ColumnOrder pk = 6;
   repeated uint32 dependent_relations = 8;
-  oneof optional_associated_source_id {
-    uint32 associated_source_id = 9;
-  }
+  oneof optional_associated_source_id { uint32 associated_source_id = 9; }
   TableType table_type = 10;
   repeated int32 distribution_key = 12;
   // pk_indices of the corresponding materialize operator's output.
@@ -319,8 +326,8 @@ message Table {
   // an optional column index which is the vnode of each row computed by the
   // table's consistent hash distribution
   optional uint32 vnode_col_index = 18;
-  // An optional column index of row id. If the primary key is specified by users,
-  // this will be `None`.
+  // An optional column index of row id. If the primary key is specified by
+  // users, this will be `None`.
   optional uint32 row_id_index = 19;
   // The column indices which are stored in the state store's value with
   // row-encoding. Currently is not supported yet and expected to be
@@ -329,23 +336,26 @@ message Table {
   string definition = 21;
   // Used to control whether handling pk conflict for incoming data.
   HandleConflictBehavior handle_pk_conflict_behavior = 22;
-  // Anticipated read prefix pattern (number of fields) for the table, which can be utilized
-  // for implementing the table's bloom filter or other storage optimization techniques.
+  // Anticipated read prefix pattern (number of fields) for the table, which can
+  // be utilized for implementing the table's bloom filter or other storage
+  // optimization techniques.
   uint32 read_prefix_len_hint = 23;
   repeated int32 watermark_indices = 24;
   repeated int32 dist_key_in_pk = 25;
-  // A dml fragment id corresponds to the table, used to decide where the dml statement is executed.
+  // A dml fragment id corresponds to the table, used to decide where the dml
+  // statement is executed.
   optional uint32 dml_fragment_id = 26;
   // The range of row count of the table.
-  // This field is not always present due to backward compatibility. Use `Cardinality::unknown` in this case.
+  // This field is not always present due to backward compatibility. Use
+  // `Cardinality::unknown` in this case.
   plan_common.Cardinality cardinality = 27;
 
   optional uint64 initialized_at_epoch = 28;
   optional uint64 created_at_epoch = 29;
 
-  // This field is introduced in v1.2.0. It is used to indicate whether the table should use
-  // watermark_cache to avoid state cleaning as a performance optimization.
-  // In older versions we can just initialize without it.
+  // This field is introduced in v1.2.0. It is used to indicate whether the
+  // table should use watermark_cache to avoid state cleaning as a performance
+  // optimization. In older versions we can just initialize without it.
   bool cleaned_by_watermark = 30;
 
   // Used to filter created / creating tables in meta.
@@ -363,14 +373,18 @@ message Table {
   optional string initialized_at_cluster_version = 35;
   optional string created_at_cluster_version = 36;
 
-  // TTL of the record in the table, to ensure the consistency with other tables in the streaming plan, it only applies to append-only tables.
+  // TTL of the record in the table, to ensure the consistency with other tables
+  // in the streaming plan, it only applies to append-only tables.
   optional uint32 retention_seconds = 37;
 
-  // This field specifies the index of the column set in the "with version column" within all the columns. It is used for filtering during "on conflict" operations.
+  // This field specifies the index of the column set in the "with version
+  // column" within all the columns. It is used for filtering during "on
+  // conflict" operations.
   optional uint32 version_column_index = 38;
 
-  // Per-table catalog version, used by schema change. `None` for internal tables and tests.
-  // Not to be confused with the global catalog version for notification service.
+  // Per-table catalog version, used by schema change. `None` for internal
+  // tables and tests. Not to be confused with the global catalog version for
+  // notification service.
   TableVersion version = 100;
 }
 
@@ -415,3 +429,13 @@ message Comment {
   optional uint32 column_index = 4;
   optional string description = 5;
 }
+
+message Secret {
+  uint32 id = 1;
+  string name = 2;
+  uint32 database_id = 3;
+  // The secret here is encrypted to bytes.
+  bytes value = 4;
+  uint32 owner = 5;
+  uint32 schema_id = 6;
+}

proto/ddl_service.proto

@@ -358,6 +358,26 @@ message GetDdlProgressResponse {
   repeated DdlProgress ddl_progress = 1;
 }
 
+message CreateSecretRequest {
+  string name = 1;
+  bytes value = 2;
+  uint32 database_id = 3;
+  uint32 schema_id = 4;
+  uint32 owner_id = 5;
+}
+
+message CreateSecretResponse {
+  uint64 version = 1;
+}
+
+message DropSecretRequest {
+  uint32 secret_id = 1;
+}
+
+message DropSecretResponse {
+  uint64 version = 1;
+}
+
 message CreateConnectionRequest {
   message PrivateLink {
     catalog.Connection.PrivateLinkService.PrivateLinkProvider provider = 1;
@@ -428,6 +448,8 @@ service DdlService {
   rpc CreateMaterializedView(CreateMaterializedViewRequest) returns (CreateMaterializedViewResponse);
   rpc DropMaterializedView(DropMaterializedViewRequest) returns (DropMaterializedViewResponse);
   rpc CreateTable(CreateTableRequest) returns (CreateTableResponse);
+  rpc CreateSecret(CreateSecretRequest) returns (CreateSecretResponse);
+  rpc DropSecret(DropSecretRequest) returns (DropSecretResponse);
   rpc AlterName(AlterNameRequest) returns (AlterNameResponse);
   rpc AlterSource(AlterSourceRequest) returns (AlterSourceResponse);
   rpc AlterOwner(AlterOwnerRequest) returns (AlterOwnerResponse);

proto/meta.proto

@@ -406,6 +406,8 @@ message MetaSnapshot {
   reserved 9;
   reserved "parallel_unit_mappings";
   GetSessionParamsResponse session_params = 20;
+  repeated catalog.Secret secrets = 23;
+  // for streaming
   repeated common.WorkerNode nodes = 10;
   hummock.HummockSnapshot hummock_snapshot = 11;
   hummock.HummockVersion hummock_version = 12;
@@ -468,6 +470,7 @@ message SubscribeResponse {
     Recovery recovery = 25;
     FragmentWorkerMapping streaming_worker_mapping = 27;
     FragmentWorkerMappings serving_worker_mappings = 28;
+    catalog.Secret secret = 29;
   }
   reserved 12;
   reserved "parallel_unit_mapping";

src/common/common_service/src/observer_manager.rs

@@ -27,27 +27,31 @@ pub trait SubscribeTypeEnum {
 }
 
 pub struct SubscribeFrontend {}
+
 impl SubscribeTypeEnum for SubscribeFrontend {
     fn subscribe_type() -> SubscribeType {
         SubscribeType::Frontend
     }
 }
 
 pub struct SubscribeHummock {}
+
 impl SubscribeTypeEnum for SubscribeHummock {
     fn subscribe_type() -> SubscribeType {
         SubscribeType::Hummock
     }
 }
 
 pub struct SubscribeCompactor {}
+
 impl SubscribeTypeEnum for SubscribeCompactor {
     fn subscribe_type() -> SubscribeType {
         SubscribeType::Compactor
     }
 }
 
 pub struct SubscribeCompute {}
+
 impl SubscribeTypeEnum for SubscribeCompute {
     fn subscribe_type() -> SubscribeType {
         SubscribeType::Compute
@@ -142,6 +146,7 @@ where
             | Info::RelationGroup(_)
             | Info::User(_)
             | Info::Connection(_)
+            | Info::Secret(_)
             | Info::Function(_) => {
                 notification.version > info.version.as_ref().unwrap().catalog_version
             }
@@ -231,6 +236,7 @@ where
         }
     }
 }
+
 const RE_SUBSCRIBE_RETRY_INTERVAL: Duration = Duration::from_millis(100);
 
 #[async_trait::async_trait]

src/common/src/catalog/mod.rs

@@ -437,6 +437,41 @@ impl From<ConnectionId> for u32 {
     }
 }
 
+#[derive(Clone, Copy, Debug, Display, Default, Hash, PartialOrd, PartialEq, Eq, Ord)]
+pub struct SecretId(pub u32);
+
+impl SecretId {
+    pub const fn new(id: u32) -> Self {
+        SecretId(id)
+    }
+
+    pub const fn placeholder() -> Self {
+        SecretId(OBJECT_ID_PLACEHOLDER)
+    }
+
+    pub fn secret_id(&self) -> u32 {
+        self.0
+    }
+}
+
+impl From<u32> for SecretId {
+    fn from(id: u32) -> Self {
+        Self::new(id)
+    }
+}
+
+impl From<&u32> for SecretId {
+    fn from(id: &u32) -> Self {
+        Self::new(*id)
+    }
+}
+
+impl From<SecretId> for u32 {
+    fn from(id: SecretId) -> Self {
+        id.0
+    }
+}
+
 #[derive(Default, Debug, Clone, Copy, PartialEq, Eq, Hash)]
 pub enum ConflictBehavior {
     #[default]

src/connector/src/sink/catalog/desc.rs

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::collections::BTreeMap;
+use std::collections::{BTreeMap, HashMap};
 
 use itertools::Itertools;
 use risingwave_common::catalog::{
@@ -83,6 +83,7 @@ impl SinkDesc {
         owner: UserId,
         connection_id: Option<ConnectionId>,
         dependent_relations: Vec<TableId>,
+        secret_ref: HashMap<String, u32>,
     ) -> SinkCatalog {
         SinkCatalog {
             id: self.id,
@@ -108,6 +109,7 @@ impl SinkDesc {
             created_at_cluster_version: None,
             initialized_at_cluster_version: None,
             create_type: self.create_type,
+            secret_ref,
         }
     }
 

src/connector/src/sink/catalog/mod.rs

@@ -310,6 +310,9 @@ pub struct SinkCatalog {
     pub created_at_cluster_version: Option<String>,
     pub initialized_at_cluster_version: Option<String>,
     pub create_type: CreateType,
+
+    /// The secret reference for the sink, mapping from property name to secret id.
+    pub secret_ref: HashMap<String, u32>,
 }
 
 impl SinkCatalog {
@@ -351,6 +354,7 @@ impl SinkCatalog {
             created_at_cluster_version: self.created_at_cluster_version.clone(),
             initialized_at_cluster_version: self.initialized_at_cluster_version.clone(),
             create_type: self.create_type.to_proto() as i32,
+            secret_ref: self.secret_ref.clone(),
         }
     }
 
@@ -444,6 +448,7 @@ impl From<PbSink> for SinkCatalog {
             initialized_at_cluster_version: pb.initialized_at_cluster_version,
             created_at_cluster_version: pb.created_at_cluster_version,
             create_type: CreateType::from_proto(create_type),
+            secret_ref: pb.secret_ref,
         }
     }
 }