Add a feature flag for PARTITION BY

Author: bkirwi

src/sql/src/plan/statement/ddl.rs

@@ -160,8 +160,8 @@ use crate::plan::{
     WebhookHeaders, WebhookValidation,
 };
 use crate::session::vars::{
-    self, ENABLE_CLUSTER_SCHEDULE_REFRESH, ENABLE_KAFKA_SINK_HEADERS,
-    ENABLE_KAFKA_SINK_PARTITION_BY, ENABLE_REFRESH_EVERY_MVS,
+    self, ENABLE_CLUSTER_SCHEDULE_REFRESH, ENABLE_COLLECTION_PARTITION_BY,
+    ENABLE_KAFKA_SINK_HEADERS, ENABLE_KAFKA_SINK_PARTITION_BY, ENABLE_REFRESH_EVERY_MVS,
 };
 use crate::{names, parse};
 
@@ -2607,6 +2607,7 @@ pub fn plan_create_materialized_view(
     }: MaterializedViewOptionExtracted = stmt.with_options.try_into()?;
 
     if let Some(partition_by) = partition_by {
+        scx.require_feature_flag(&ENABLE_COLLECTION_PARTITION_BY)?;
         let partition_by: Vec<_> = partition_by
             .into_iter()
             .map(normalize::column_name)

src/sql/src/session/vars/definitions.rs

@@ -1866,6 +1866,12 @@ feature_flags!(
         default: false,
         enable_for_item_parsing: true,
     },
+    {
+        name: enable_collection_partition_by,
+        desc: "PARTITION BY",
+        default: false,
+        enable_for_item_parsing: true,
+    },
     {
         name: enable_multi_worker_storage_persist_sink,
         desc: "multi-worker storage persist sink",