fix: ban pk comprising impure generated column

e2e_test/ddl/table/generated_columns.slt.part

@@ -149,3 +149,11 @@ CREATE TABLE t (v INT, t timestamptz as now()) WITH (
   datagen.rows.per.second='15',
   datagen.split.num = '1'
 ) FORMAT PLAIN ENCODE JSON;
+
+# create a table with impure generated column as pk.
+statement error QueryError: Bind error: Generated columns should not be part of the primary key. Here column "v2" is defined as part of the primary key.
+CREATE TABLE t (
+  v1 INT,
+  v2 timestamptz AS proctime(),
+  PRIMARY KEY (v1, v2)
+);

risedev.yml

@@ -20,14 +20,14 @@ profile:
     # config-path: src/config/example.toml
     steps:
       # If you want to use the local s3 storage, enable the following line
-      - use: minio
+      # - use: minio
 
       # If you want to use aws-s3, configure AK and SK in env var and enable the following lines:
       # - use: aws-s3
       #   bucket: test-bucket
 
       # If you want to create CDC source table, uncomment the following line
-      - use: connector-node
+      # - use: connector-node
 
       # if you want to enable etcd backend, uncomment the following lines.
       # - use: etcd
@@ -43,7 +43,7 @@ profile:
       - use: frontend
 
       # If you want to enable compactor, uncomment the following line, and enable either minio or aws-s3 as well.
-      - use: compactor
+      # - use: compactor
 
       # If you want to create source from Kafka, uncomment the following lines
       # Note that kafka depends on zookeeper, so zookeeper must be started beforehand.

src/frontend/src/handler/create_source.rs

@@ -1107,7 +1107,13 @@ pub async fn handle_create_source(
     // TODO(yuhao): allow multiple watermark on source.
     assert!(watermark_descs.len() <= 1);
 
-    bind_sql_column_constraints(&session, name.clone(), &mut columns, stmt.columns)?;
+    bind_sql_column_constraints(
+        &session,
+        name.clone(),
+        &mut columns,
+        stmt.columns,
+        &pk_column_ids,
+    )?;
 
     check_source_schema(&with_properties, row_id_index, &columns)?;
 

src/frontend/src/handler/create_table.rs

@@ -197,9 +197,11 @@ pub fn bind_sql_columns(column_defs: &[ColumnDef]) -> Result<Vec<ColumnCatalog>>
 
 fn check_generated_column_constraints(
     column_name: &String,
+    column_id: ColumnId,
     expr: &ExprImpl,
     column_catalogs: &[ColumnCatalog],
     generated_column_names: &[String],
+    pk_column_ids: &[ColumnId],
 ) -> Result<()> {
     let input_refs = expr.collect_input_refs(column_catalogs.len());
     for idx in input_refs.ones() {
@@ -214,6 +216,14 @@ fn check_generated_column_constraints(
             .into());
         }
     }
+
+    if pk_column_ids.contains(&column_id) && expr.is_impure() {
+        return Err(ErrorCode::BindError(
+            format!("Generated columns should not be part of the primary key. Here column \"{}\" is defined as part of the primary key.", column_name),
+        )
+        .into());
+    }
+
     Ok(())
 }
 
@@ -243,6 +253,7 @@ pub fn bind_sql_column_constraints(
     table_name: String,
     column_catalogs: &mut [ColumnCatalog],
     columns: Vec<ColumnDef>,
+    pk_column_ids: &[ColumnId],
 ) -> Result<()> {
     let generated_column_names = {
         let mut names = vec![];
@@ -271,9 +282,11 @@ pub fn bind_sql_column_constraints(
 
                     check_generated_column_constraints(
                         &column.name.real_value(),
+                        column_catalogs[idx].column_id(),
                         &expr_impl,
                         column_catalogs,
                         &generated_column_names,
+                        pk_column_ids,
                     )?;
 
                     column_catalogs[idx].column_desc.generated_or_default_column = Some(
@@ -460,7 +473,13 @@ pub(crate) async fn gen_create_table_plan_with_source(
 
     let definition = context.normalized_sql().to_owned();
 
-    bind_sql_column_constraints(session, table_name.real_value(), &mut columns, column_defs)?;
+    bind_sql_column_constraints(
+        session,
+        table_name.real_value(),
+        &mut columns,
+        column_defs,
+        &pk_column_ids,
+    )?;
 
     check_source_schema(&properties, row_id_index, &columns)?;
 
@@ -592,6 +611,7 @@ pub(crate) fn gen_create_table_plan_without_bind(
         table_name.real_value(),
         &mut columns,
         column_defs,
+        &pk_column_ids,
     )?;
 
     gen_table_plan_inner(