fix(frontend): reject defining watermarks on upsert source

e2e_test/ddl/table/watermark.slt.part

@@ -1,6 +1,10 @@
-# Create a table with watermark.
+# Create an append-only table with watermark.
 statement ok
 create table t1 (v1 int, v2 timestamp with time zone, watermark for v2 as v2 - INTERVAL '10 MINUTES') append only;
 
+# Creating a normal table with watermark should fail.
+statement error Defining watermarks on table requires the table to be append only
+create table t2 (v1 int, v2 timestamp with time zone, watermark for v2 as v2 - INTERVAL '10 MINUTES');
+
 statement ok
 drop table t1;

e2e_test/source_inline/kafka/upsert_source.slt

@@ -23,6 +23,19 @@ WITH (
  topic = 'test_include_key')
 FORMAT UPSERT ENCODE JSON
 
+statement error Defining watermarks on source requires the source connector to be append only
+CREATE SOURCE upsert_students_with_watermark (
+    primary key (rw_key),
+    ts TIMESTAMP,
+    WATERMARK FOR ts AS ts - INTERVAL '5' SECOND
+)
+INCLUDE KEY AS rw_key
+WITH (
+ ${RISEDEV_KAFKA_WITH_OPTIONS_COMMON},
+ topic = 'test_include_key')
+FORMAT UPSERT ENCODE JSON
+
+
 query TT
 SHOW CREATE SOURCE upsert_students
 ----

src/frontend/src/handler/create_source.rs

@@ -1015,6 +1015,16 @@ HINT: use `CREATE SOURCE <name> WITH (...)` instead of `CREATE SOURCE <name> (<c
     // TODO(yuhao): allow multiple watermark on source.
     assert!(watermark_descs.len() <= 1);
 
+    let append_only = row_id_index.is_some();
+    if is_create_source && !append_only && !watermark_descs.is_empty() {
+        return Err(ErrorCode::NotSupported(
+            "Defining watermarks on source requires the source connector to be append only."
+                .to_owned(),
+            "Use the key words `FORMAT PLAIN`".to_owned(),
+        )
+        .into());
+    }
+
     bind_sql_column_constraints(
         session,
         source_name.clone(),
@@ -1039,7 +1049,7 @@ HINT: use `CREATE SOURCE <name> WITH (...)` instead of `CREATE SOURCE <name> (<c
         database_id,
         columns,
         pk_col_ids,
-        append_only: row_id_index.is_some(),
+        append_only,
         owner: session.user_id(),
         info: source_info,
         row_id_index,

src/frontend/src/optimizer/plan_node/stream_watermark_filter.rs

@@ -36,6 +36,12 @@ pub struct StreamWatermarkFilter {
 
 impl StreamWatermarkFilter {
     pub fn new(input: PlanRef, watermark_descs: Vec<WatermarkDesc>) -> Self {
+        assert!(
+            input.append_only(),
+            "StreamWatermarkFilter only supports append-only input, got {}",
+            input.stream_kind()
+        );
+
         let ctx = input.ctx();
         let mut watermark_columns = input.watermark_columns().clone();
         for i in &watermark_descs {