feat: create secret catalog

e2e_test/ddl/secret.slt

@@ -0,0 +1,23 @@
+statement error secret backend "fake-backend" is not supported
+create secret secret_1 with (
+  backend = 'fake-backend'
+) as 'demo_secret';
+
+statement ok
+create secret secret_1 with (
+  backend = 'meta'
+) as 'demo_secret';
+
+# wait for support for hashicorp_vault backend
+# statement ok
+# create secret secret_2 with (
+# backend = 'hashicorp_vault'
+# );
+
+query T
+show secrets;
+----
+secret_1
+
+statement ok
+drop secret secret_1;

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 {
@@ -174,6 +179,9 @@ message Sink {
 
   // Whether it should use background ddl or block until backfill finishes.
   CreateType create_type = 24;
+
+  // Handle the sink 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;
@@ -319,8 +328,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 +338,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 +375,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 +431,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

@@ -357,6 +357,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;
@@ -427,6 +447,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

@@ -407,6 +407,7 @@ message MetaSnapshot {
   reserved 9;
   reserved "parallel_unit_mappings";
   GetSessionParamsResponse session_params = 20;
+  repeated catalog.Secret secrets = 23;
   repeated common.WorkerNode nodes = 10;
   hummock.HummockSnapshot hummock_snapshot = 11;
   hummock.HummockVersion hummock_version = 12;
@@ -469,6 +470,7 @@ message SubscribeResponse {
     Recovery recovery = 25;
     FragmentWorkerSlotMapping streaming_worker_slot_mapping = 27;
     FragmentWorkerSlotMappings serving_worker_slot_mappings = 28;
+    catalog.Secret secret = 29;
   }
   reserved 12;
   reserved "parallel_unit_mapping";

proto/secret.proto

@@ -0,0 +1,20 @@
+syntax = "proto3";
+
+package secret;
+
+message SecretMetaBackend {
+  bytes value = 1;
+}
+
+message SecretHashicropValutBackend {
+  string host = 1;
+  string vault_token = 2;
+}
+
+message Secret {
+  // the message is stored in meta as encrypted bytes and is interpreted as bytes by catalog
+  oneof secret_backend {
+    SecretMetaBackend meta = 1;
+    SecretHashicropValutBackend hashicorp_vault = 2;
+  }
+}

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

@@ -451,6 +451,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]