feat: Connection for connector usage

Cargo.toml

@@ -153,7 +153,11 @@ arrow-udf-flight = "0.4"
 clap = { version = "4", features = ["cargo", "derive", "env"] }
 # Use a forked version which removes the dependencies on dynamo db to reduce
 # compile time and binary size.
-deltalake = { version = "0.20.1", features = ["s3", "gcs", "datafusion"] }
+deltalake = { version = "0.20.1", features = [
+    "s3",
+    "gcs",
+    "datafusion",
+] }
 itertools = "0.13.0"
 jsonbb = "0.1.4"
 lru = { git = "https://github.com/risingwavelabs/lru-rs.git", rev = "2682b85" }

proto/catalog.proto

@@ -92,6 +92,9 @@ message StreamSourceInfo {
   // Handle the source relies on any sceret. The key is the propertity name and the value is the secret id and type.
   // For format and encode options.
   map<string, secret.SecretRef> format_encode_secret_refs = 16;
+
+  // ref connection for schema registry
+  optional uint32 connection_id = 17;
 }
 
 message Source {
@@ -123,6 +126,8 @@ message Source {
     uint32 associated_table_id = 12;
   }
   string definition = 13;
+
+  // ref connection for connector
   optional uint32 connection_id = 14;
 
   optional uint64 initialized_at_epoch = 15;
@@ -156,6 +161,9 @@ message SinkFormatDesc {
   optional plan_common.EncodeType key_encode = 4;
   // Secret used for format encode options.
   map<string, secret.SecretRef> secret_refs = 5;
+
+  // ref connection for schema registry
+  optional uint32 connection_id = 6;
 }
 
 // the catalog of the sink. There are two kind of schema here. The full schema is all columns
@@ -178,6 +186,8 @@ message Sink {
   uint32 owner = 11;
   map<string, string> properties = 12;
   string definition = 13;
+
+  // ref connection for connector
   optional uint32 connection_id = 14;
   optional uint64 initialized_at_epoch = 15;
   optional uint64 created_at_epoch = 16;
@@ -226,6 +236,19 @@ message Subscription {
   SubscriptionState subscription_state = 19;
 }
 
+message ConnectionParams {
+  enum ConnectionType {
+    CONNECTION_TYPE_UNSPECIFIED = 0;
+    CONNECTION_TYPE_KAFKA = 1;
+    CONNECTION_TYPE_ICEBERG = 2;
+    CONNECTION_TYPE_SCHEMA_REGISTRY = 3;
+  }
+
+  ConnectionType connection_type = 1;
+  map<string, string> properties = 2;
+  map<string, secret.SecretRef> secret_refs = 3;
+}
+
 message Connection {
   message PrivateLinkService {
     enum PrivateLinkProvider {
@@ -246,6 +269,7 @@ message Connection {
   string name = 4;
   oneof info {
     PrivateLinkService private_link_service = 5 [deprecated = true];
+    ConnectionParams connection_params = 7;
   }
   uint32 owner = 6;
 }

proto/ddl_service.proto

@@ -424,7 +424,8 @@ message CreateConnectionRequest {
   uint32 database_id = 2;
   uint32 schema_id = 3;
   oneof payload {
-    PrivateLink private_link = 4;
+    PrivateLink private_link = 4 [deprecated = true];
+    catalog.ConnectionParams connection_params = 6;
   }
   uint32 owner_id = 5;
 }

src/connector/src/connector_common/mod.rs

@@ -26,3 +26,16 @@ pub use common::{
 
 mod iceberg;
 pub use iceberg::IcebergCommon;
+
+// #[derive(Debug, Clone, Deserialize)]
+// pub enum ConnectionImpl {
+//     Kafka(KafkaConnection),
+// }
+
+// macro_rules! impl_connection_enum {
+//     ($impl:expr, $inner_name:ident, $prop_type_name:ident, $body:expr) => {
+//         impl ConnectionImpl {
+//             pub fn get_
+//         }
+//     };
+// }

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

@@ -221,6 +221,7 @@ impl SinkFormatDesc {
             options,
             key_encode,
             secret_refs: self.secret_refs.clone(),
+            connection_id: None,
         }
     }
 }

src/ctl/src/cmd_impl/meta/connection.rs

@@ -35,6 +35,13 @@ pub async fn list_connections(context: &CtlContext) -> anyhow::Result<()> {
                     "PrivateLink: service_name: {}, endpoint_id: {}, dns_entries: {:?}",
                     svc.service_name, svc.endpoint_id, svc.dns_entries,
                 ),
+                Some(Info::ConnectionParams(params)) => {
+                    format!(
+                        "CONNECTION_PARAMS_{}: {}",
+                        params.get_connection_type().unwrap().as_str_name(),
+                        serde_json::to_string(&params.get_properties()).unwrap()
+                    )
+                }
                 None => "None".to_string(),
             }
         );

src/frontend/src/catalog/connection_catalog.rs

@@ -30,12 +30,14 @@ impl ConnectionCatalog {
     pub fn connection_type(&self) -> &str {
         match &self.info {
             Info::PrivateLinkService(srv) => srv.get_provider().unwrap().as_str_name(),
+            Info::ConnectionParams(params) => params.get_connection_type().unwrap().as_str_name(),
         }
     }
 
     pub fn provider(&self) -> &str {
         match &self.info {
             Info::PrivateLinkService(_) => "PRIVATELINK",
+            Info::ConnectionParams(_) => panic!("ConnectionParams is not supported as provider."),
         }
     }
 }

src/frontend/src/catalog/system_catalog/rw_catalog/rw_connections.rs

@@ -28,23 +28,39 @@ struct RwConnection {
     type_: String,
     provider: String,
     acl: String,
+    connection_params: String,
 }
 
 #[system_catalog(table, "rw_catalog.rw_connections")]
 fn read_rw_connections(reader: &SysCatalogReaderImpl) -> Result<Vec<RwConnection>> {
     let catalog_reader = reader.catalog_reader.read_guard();
     let schemas = catalog_reader.iter_schemas(&reader.auth_context.database)?;
 
+    // todo: redesign the internal table for connection params
     Ok(schemas
         .flat_map(|schema| {
-            schema.iter_connections().map(|conn| RwConnection {
-                id: conn.id as i32,
-                name: conn.name.clone(),
-                schema_id: schema.id() as i32,
-                owner: conn.owner as i32,
-                type_: conn.connection_type().into(),
-                provider: conn.provider().into(),
-                acl: "".into(),
+            schema.iter_connections().map(|conn| {
+                let mut rw_connection = RwConnection {
+                    id: conn.id as i32,
+                    name: conn.name.clone(),
+                    schema_id: schema.id() as i32,
+                    owner: conn.owner as i32,
+                    type_: conn.connection_type().into(),
+                    provider: "".to_string(),
+                    acl: "".into(),
+                    connection_params: "".to_string(),
+                };
+                match &conn.info {
+                    risingwave_pb::catalog::connection::Info::PrivateLinkService(_) => {
+                        rw_connection.provider = conn.provider().into();
+                    }
+                    risingwave_pb::catalog::connection::Info::ConnectionParams(params) => {
+                        rw_connection.connection_params =
+                            serde_json::to_string(&params.get_properties()).unwrap();
+                    }
+                };
+
+                rw_connection
             })
         })
         .collect())

src/frontend/src/handler/create_connection.rs

@@ -15,7 +15,10 @@
 use std::collections::BTreeMap;
 
 use pgwire::pg_response::{PgResponse, StatementType};
-use risingwave_connector::source::kafka::PRIVATELINK_CONNECTION;
+use risingwave_connector::source::iceberg::ICEBERG_CONNECTOR;
+use risingwave_connector::source::kafka::{KAFKA_CONNECTOR, PRIVATELINK_CONNECTION};
+use risingwave_pb::catalog::connection_params::ConnectionType;
+use risingwave_pb::catalog::ConnectionParams;
 use risingwave_pb::ddl_service::create_connection_request;
 use risingwave_sqlparser::ast::CreateConnectionStatement;
 
@@ -24,36 +27,58 @@ use crate::binder::Binder;
 use crate::error::ErrorCode::ProtocolError;
 use crate::error::{ErrorCode, Result, RwError};
 use crate::handler::HandlerArgs;
+use crate::session::SessionImpl;
+use crate::utils::resolve_secret_ref_in_with_options;
+use crate::WithOptions;
 
 pub(crate) const CONNECTION_TYPE_PROP: &str = "type";
 
 #[inline(always)]
 fn get_connection_property_required(
-    with_properties: &BTreeMap<String, String>,
+    with_properties: &mut BTreeMap<String, String>,
     property: &str,
 ) -> Result<String> {
-    with_properties
-        .get(property)
-        .map(|s| s.to_lowercase())
-        .ok_or_else(|| {
-            RwError::from(ProtocolError(format!(
-                "Required property \"{property}\" is not provided"
-            )))
-        })
+    with_properties.remove(property).ok_or_else(|| {
+        RwError::from(ProtocolError(format!(
+            "Required property \"{property}\" is not provided"
+        )))
+    })
 }
 fn resolve_create_connection_payload(
-    with_properties: &BTreeMap<String, String>,
+    with_properties: WithOptions,
+    session: &SessionImpl,
 ) -> Result<create_connection_request::Payload> {
-    let connection_type = get_connection_property_required(with_properties, CONNECTION_TYPE_PROP)?;
-    match connection_type.as_str() {
-        PRIVATELINK_CONNECTION => Err(RwError::from(ErrorCode::Deprecated(
+    if !with_properties.connection_ref().is_empty() {
+        return Err(RwError::from(ErrorCode::InvalidParameterValue(
+            "Connection reference is not allowed in options in CREATE CONNECTION".to_string(),
+        )));
+    }
+
+    let (mut props, secret_refs) =
+        resolve_secret_ref_in_with_options(with_properties, session)?.into_parts();
+    let connection_type = get_connection_property_required(&mut props, CONNECTION_TYPE_PROP)?;
+    let connection_type = match connection_type.as_str() {
+        PRIVATELINK_CONNECTION => {
+            return Err(RwError::from(ErrorCode::Deprecated(
             "CREATE CONNECTION to Private Link".to_string(),
             "RisingWave Cloud Portal (Please refer to the doc https://docs.risingwave.com/cloud/create-a-connection/)".to_string(),
-        ))),
-        _ => Err(RwError::from(ProtocolError(format!(
-            "Connection type \"{connection_type}\" is not supported"
-        )))),
-    }
+        )));
+        }
+        KAFKA_CONNECTOR => ConnectionType::Kafka,
+        ICEBERG_CONNECTOR => ConnectionType::Iceberg,
+        _ => {
+            return Err(RwError::from(ProtocolError(format!(
+                "Connection type \"{connection_type}\" is not supported"
+            ))));
+        }
+    };
+    Ok(create_connection_request::Payload::ConnectionParams(
+        ConnectionParams {
+            connection_type: connection_type as i32,
+            properties: props.into_iter().collect(),
+            secret_refs: secret_refs.into_iter().collect(),
+        },
+    ))
 }
 
 pub async fn handle_create_connection(
@@ -79,8 +104,7 @@ pub async fn handle_create_connection(
     }
     let (database_id, schema_id) = session.get_database_and_schema_id_for_create(schema_name)?;
     let with_properties = handler_args.with_options.clone().into_connector_props();
-
-    let create_connection_payload = resolve_create_connection_payload(&with_properties)?;
+    let create_connection_payload = resolve_create_connection_payload(with_properties, &session)?;
 
     let catalog_writer = session.catalog_writer()?;
     catalog_writer