refactor(frontend): refine handle create table (#7503)

- remove `DMLFlag` which is just the workaround for materialzied source
- return error when user want create a table with pk constrain and append_only property at the same time

Approved-By: BugenZhao

src/frontend/src/handler/create_table.rs

@@ -42,14 +42,6 @@ use crate::optimizer::{OptimizerContext, OptimizerContextRef, PlanRef, PlanRoot}
 use crate::stream_fragmenter::build_graph;
 use crate::{Binder, TableCatalog, WithOptions};
 
-#[derive(PartialEq, Clone, Debug)]
-pub enum DmlFlag {
-    /// used for `create table`
-    All,
-    /// used for `create table with (append_only = true)`
-    AppendOnly,
-}
-
 /// Column ID generator for a new table or a new version of an existing table to alter.
 #[derive(Debug)]
 pub struct ColumnIdGenerator {
@@ -434,20 +426,21 @@ fn gen_table_plan_inner(
         out_names,
     );
 
-    // Handle pk conflict in materialize executor only when the table is not append-only.
-    let (handle_pk_conflict, dml_flag) = if context.with_options().append_only() {
-        (false, DmlFlag::AppendOnly)
-    } else {
-        (true, DmlFlag::All)
-    };
+    let append_only = context.with_options().append_only();
+
+    if append_only && row_id_index.is_none() {
+        return Err(ErrorCode::InvalidInputSyntax(
+            "PRIMARY KEY constraint can not be appiled on a append only table.".to_owned(),
+        )
+        .into());
+    }
 
     let materialize = plan_root.gen_table_plan(
         name,
         columns,
         definition,
-        handle_pk_conflict,
         row_id_index,
-        dml_flag,
+        append_only,
         version,
     )?;
 

src/frontend/src/optimizer/mod.rs

@@ -48,7 +48,6 @@ use self::property::RequiredDist;
 use self::rule::*;
 use crate::catalog::column_catalog::ColumnCatalog;
 use crate::catalog::table_catalog::{TableType, TableVersion};
-use crate::handler::create_table::DmlFlag;
 use crate::optimizer::plan_node::{
     BatchExchange, ColumnPruningContext, PlanNodeType, PlanTreeNode, PredicatePushdownContext,
 };
@@ -615,17 +614,16 @@ impl PlanRoot {
         table_name: String,
         columns: Vec<ColumnCatalog>,
         definition: String,
-        handle_pk_conflict: bool,
         row_id_index: Option<usize>,
-        dml_flag: DmlFlag,
+        append_only: bool,
         version: Option<TableVersion>,
     ) -> Result<StreamMaterialize> {
         let mut stream_plan = self.gen_stream_plan()?;
 
         // Add DML node.
         stream_plan = StreamDml::new(
             stream_plan,
-            dml_flag == DmlFlag::AppendOnly,
+            append_only,
             columns.iter().map(|c| c.column_desc.clone()).collect(),
         )
         .into();
@@ -641,7 +639,7 @@ impl PlanRoot {
             self.required_order.clone(),
             columns,
             definition,
-            handle_pk_conflict,
+            !append_only,
             row_id_index,
             version,
         )