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 {
@@ -238,7 +246,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;
@@ -299,9 +308,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.
@@ -314,8 +321,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
@@ -324,23 +331,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.
@@ -358,14 +368,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;
 }
 
@@ -410,3 +424,12 @@ message Comment {
   optional uint32 column_index = 4;
   optional string description = 5;
 }
+
+message Secret {
+  uint32 id = 1;
+  string name = 2;
+  uint32 database_id = 3;
+  bytes value = 4;
+  uint32 owner = 5;
+
+}

src/common/src/catalog/mod.rs

@@ -436,6 +436,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,
         }
     }
 }

src/frontend/src/catalog/mod.rs

@@ -55,6 +55,7 @@ pub(crate) type SchemaId = u32;
 pub(crate) type TableId = risingwave_common::catalog::TableId;
 pub(crate) type ColumnId = risingwave_common::catalog::ColumnId;
 pub(crate) type FragmentId = u32;
+pub(crate) type SecretId = risingwave_common::catalog::SecretId;
 
 /// Check if the column name does not conflict with the internally reserved column name.
 pub fn check_valid_column_name(column_name: &str) -> Result<()> {

src/frontend/src/catalog/schema_catalog.rs

@@ -34,7 +34,7 @@ use crate::catalog::source_catalog::SourceCatalog;
 use crate::catalog::system_catalog::SystemTableCatalog;
 use crate::catalog::table_catalog::TableCatalog;
 use crate::catalog::view_catalog::ViewCatalog;
-use crate::catalog::{ConnectionId, DatabaseId, SchemaId, SinkId, SourceId, ViewId};
+use crate::catalog::{ConnectionId, DatabaseId, SchemaId, SecretId, SinkId, SourceId, ViewId};
 use crate::expr::{infer_type_name, infer_type_with_sigmap, Expr, ExprImpl};
 use crate::user::UserId;
 
@@ -483,6 +483,10 @@ impl SchemaCatalog {
             .expect("connection not found by name");
     }
 
+    pub fn create_secret(&mut self, secret_id: SecretId) {
+        todo!()
+    }
+
     pub fn iter_all(&self) -> impl Iterator<Item = &Arc<TableCatalog>> {
         self.table_by_name.values()
     }

src/frontend/src/handler/create_sink.rs

@@ -63,7 +63,7 @@ use crate::optimizer::{OptimizerContext, OptimizerContextRef, PlanRef, RelationC
 use crate::scheduler::streaming_manager::CreatingStreamingJobInfo;
 use crate::session::SessionImpl;
 use crate::stream_fragmenter::build_graph;
-use crate::utils::resolve_privatelink_in_with_option;
+use crate::utils::{resolve_privatelink_in_with_option, resolve_secret_in_with_options};
 use crate::{Explain, Planner, TableCatalog, WithOptions};
 
 pub fn gen_sink_subscription_query_from_name(from_name: ObjectName) -> Result<Query> {
@@ -166,6 +166,7 @@ pub fn gen_sink_plan(
             resolve_privatelink_in_with_option(&mut with_options, &sink_schema_name, session)?;
         conn_id.map(ConnectionId)
     };
+    let secret_ref = resolve_secret_in_with_options(&mut with_options)?;
 
     let emit_on_window_close = stmt.emit_mode == Some(EmitMode::OnWindowClose);
     if emit_on_window_close {
@@ -256,6 +257,7 @@ pub fn gen_sink_plan(
         UserId::new(session.user_id()),
         connection_id,
         dependent_relations.into_iter().collect_vec(),
+        secret_ref,
     );
 
     if let Some(table_catalog) = &target_table_catalog {

src/frontend/src/utils/with_options.rs

@@ -112,6 +112,12 @@ impl WithOptions {
     }
 }
 
+pub(crate) fn resolve_secret_in_with_options(
+    with_options: &mut WithOptions,
+) -> RwResult<HashMap<String, u32>> {
+    todo!()
+}
+
 pub(crate) fn resolve_privatelink_in_with_option(
     with_options: &mut WithOptions,
     schema_name: &Option<String>,

src/meta/model_v2/src/lib.rs

@@ -71,6 +71,7 @@ pub type IndexId = ObjectId;
 pub type ViewId = ObjectId;
 pub type FunctionId = ObjectId;
 pub type ConnectionId = ObjectId;
+pub type SecretId = ObjectId;
 pub type UserId = i32;
 pub type PrivilegeId = i32;
 

src/meta/model_v2/src/secret.rs

@@ -0,0 +1,60 @@
+// Copyright 2024 RisingWave Labs
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use sea_orm::entity::prelude::*;
+use sea_orm::ActiveValue::Set;
+
+#[derive(Clone, Debug, PartialEq, DeriveEntityModel, Eq)]
+#[sea_orm(table_name = "secret")]
+pub struct Model {
+    pub secret_id: u32,
+    pub name: String,
+}
+
+#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]
+pub enum Relation {
+    #[sea_orm(
+        belongs_to = "super::object::Entity",
+        from = "Column::SecretId",
+        to = "super::object::Column::Oid",
+        on_update = "NoAction",
+        on_delete = "Cascade"
+    )]
+    Object,
+    // #[sea_orm(
+    //     belongs_to = "super"
+    // )]
+    // Sink,
+    // #[sea_orm(has_many = "super::source::Entity")]
+    // Source,
+}
+
+impl Related<super::object::Entity> for Entity {
+    fn to() -> RelationDef {
+        Relation::Object.def()
+    }
+}
+
+// impl Related<super::sink::Entity> for Entity {
+//     fn to() -> RelationDef {
+//         Relation::Sink.def()
+//     }
+// }
+
+// impl Related<super::source::Entity> for Entity {
+//     fn to() -> RelationDef {
+//         Relation::Source.def()
+//     }
+// }
+