Repository: risingwavelabs/risingwave (language: python)

Issue:
feat(system table): add `properties map(varchar,varchar)` in rw_sources and rw_sinks
### Is your feature request related to a problem? Please describe.

Required for the source page of the cloud portal: https://linear.app/risingwave-labs/issue/CLOUD-3318/%5Bfeature-request%5D-source-details-page

### Describe the solution you'd like

_No response_

### Describe alternatives you've considered

_No response_

### Additional context

_No response_
Sqlsmith: Sql feature generation
As we add more features, these should be tested via sqlsmith as well.

SQL Features:
- [ ] Generate other table functions: `generate_series`, `unnest`. See: https://github.com/singularity-data/risingwave/issues/4540
- [ ] Generate array_agg, jsonb_agg, jsonb_object_agg
- [ ] Generate select from System Tables
- [ ] Generate `NATURAL JOIN`
- [ ] Generate `JOIN` with `USING`
- [ ] Generate query w/o `GROUP BY`
- [ ] Randomly toggle session variables
- [ ] Generate `pow`/`exp`.
- [ ] Generate TopN by group: https://www.risingwave.dev/docs/current/sql-pattern-topn/
- [ ] Generate jsonb exprs, see: https://github.com/risingwavelabs/risingwave/issues/7714
- [ ] #7132
- [x] Generate order-by-clauses with `NULLS { FIRST | LAST }` (will be done in https://github.com/risingwavelabs/risingwave/pull/10864/commits/000cbc11cbc04f20abfd127b1bf4b3fd24e75c44)
- [ ] Generate `max`,`min`,`count` for `timestamptz`
- [ ] Generate `bit_and`, `bit_or` and `big_xor` in batch queries, and `bit_xor` in streaming queries.
- [ ] Generate array range access.
- [ ] https://github.com/risingwavelabs/risingwave/issues/10241
- [ ] Generate window functions other than `row_number`, `rank`.
- [ ] Generate `ALTER` DDL statements.\
- [ ] https://github.com/risingwavelabs/risingwave/issues/10516
- [ ] Generate temporal filter in sqlsmith


Discussion / hints:
In addition to the properties map of WITH options, we also need a field for encoding options, e.g, 

```sql
ENCODE AVRO (
  aws.glue.schema_arn = 'arn:aws:glue:ap-southeast-1:123456123456:schema/default-registry/MyEvent',
  aws.region = 'US-WEST-2',
  aws.credentials.access_key_id = 'your_access_key_id',
  aws.credentials.secret_access_key = 'your_secret_access_key',
  aws.credentials.role.arn = 'arn:aws:iam::123456123456:role/MyGlueRole'
  ...
)
```
@xxchan Can you please help with this
dissussed with @neverchanje and @xuhui-lu

* for v2.1 and before, the cloud team will parse SQL and extract the props
* after v2.2, the the cloud relies on rw_catalog for resolved props (after resolving connection catalog)
  * had an agreement on value protocol
  * for normal value, it should be `{"some key": {"type": "plaintext", "value": "xxxx"}}`
  * for secret, it should be `{"some key": {"type": "secret", "value": "<secret name>"}}`


Fix patch:
diff --git a/src/frontend/src/catalog/source_catalog.rs b/src/frontend/src/catalog/source_catalog.rs
index f453a8920450..86cefe03c149 100644
--- a/src/frontend/src/catalog/source_catalog.rs
+++ b/src/frontend/src/catalog/source_catalog.rs
@@ -105,7 +105,7 @@ impl From<&PbSource> for SourceCatalog {
             .into_iter()
             .map(Into::into)
             .collect();
-        let options_with_secrets =
+        let connector_props_with_secrets =
             WithOptionsSecResolved::new(prost.with_properties.clone(), prost.secret_refs.clone());
         let columns = prost_columns.into_iter().map(ColumnCatalog::from).collect();
         let row_id_index = prost.row_id_index.map(|idx| idx as _);
@@ -131,7 +131,7 @@ impl From<&PbSource> for SourceCatalog {
             owner,
             info: prost.info.clone().unwrap(),
             row_id_index,
-            with_properties: options_with_secrets,
+            with_properties: connector_props_with_secrets,
             watermark_descs,
             associated_table_id: associated_table_id.map(|x| x.into()),
             definition: prost.definition.clone(),
diff --git a/src/frontend/src/catalog/system_catalog/rw_catalog/rw_sinks.rs b/src/frontend/src/catalog/system_catalog/rw_catalog/rw_sinks.rs
index e382a5b7dfaf..609850b9951a 100644
--- a/src/frontend/src/catalog/system_catalog/rw_catalog/rw_sinks.rs
+++ b/src/frontend/src/catalog/system_catalog/rw_catalog/rw_sinks.rs
@@ -12,10 +12,12 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use risingwave_common::types::{Fields, Timestamptz};
+use risingwave_common::types::{Fields, JsonbVal, Timestamptz};
+use risingwave_connector::WithOptionsSecResolved;
 use risingwave_frontend_macro::system_catalog;
 use risingwave_pb::user::grant_privilege::Object;
 
+use crate::catalog::system_catalog::rw_catalog::rw_sources::serialize_props_with_secret;
 use crate::catalog::system_catalog::{get_acl_items, SysCatalogReaderImpl};
 use crate::error::Result;
 use crate::handler::create_source::UPSTREAM_SOURCE_KEY;
@@ -36,6 +38,11 @@ struct RwSink {
     created_at: Option<Timestamptz>,
     initialized_at_cluster_version: Option<String>,
     created_at_cluster_version: Option<String>,
+
+    // connector properties in json format
+    connector_props: JsonbVal,
+    // format and encode properties in json format
+    format_encode_options: JsonbVal,
 }
 
 #[system_catalog(table, "rw_catalog.rw_sinks")]
@@ -48,30 +55,53 @@ fn read_rw_sinks_info(reader: &SysCatalogReaderImpl) -> Result<Vec<RwSink>> {
 
     Ok(schemas
         .flat_map(|schema| {
-            schema.iter_sink().map(|sink| RwSink {
-                id: sink.id.sink_id as i32,
-                name: sink.name.clone(),
-                schema_id: schema.id() as i32,
-                owner: sink.owner.user_id as i32,
-                connector: sink
-                    .properties
-                    .get(UPSTREAM_SOURCE_KEY)
-                    .cloned()
-                    .unwrap_or("".to_string())
-                    .to_uppercase(),
-                sink_type: sink.sink_type.to_proto().as_str_name().into(),
-                connection_id: sink.connection_id.map(|id| id.connection_id() as i32),
-                definition: sink.create_sql(),
-                acl: get_acl_items(
-                    &Object::SinkId(sink.id.sink_id),
-                    false,
-                    &users,
-                    username_map,
-                ),
-                initialized_at: sink.initialized_at_epoch.map(|e| e.as_timestamptz()),
-                created_at: sink.created_at_epoch.map(|e| e.as_timestamptz()),
-                initialized_at_cluster_version: sink.initialized_at_cluster_version.clone(),
-                created_at_cluster_version: sink.created_at_cluster_version.clone(),
+            schema.iter_sink().map(|sink| {
+                let connector_props = serialize_props_with_secret(
+                    schema,
+                    WithOptionsSecResolved::new(sink.properties.clone(), sink.secret_refs.clone()),
+                )
+                .into();
+                let format_encode_options = sink
+                    .format_desc
+                    .as_ref()
+                    .map(|desc| {
+                        serialize_props_with_secret(
+                            schema,
+                            WithOptionsSecResolved::new(
+                                desc.options.clone(),
+                                desc.secret_refs.clone(),
+                            ),
+                        )
+                    })
+                    .unwrap_or_else(jsonbb::Value::null)
+                    .into();
+                RwSink {
+                    id: sink.id.sink_id as i32,
+                    name: sink.name.clone(),
+                    schema_id: schema.id() as i32,
+                    owner: sink.owner.user_id as i32,
+                    connector: sink
+                        .properties
+                        .get(UPSTREAM_SOURCE_KEY)
+                        .cloned()
+                        .unwrap_or("".to_string())
+                        .to_uppercase(),
+                    sink_type: sink.sink_type.to_proto().as_str_name().into(),
+                    connection_id: sink.connection_id.map(|id| id.connection_id() as i32),
+                    definition: sink.create_sql(),
+                    acl: get_acl_items(
+                        &Object::SinkId(sink.id.sink_id),
+                        false,
+                        &users,
+                        username_map,
+                    ),
+                    initialized_at: sink.initialized_at_epoch.map(|e| e.as_timestamptz()),
+                    created_at: sink.created_at_epoch.map(|e| e.as_timestamptz()),
+                    initialized_at_cluster_version: sink.initialized_at_cluster_version.clone(),
+                    created_at_cluster_version: sink.created_at_cluster_version.clone(),
+                    connector_props,
+                    format_encode_options,
+                }
             })
         })
         .collect())
diff --git a/src/frontend/src/catalog/system_catalog/rw_catalog/rw_sources.rs b/src/frontend/src/catalog/system_catalog/rw_catalog/rw_sources.rs
index bdcfe355057a..303ce645fbd0 100644
--- a/src/frontend/src/catalog/system_catalog/rw_catalog/rw_sources.rs
+++ b/src/frontend/src/catalog/system_catalog/rw_catalog/rw_sources.rs
@@ -12,13 +12,17 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use risingwave_common::types::{Fields, Timestamptz};
+use risingwave_common::catalog::SecretId;
+use risingwave_common::types::{Fields, JsonbVal, Timestamptz};
 use risingwave_frontend_macro::system_catalog;
 use risingwave_pb::user::grant_privilege::Object;
+use serde_json::{json, Map as JsonMap};
 
+use crate::catalog::schema_catalog::SchemaCatalog;
 use crate::catalog::system_catalog::{get_acl_items, SysCatalogReaderImpl};
 use crate::error::Result;
 use crate::handler::create_source::UPSTREAM_SOURCE_KEY;
+use crate::WithOptionsSecResolved;
 
 #[derive(Fields)]
 struct RwSource {
@@ -41,6 +45,10 @@ struct RwSource {
     initialized_at_cluster_version: Option<String>,
     created_at_cluster_version: Option<String>,
     is_shared: bool,
+    // connector properties in json format
+    connector_props: JsonbVal,
+    // format-encode options in json format
+    format_encode_options: JsonbVal,
 }
 
 #[system_catalog(table, "rw_catalog.rw_sources")]
@@ -53,39 +61,83 @@ fn read_rw_sources_info(reader: &SysCatalogReaderImpl) -> Result<Vec<RwSource>>
 
     Ok(schemas
         .flat_map(|schema| {
-            schema.iter_source().map(|source| RwSource {
-                id: source.id as i32,
-                name: source.name.clone(),
-                schema_id: schema.id() as i32,
-                owner: source.owner as i32,
-                connector: source
-                    .with_properties
-                    .get(UPSTREAM_SOURCE_KEY)
-                    .cloned()
-                    .unwrap_or("".to_string())
-                    .to_uppercase(),
-                columns: source.columns.iter().map(|c| c.name().into()).collect(),
-                format: source
-                    .info
-                    .get_format()
-                    .ok()
-                    .map(|format| format.as_str_name().into()),
-                row_encode: source
-                    .info
-                    .get_row_encode()
-                    .ok()
-                    .map(|row_encode| row_encode.as_str_name().into()),
-                append_only: source.append_only,
-                associated_table_id: source.associated_table_id.map(|id| id.table_id as i32),
-                connection_id: source.connection_id.map(|id| id as i32),
-                definition: source.create_sql(),
-                acl: get_acl_items(&Object::SourceId(source.id), false, &users, username_map),
-                initialized_at: source.initialized_at_epoch.map(|e| e.as_timestamptz()),
-                created_at: source.created_at_epoch.map(|e| e.as_timestamptz()),
-                initialized_at_cluster_version: source.initialized_at_cluster_version.clone(),
-                created_at_cluster_version: source.created_at_cluster_version.clone(),
-                is_shared: source.info.is_shared(),
+            schema.iter_source().map(|source| {
+                let format_encode_props_with_secrets = WithOptionsSecResolved::new(
+                    source.info.format_encode_options.clone(),
+                    source.info.format_encode_secret_refs.clone(),
+                );
+                RwSource {
+                    id: source.id as i32,
+                    name: source.name.clone(),
+                    schema_id: schema.id() as i32,
+                    owner: source.owner as i32,
+                    connector: source
+                        .with_properties
+                        .get(UPSTREAM_SOURCE_KEY)
+                        .cloned()
+                        .unwrap_or("".to_string())
+                        .to_uppercase(),
+                    columns: source.columns.iter().map(|c| c.name().into()).collect(),
+                    format: source
+                        .info
+                        .get_format()
+                        .ok()
+                        .map(|format| format.as_str_name().into()),
+                    row_encode: source
+                        .info
+                        .get_row_encode()
+                        .ok()
+                        .map(|row_encode| row_encode.as_str_name().into()),
+                    append_only: source.append_only,
+                    associated_table_id: source.associated_table_id.map(|id| id.table_id as i32),
+                    connection_id: source.connection_id.map(|id| id as i32),
+                    definition: source.create_sql(),
+                    acl: get_acl_items(&Object::SourceId(source.id), false, &users, username_map),
+                    initialized_at: source.initialized_at_epoch.map(|e| e.as_timestamptz()),
+                    created_at: source.created_at_epoch.map(|e| e.as_timestamptz()),
+                    initialized_at_cluster_version: source.initialized_at_cluster_version.clone(),
+                    created_at_cluster_version: source.created_at_cluster_version.clone(),
+                    is_shared: source.info.is_shared(),
+
+                    connector_props: serialize_props_with_secret(
+                        schema,
+                        source.with_properties.clone(),
+                    )
+                    .into(),
+                    format_encode_options: serialize_props_with_secret(
+                        schema,
+                        format_encode_props_with_secrets,
+                    )
+                    .into(),
+                }
             })
         })
         .collect())
 }
+
+pub fn serialize_props_with_secret(
+    schema: &SchemaCatalog,
+    props_with_secret: WithOptionsSecResolved,
+) -> jsonbb::Value {
+    let (inner, secret_ref) = props_with_secret.into_parts();
+    // if not secret, {"some key": {"type": "plaintext", "value": "xxxx"}}
+    // if secret, {"some key": {"type": "secret", "value": {"value": "<secret name>"}}}
+    let mut result: JsonMap<String, serde_json::Value> = JsonMap::new();
+
+    for (k, v) in inner {
+        result.insert(k, json!({"type": "plaintext", "value": v}));
+    }
+    for (k, v) in secret_ref {
+        let secret_name = schema
+            .get_secret_by_id(&SecretId(v.secret_id))
+            .unwrap()
+            .name
+            .clone();
+        result.insert(
+            k,
+            json!({"type": "secret", "value": {"value": secret_name}}),
+        );
+    }
+
+    jsonbb::Value::from(serde_json::Value::Object(result))
+}


Test patch:
diff --git a/src/frontend/planner_test/tests/testdata/output/subquery.yaml b/src/frontend/planner_test/tests/testdata/output/subquery.yaml
index 5ccdb0050179..a87b116a658c 100644
--- a/src/frontend/planner_test/tests/testdata/output/subquery.yaml
+++ b/src/frontend/planner_test/tests/testdata/output/subquery.yaml
@@ -241,11 +241,11 @@
         │       │ │ │ │ │ │ └─LogicalProject { exprs: [rw_system_tables.id, rw_system_tables.name, 'system table':Varchar, rw_system_tables.schema_id, rw_system_tables.owner, rw_system_tables.definition, rw_system_tables.acl] }
         │       │ │ │ │ │ │   └─LogicalSysScan { table: rw_system_tables, columns: [rw_system_tables.id, rw_system_tables.name, rw_system_tables.schema_id, rw_system_tables.owner, rw_system_tables.definition, rw_system_tables.acl] }
         │       │ │ │ │ │ └─LogicalProject { exprs: [rw_sources.id, rw_sources.name, 'source':Varchar, rw_sources.schema_id, rw_sources.owner, rw_sources.definition, rw_sources.acl] }
-        │       │ │ │ │ │   └─LogicalSysScan { table: rw_sources, columns: [rw_sources.id, rw_sources.name, rw_sources.schema_id, rw_sources.owner, rw_sources.connector, rw_sources.columns, rw_sources.format, rw_sources.row_encode, rw_sources.append_only, rw_sources.associated_table_id, rw_sources.connection_id, rw_sources.definition, rw_sources.acl, rw_sources.initialized_at, rw_sources.created_at, rw_sources.initialized_at_cluster_version, rw_sources.created_at_cluster_version, rw_sources.is_shared] }
+        │       │ │ │ │ │   └─LogicalSysScan { table: rw_sources, columns: [rw_sources.id, rw_sources.name, rw_sources.schema_id, rw_sources.owner, rw_sources.connector, rw_sources.columns, rw_sources.format, rw_sources.row_encode, rw_sources.append_only, rw_sources.associated_table_id, rw_sources.connection_id, rw_sources.definition, rw_sources.acl, rw_sources.initialized_at, rw_sources.created_at, rw_sources.initialized_at_cluster_version, rw_sources.created_at_cluster_version, rw_sources.is_shared, rw_sources.connector_props, rw_sources.format_encode_options] }
         │       │ │ │ │ └─LogicalProject { exprs: [rw_indexes.id, rw_indexes.name, 'index':Varchar, rw_indexes.schema_id, rw_indexes.owner, rw_indexes.definition, rw_indexes.acl] }
         │       │ │ │ │   └─LogicalSysScan { table: rw_indexes, columns: [rw_indexes.id, rw_indexes.name, rw_indexes.primary_table_id, rw_indexes.key_columns, rw_indexes.include_columns, rw_indexes.schema_id, rw_indexes.owner, rw_indexes.definition, rw_indexes.acl, rw_indexes.initialized_at, rw_indexes.created_at, rw_indexes.initialized_at_cluster_version, rw_indexes.created_at_cluster_version] }
         │       │ │ │ └─LogicalProject { exprs: [rw_sinks.id, rw_sinks.name, 'sink':Varchar, rw_sinks.schema_id, rw_sinks.owner, rw_sinks.definition, rw_sinks.acl] }
-        │       │ │ │   └─LogicalSysScan { table: rw_sinks, columns: [rw_sinks.id, rw_sinks.name, rw_sinks.schema_id, rw_sinks.owner, rw_sinks.connector, rw_sinks.sink_type, rw_sinks.connection_id, rw_sinks.definition, rw_sinks.acl, rw_sinks.initialized_at, rw_sinks.created_at, rw_sinks.initialized_at_cluster_version, rw_sinks.created_at_cluster_version] }
+        │       │ │ │   └─LogicalSysScan { table: rw_sinks, columns: [rw_sinks.id, rw_sinks.name, rw_sinks.schema_id, rw_sinks.owner, rw_sinks.connector, rw_sinks.sink_type, rw_sinks.connection_id, rw_sinks.definition, rw_sinks.acl, rw_sinks.initialized_at, rw_sinks.created_at, rw_sinks.initialized_at_cluster_version, rw_sinks.created_at_cluster_version, rw_sinks.connector_props, rw_sinks.format_encode_options] }
         │       │ │ └─LogicalProject { exprs: [rw_subscriptions.id, rw_subscriptions.name, 'subscription':Varchar, rw_subscriptions.schema_id, rw_subscriptions.owner, rw_subscriptions.definition, rw_subscriptions.acl] }
         │       │ │   └─LogicalSysScan { table: rw_subscriptions, columns: [rw_subscriptions.id, rw_subscriptions.name, rw_subscriptions.schema_id, rw_subscriptions.owner, rw_subscriptions.definition, rw_subscriptions.acl, rw_subscriptions.initialized_at, rw_subscriptions.created_at, rw_subscriptions.initialized_at_cluster_version, rw_subscriptions.created_at_cluster_version] }
         │       │ └─LogicalProject { exprs: [rw_materialized_views.id, rw_materialized_views.name, 'materialized view':Varchar, rw_materialized_views.schema_id, rw_materialized_views.owner, rw_materialized_views.definition, rw_materialized_views.acl] }