fix: cannot create a materialized view with struct column

e2e_test/streaming/struct_table.slt

@@ -0,0 +1,26 @@
+statement ok
+SET RW_IMPLICIT_FLUSH TO true;
+
+statement ok
+create table st (v1 int, v2 struct<v1 int, v2 struct<v1 int, v2 int>>);
+
+statement ok
+create materialized view mv1 as select (v2).v2 from st;
+
+statement ok
+insert into st values(1,(1,(1,2)));
+
+statement ok
+insert into st values(1,(1,(1,3)));
+
+query I
+select * from mv1;
+----
+(1,2)
+(1,3)
+
+statement ok
+drop materialized view mv1
+
+statement ok
+drop table st

src/common/src/catalog/column.rs

@@ -170,19 +170,10 @@ impl ColumnDesc {
         }
     }
 
-    /// Generate incremental `column_id` for `column_desc` and `field_descs`
-    pub fn generate_increment_id(&mut self, index: &mut i32) {
-        self.column_id = ColumnId::new(*index);
-        *index += 1;
-        for field in &mut self.field_descs {
-            field.generate_increment_id(index);
-        }
-    }
-
-    pub fn from_field_without_column_id(field: &Field) -> Self {
+    pub fn from_field_with_column_id(field: &Field, id: i32) -> Self {
         Self {
             data_type: field.data_type.clone(),
-            column_id: ColumnId::new(0),
+            column_id: ColumnId::new(id),
             name: field.name.clone(),
             field_descs: field
                 .sub_fields
@@ -192,6 +183,10 @@ impl ColumnDesc {
             type_name: field.type_name.clone(),
         }
     }
+
+    pub fn from_field_without_column_id(field: &Field) -> Self {
+        Self::from_field_with_column_id(field, 0)
+    }
 }
 
 impl From<ProstColumnDesc> for ColumnDesc {

src/frontend/src/catalog/column_catalog.rs

@@ -62,14 +62,6 @@ impl ColumnCatalog {
             is_hidden: true,
         }
     }
-
-    /// Generate incremental `column_id` for every `column_desc` and `column_desc.field_descs`
-    pub fn generate_increment_id(catalogs: &mut Vec<ColumnCatalog>) {
-        let mut index = 0;
-        for catalog in catalogs {
-            catalog.column_desc.generate_increment_id(&mut index);
-        }
-    }
 }
 
 impl From<ProstColumnCatalog> for ColumnCatalog {

src/frontend/src/optimizer/plan_node/stream_materialize.rs

@@ -105,13 +105,13 @@ impl StreamMaterialize {
             }
         }
         let mut out_name_iter = out_names.into_iter();
-        let mut columns = schema
+        let columns = schema
             .fields()
             .iter()
             .enumerate()
             .map(|(i, field)| {
                 let mut c = ColumnCatalog {
-                    column_desc: ColumnDesc::from_field_without_column_id(field),
+                    column_desc: ColumnDesc::from_field_with_column_id(field, i as i32),
                     is_hidden: !user_cols.contains(i),
                 };
                 c.column_desc.name = if !c.is_hidden {
@@ -130,10 +130,6 @@ impl StreamMaterialize {
             })
             .collect_vec();
 
-        // Since the `field.into()` only generate same ColumnId,
-        // so rewrite ColumnId for each `column_desc` and `column_desc.field_desc`.
-        ColumnCatalog::generate_increment_id(&mut columns);
-
         let mut in_order = FixedBitSet::with_capacity(schema.len());
         let mut order_desc = vec![];