feat: Revert "feat(sink): support format encode syntax for sink into table (#14134)

e2e_test/sink/sink_into_table.slt

@@ -55,7 +55,7 @@ statement error Only append-only sinks can sink to a table without primary keys.
 create sink s1 into t_row_id_as_primary_key as select v1, v2 from t_s1;
 
 statement ok
-create sink s1 into t_row_id_as_primary_key as select v1, v2 from t_s1 FORMAT PLAIN ENCODE NATIVE(force_append_only='true');
+create sink s1 into t_row_id_as_primary_key as select v1, v2 from t_s1 with (type = 'append-only', force_append_only = 'true');
 
 statement ok
 flush;
@@ -154,7 +154,7 @@ statement error Only append-only sinks can sink to a table without primary keys.
 create sink s2 into t_append_only as select v1, v2 from t_s2;
 
 statement ok
-create sink s2 into t_append_only as select v1, v2 from t_s2 FORMAT PLAIN ENCODE NATIVE(force_append_only='true');
+create sink s2 into t_append_only as select v1, v2 from t_s2 with (type = 'append-only', force_append_only = 'true');
 
 statement ok
 flush;

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

@@ -133,7 +133,6 @@ pub enum SinkEncode {
     Protobuf,
     Avro,
     Template,
-    Native,
 }
 
 impl SinkFormatDesc {
@@ -180,7 +179,6 @@ impl SinkFormatDesc {
             SinkEncode::Protobuf => E::Protobuf,
             SinkEncode::Avro => E::Avro,
             SinkEncode::Template => E::Template,
-            SinkEncode::Native => E::Native,
         };
         let options = self
             .options
@@ -218,8 +216,7 @@ impl TryFrom<PbSinkFormatDesc> for SinkFormatDesc {
             E::Protobuf => SinkEncode::Protobuf,
             E::Template => SinkEncode::Template,
             E::Avro => SinkEncode::Avro,
-            E::Native => SinkEncode::Native,
-            e @ (E::Unspecified | E::Csv | E::Bytes) => {
+            e @ (E::Unspecified | E::Native | E::Csv | E::Bytes) => {
                 return Err(SinkError::Config(anyhow!(
                     "sink encode unsupported: {}",
                     e.as_str_name()

src/connector/src/sink/formatter/mod.rs

@@ -150,7 +150,6 @@ impl SinkFormatterImpl {
                             AppendOnlyFormatter::new(Some(key_encoder), val_encoder),
                         ))
                     }
-                    SinkEncode::Native => err_unsupported(),
                 }
             }
             SinkFormat::Debezium => {
@@ -252,7 +251,6 @@ impl SinkFormatterImpl {
                         Ok(SinkFormatterImpl::UpsertAvro(formatter))
                     }
                     SinkEncode::Protobuf => err_unsupported(),
-                    SinkEncode::Native => err_unsupported(),
                 }
             }
         }

src/frontend/src/handler/create_sink.rs

@@ -173,23 +173,13 @@ pub fn gen_sink_plan(
         }
         None => match with_options.get(SINK_TYPE_OPTION) {
             // Case B: old syntax `type = '...'`
-            Some(t) => {
-                if !allow_old_sink_type_syntax(&connector) {
-                    return Err(ErrorCode::BindError(format!(
-                        "connector {} does not support `type = '...'`, please use syntax `FORMAT ... ENCODE ...` instead.",
-                        connector
-                    ))
-                    .into());
-                } else {
-                    SinkFormatDesc::from_legacy_type(&connector, t)?.map(|mut f| {
-                        session.notice_to_user("Consider using the newer syntax `FORMAT ... ENCODE ...` instead of `type = '...'`.");
-                        if let Some(v) = with_options.get(SINK_USER_FORCE_APPEND_ONLY_OPTION) {
-                              f.options.insert(SINK_USER_FORCE_APPEND_ONLY_OPTION.into(), v.into());
-                        }
-                        f
-                    })
+            Some(t) => SinkFormatDesc::from_legacy_type(&connector, t)?.map(|mut f| {
+                session.notice_to_user("Consider using the newer syntax `FORMAT ... ENCODE ...` instead of `type = '...'`.");
+                if let Some(v) = with_options.get(SINK_USER_FORCE_APPEND_ONLY_OPTION) {
+                    f.options.insert(SINK_USER_FORCE_APPEND_ONLY_OPTION.into(), v.into());
                 }
-            }
+                f
+            }),
             // Case C: no format + encode required
             None => None,
         },
@@ -255,7 +245,7 @@ pub fn gen_sink_plan(
             || sink_catalog.sink_type == SinkType::ForceAppendOnly)
         {
             return Err(RwError::from(ErrorCode::BindError(
-                "Only append-only sinks can sink to a table without primary keys. Please try to add \"FORMAT PLAIN ENCODE NATIVE\"".to_string(),
+                "Only append-only sinks can sink to a table without primary keys.".to_string(),
             )));
         }
 
@@ -591,8 +581,7 @@ fn bind_sink_format_desc(value: ConnectorSchema) -> Result<SinkFormatDesc> {
         E::Protobuf => SinkEncode::Protobuf,
         E::Avro => SinkEncode::Avro,
         E::Template => SinkEncode::Template,
-        E::Native => SinkEncode::Native,
-        e @ (E::Csv | E::Bytes) => {
+        e @ (E::Native | E::Csv | E::Bytes) => {
             return Err(ErrorCode::BindError(format!("sink encode unsupported: {e}")).into());
         }
     };
@@ -637,17 +626,9 @@ static CONNECTORS_COMPATIBLE_FORMATS: LazyLock<HashMap<String, HashMap<Format, V
                     Format::Plain => vec![Encode::Json,Encode::Template],
                     Format::Upsert => vec![Encode::Json,Encode::Template],
                 ),
-                "table" => hashmap!(
-                    Format::Plain => vec![Encode::Native],
-                    Format::Upsert => vec![Encode::Native],
-                ),
         ))
     });
 
-pub fn allow_old_sink_type_syntax(connector: &str) -> bool {
-    !matches!(connector, "table")
-}
-
 pub fn validate_compatibility(connector: &str, format_desc: &ConnectorSchema) -> Result<()> {
     let compatible_formats = CONNECTORS_COMPATIBLE_FORMATS
         .get(connector)

src/sqlparser/src/keywords.rs

@@ -601,7 +601,6 @@ pub const RESERVED_FOR_TABLE_ALIAS: &[Keyword] = &[
     Keyword::SET,
     Keyword::RETURNING,
     Keyword::EMIT,
-    Keyword::FORMAT,
 ];
 
 /// Can't be used as a column alias, so that `SELECT <expr> alias`