risingwave_frontend/optimizer/plan_node/
stream_materialize.rs

1// Copyright 2022 RisingWave Labs
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::assert_matches;
16use std::num::NonZeroU32;
17
18use fixedbitset::FixedBitSet;
19use itertools::Itertools;
20use pretty_xmlish::{Pretty, XmlNode};
21use risingwave_common::catalog::{
22    ColumnCatalog, ConflictBehavior, CreateType, Engine, StreamJobStatus, TableId,
23};
24use risingwave_common::hash::VnodeCount;
25use risingwave_common::id::FragmentId;
26use risingwave_common::types::DataType;
27use risingwave_common::util::column_index_mapping::ColIndexMapping;
28use risingwave_common::util::iter_util::ZipEqFast;
29use risingwave_common::util::sort_util::{ColumnOrder, OrderType};
30use risingwave_pb::catalog::PbWebhookSourceInfo;
31use risingwave_pb::stream_plan::stream_node::PbNodeBody;
32
33use super::derive::derive_columns;
34use super::stream::prelude::*;
35use super::utils::{Distill, TableCatalogBuilder, childless_record};
36use super::{
37    ExprRewritable, PlanTreeNodeUnary, StreamNode, StreamPlanRef as PlanRef, reorganize_elements_id,
38};
39use crate::catalog::table_catalog::{TableCatalog, TableType, TableVersion};
40use crate::catalog::{DatabaseId, SchemaId};
41use crate::error::Result;
42use crate::optimizer::StreamOptimizedLogicalPlanRoot;
43use crate::optimizer::plan_node::derive::derive_pk;
44use crate::optimizer::plan_node::expr_visitable::ExprVisitable;
45use crate::optimizer::plan_node::utils::plan_can_use_background_ddl;
46use crate::optimizer::plan_node::{PlanBase, PlanNodeMeta};
47use crate::optimizer::property::{Cardinality, Distribution, Order, RequiredDist};
48use crate::optimizer::variant_key::variant_key_error;
49use crate::stream_fragmenter::BuildFragmentGraphState;
50
51/// Materializes a stream.
52#[derive(Debug, Clone, PartialEq, Eq, Hash)]
53pub struct StreamMaterialize {
54    pub base: PlanBase<Stream>,
55    /// Child of Materialize plan
56    input: PlanRef,
57    table: TableCatalog,
58    /// For refreshable tables, staging table for collecting new data during refresh
59    staging_table: Option<TableCatalog>,
60    /// For refreshable tables, progress table for tracking refresh progress
61    refresh_progress_table: Option<TableCatalog>,
62}
63
64impl StreamMaterialize {
65    pub fn new(input: PlanRef, table: TableCatalog) -> Result<Self> {
66        Self::new_with_staging_and_progress(input, table, None, None)
67    }
68
69    pub fn new_with_staging_and_progress(
70        input: PlanRef,
71        table: TableCatalog,
72        staging_table: Option<TableCatalog>,
73        refresh_progress_table: Option<TableCatalog>,
74    ) -> Result<Self> {
75        let kind = match table.conflict_behavior() {
76            ConflictBehavior::NoCheck => {
77                reject_upsert_input!(input, "Materialize without conflict handling")
78            }
79
80            // When conflict handling is enabled, upsert stream can be converted to retract stream.
81            ConflictBehavior::Overwrite
82            | ConflictBehavior::IgnoreConflict
83            | ConflictBehavior::DoUpdateIfNotNull => match input.stream_kind() {
84                StreamKind::AppendOnly => StreamKind::AppendOnly,
85                StreamKind::Retract | StreamKind::Upsert => StreamKind::Retract,
86            },
87        };
88        let base = PlanBase::new_stream(
89            input.ctx(),
90            input.schema().clone(),
91            Some(table.stream_key()),
92            input.functional_dependency().clone(),
93            input.distribution().clone(),
94            kind,
95            input.emit_on_window_close(),
96            input.watermark_columns().clone(),
97            input.columns_monotonicity().clone(),
98        );
99
100        Ok(Self {
101            base,
102            input,
103            table,
104            staging_table,
105            refresh_progress_table,
106        })
107    }
108
109    /// Create a materialize node, for `MATERIALIZED VIEW` and `INDEX`.
110    ///
111    /// When creating index, `TableType` should be `Index`. Then, materialize will distribute keys
112    /// using `user_distributed_by`.
113    pub fn create(
114        StreamOptimizedLogicalPlanRoot {
115            plan: input,
116            required_dist: user_distributed_by,
117            required_order: user_order_by,
118            out_fields: user_cols,
119            out_names,
120            ..
121        }: StreamOptimizedLogicalPlanRoot,
122        name: String,
123        database_id: DatabaseId,
124        schema_id: SchemaId,
125        definition: String,
126        table_type: TableType,
127        cardinality: Cardinality,
128        retention_seconds: Option<NonZeroU32>,
129    ) -> Result<Self> {
130        let input = Self::rewrite_input(input, user_distributed_by.clone(), table_type)?;
131        // the hidden column name might refer some expr id
132        let input = reorganize_elements_id(input);
133        let columns = derive_columns(input.schema(), out_names, &user_cols)?;
134
135        let create_type = if matches!(table_type, TableType::MaterializedView)
136            && input.ctx().session_ctx().config().background_ddl()
137            && plan_can_use_background_ddl(&input)
138        {
139            CreateType::Background
140        } else {
141            CreateType::Foreground
142        };
143
144        // For upsert stream, use `Overwrite` conflict behavior to convert into retract stream.
145        let conflict_behavior = match input.stream_kind() {
146            StreamKind::Retract | StreamKind::AppendOnly => ConflictBehavior::NoCheck,
147            StreamKind::Upsert => ConflictBehavior::Overwrite,
148        };
149
150        let table = Self::derive_table_catalog(
151            input.clone(),
152            name,
153            database_id,
154            schema_id,
155            user_distributed_by,
156            user_order_by,
157            columns,
158            definition,
159            conflict_behavior,
160            vec![],
161            None,
162            vec![],
163            None,
164            table_type,
165            None,
166            cardinality,
167            retention_seconds,
168            create_type,
169            None,
170            Engine::Hummock,
171            false,
172        )?;
173
174        Self::new(input, table)
175    }
176
177    /// Create a materialize node, for `TABLE`.
178    ///
179    /// Different from `create`, the `columns` are passed in directly, instead of being derived from
180    /// the input. So the column IDs are preserved from the SQL columns binding step and will be
181    /// consistent with the source node and DML node.
182    #[expect(clippy::too_many_arguments)]
183    pub fn create_for_table(
184        input: PlanRef,
185        name: String,
186        database_id: DatabaseId,
187        schema_id: SchemaId,
188        user_distributed_by: RequiredDist,
189        user_order_by: Order,
190        columns: Vec<ColumnCatalog>,
191        definition: String,
192        conflict_behavior: ConflictBehavior,
193        version_column_indices: Vec<usize>,
194        pk_column_indices: Vec<usize>,
195        ttl_watermark_indices: Vec<usize>,
196        row_id_index: Option<usize>,
197        version: TableVersion,
198        retention_seconds: Option<NonZeroU32>,
199        webhook_info: Option<PbWebhookSourceInfo>,
200        engine: Engine,
201        refreshable: bool,
202    ) -> Result<Self> {
203        let input = Self::rewrite_input(input, user_distributed_by.clone(), TableType::Table)?;
204
205        let table = Self::derive_table_catalog(
206            input.clone(),
207            name.clone(),
208            database_id,
209            schema_id,
210            user_distributed_by,
211            user_order_by,
212            columns,
213            definition,
214            conflict_behavior,
215            version_column_indices,
216            Some(pk_column_indices),
217            ttl_watermark_indices,
218            row_id_index,
219            TableType::Table,
220            Some(version),
221            Cardinality::unknown(), // unknown cardinality for tables
222            retention_seconds,
223            CreateType::Foreground,
224            webhook_info,
225            engine,
226            refreshable,
227        )?;
228
229        // For refreshable tables, create staging table and progress table
230        let (staging_table, refresh_progress_table) = if refreshable {
231            let staging = Some(Self::derive_staging_table_catalog(table.clone()));
232            let progress = Some(Self::derive_refresh_progress_table_catalog(table.clone()));
233            (staging, progress)
234        } else {
235            (None, None)
236        };
237
238        tracing::info!(
239            table_name = %name,
240            refreshable = %refreshable,
241            has_staging_table = %staging_table.is_some(),
242            has_progress_table = %refresh_progress_table.is_some(),
243            "Creating StreamMaterialize with staging and progress table info"
244        );
245
246        Self::new_with_staging_and_progress(input, table, staging_table, refresh_progress_table)
247    }
248
249    /// Rewrite the input to satisfy the required distribution if necessary, according to the type.
250    fn rewrite_input(
251        input: PlanRef,
252        user_distributed_by: RequiredDist,
253        table_type: TableType,
254    ) -> Result<PlanRef> {
255        let required_dist = match input.distribution() {
256            Distribution::Single => RequiredDist::single(),
257            _ => match table_type {
258                TableType::Table => {
259                    assert_matches!(
260                        user_distributed_by,
261                        RequiredDist::ShardByKey(_) | RequiredDist::ShardByExactKey(_)
262                    );
263                    user_distributed_by
264                }
265                TableType::MaterializedView => {
266                    assert_matches!(user_distributed_by, RequiredDist::Any);
267                    // ensure the same pk will not shuffle to different node
268                    let required_dist =
269                        RequiredDist::shard_by_key(input.schema().len(), input.expect_stream_key());
270
271                    // If the input is a stream join, enforce the stream key as the materialized
272                    // view distribution key to avoid slow backfilling caused by
273                    // data skew of the dimension table join key.
274                    // See <https://github.com/risingwavelabs/risingwave/issues/12824> for more information.
275                    let is_stream_join = matches!(input.as_stream_hash_join(), Some(_join))
276                        || matches!(input.as_stream_temporal_join(), Some(_join))
277                        || matches!(input.as_stream_delta_join(), Some(_join));
278
279                    if is_stream_join {
280                        return Ok(required_dist.stream_enforce(input));
281                    }
282
283                    required_dist
284                }
285                TableType::Index => {
286                    assert_matches!(
287                        user_distributed_by,
288                        RequiredDist::PhysicalDist(Distribution::HashShard(_))
289                    );
290                    user_distributed_by
291                }
292                TableType::VectorIndex => {
293                    unreachable!("VectorIndex should not be created by StreamMaterialize")
294                }
295                TableType::Internal => unreachable!(),
296            },
297        };
298
299        required_dist.streaming_enforce_if_not_satisfies(input)
300    }
301
302    /// Derive the table catalog with the given arguments.
303    ///
304    /// - The caller must ensure the validity of the given `columns`.
305    /// - The `rewritten_input` should be generated by `rewrite_input`.
306    #[expect(clippy::too_many_arguments)]
307    fn derive_table_catalog(
308        rewritten_input: PlanRef,
309        name: String,
310        database_id: DatabaseId,
311        schema_id: SchemaId,
312        user_distributed_by: RequiredDist,
313        user_order_by: Order,
314        columns: Vec<ColumnCatalog>,
315        definition: String,
316        conflict_behavior: ConflictBehavior,
317        version_column_indices: Vec<usize>,
318        pk_column_indices: Option<Vec<usize>>, // Is some when create table
319        ttl_watermark_indices: Vec<usize>,
320        row_id_index: Option<usize>,
321        table_type: TableType,
322        version: Option<TableVersion>,
323        cardinality: Cardinality,
324        retention_seconds: Option<NonZeroU32>,
325        create_type: CreateType,
326        webhook_info: Option<PbWebhookSourceInfo>,
327        engine: Engine,
328        refreshable: bool,
329    ) -> Result<TableCatalog> {
330        let input = rewritten_input;
331
332        let value_indices = (0..columns.len()).collect_vec();
333        let distribution_key = input.distribution().dist_column_indices().to_vec();
334        let append_only = input.append_only();
335        // TODO(rc): In `TableCatalog` we still use `FixedBitSet` for watermark columns, ignoring the watermark group information.
336        // We will record the watermark group information in `TableCatalog` in the future. For now, let's flatten the watermark columns.
337        let watermark_columns = input.watermark_columns().indices().collect();
338
339        let (table_pk, mut stream_key) = if let Some(pk_column_indices) = pk_column_indices {
340            let table_pk = pk_column_indices
341                .iter()
342                .map(|idx| ColumnOrder::new(*idx, OrderType::ascending()))
343                .collect();
344            // No order by for create table, so stream key is identical to table pk.
345            (table_pk, pk_column_indices)
346        } else {
347            derive_pk(input, user_distributed_by, user_order_by, &columns)
348        };
349
350        // Guards only this table's own pk (table pk, index keys, MV ORDER BY); internal state
351        // tables are covered by `reject_variant_in_internal_storage_key` in the stream fragmenter.
352        for order in &table_pk {
353            let column = &columns[order.column_index];
354            if column.data_type().contains_variant() {
355                return Err(variant_key_error(format!(
356                    "VARIANT column \"{}\" is part of the storage primary key",
357                    column.name(),
358                )));
359            }
360        }
361
362        // Add TTL watermark column to stream key.
363        // When a row comes in to a TTL-ed table and we cannot find it in the table, we still cannot tell
364        // whether it is a new row or an update to an expired row. Adding the TTL watermark column to the stream key
365        // can ensure there's no double-insert from the view of the downstream jobs. See RFC for more details.
366        for idx in ttl_watermark_indices.iter().copied() {
367            if !stream_key.contains(&idx) {
368                stream_key.push(idx);
369            }
370        }
371
372        let read_prefix_len_hint = table_pk.len();
373        Ok(TableCatalog {
374            id: TableId::placeholder(),
375            schema_id,
376            database_id,
377            associated_source_id: None,
378            name,
379            columns,
380            pk: table_pk,
381            stream_key,
382            distribution_key,
383            table_type,
384            append_only,
385            owner: risingwave_common::catalog::DEFAULT_SUPER_USER_ID,
386            fragment_id: FragmentId::placeholder(),
387            dml_fragment_id: None,
388            vnode_col_index: None,
389            row_id_index,
390            value_indices,
391            definition,
392            conflict_behavior,
393            version_column_indices,
394            read_prefix_len_hint,
395            version,
396            watermark_columns,
397            dist_key_in_pk: vec![],
398            cardinality,
399            created_at_epoch: None,
400            initialized_at_epoch: None,
401            create_type,
402            stream_job_status: StreamJobStatus::Creating,
403            description: None,
404            initialized_at_cluster_version: None,
405            created_at_cluster_version: None,
406            retention_seconds: retention_seconds.map(|i| i.into()),
407            cdc_table_id: None,
408            vnode_count: VnodeCount::Placeholder, // will be filled in by the meta service later
409            webhook_info,
410            job_id: None,
411            engine: match table_type {
412                TableType::Table => engine,
413                TableType::MaterializedView
414                | TableType::Index
415                | TableType::Internal
416                | TableType::VectorIndex => {
417                    assert_eq!(engine, Engine::Hummock);
418                    engine
419                }
420            },
421            clean_watermark_index_in_pk: None, // TODO: fill this field
422            clean_watermark_indices: ttl_watermark_indices,
423            refreshable,
424            vector_index_info: None,
425            cdc_table_type: None,
426        })
427    }
428
429    /// The staging table is a pk-only table.
430    fn derive_staging_table_catalog(
431        TableCatalog {
432            id,
433            schema_id,
434            database_id,
435            associated_source_id,
436            name,
437            columns,
438            pk,
439            stream_key,
440            table_type: _,
441            distribution_key,
442            append_only,
443            cardinality,
444            owner,
445            retention_seconds,
446            fragment_id,
447            dml_fragment_id: _,
448            vnode_col_index,
449            row_id_index,
450            value_indices: _,
451            definition,
452            conflict_behavior,
453            version_column_indices,
454            read_prefix_len_hint: _,
455            version,
456            watermark_columns: _,
457            dist_key_in_pk,
458            created_at_epoch,
459            initialized_at_epoch,
460            create_type,
461            stream_job_status,
462            description,
463            created_at_cluster_version,
464            initialized_at_cluster_version,
465            cdc_table_id,
466            vnode_count,
467            webhook_info,
468            job_id,
469            engine,
470            clean_watermark_index_in_pk,
471            clean_watermark_indices,
472            refreshable,
473            vector_index_info,
474            cdc_table_type,
475        }: TableCatalog,
476    ) -> TableCatalog {
477        tracing::info!(
478            table_name = %name,
479            "Creating staging table for refreshable table"
480        );
481
482        assert!(row_id_index.is_none());
483        assert!(retention_seconds.is_none());
484        assert!(refreshable);
485
486        // Keep pk columns in pk order, so the staging chunk projected by pk indices can be
487        // written directly without reordering.
488        let pk_col_indices = pk.iter().map(|pk| pk.column_index).collect_vec();
489        let pk_cols = pk_col_indices
490            .iter()
491            .map(|&col_idx| columns[col_idx].clone())
492            .collect_vec();
493        let mapping = ColIndexMapping::with_remaining_columns(&pk_col_indices, columns.len());
494        TableCatalog {
495            id,
496            schema_id,
497            database_id,
498            associated_source_id,
499            name,
500            value_indices: (0..pk_cols.len()).collect(),
501            columns: pk_cols,
502            pk: pk
503                .iter()
504                .map(|pk| ColumnOrder::new(mapping.map(pk.column_index), pk.order_type))
505                .collect(),
506            stream_key: mapping.try_map_all(stream_key).unwrap(),
507            vnode_col_index: vnode_col_index.map(|i| mapping.map(i)),
508            // `dist_key_in_pk` is already based on pk positions, so it should keep unchanged.
509            dist_key_in_pk,
510            distribution_key: mapping.try_map_all(distribution_key).unwrap(),
511            table_type: TableType::Internal,
512            watermark_columns: FixedBitSet::new(),
513            append_only,
514            cardinality,
515            owner,
516            retention_seconds: None,
517            fragment_id,
518            dml_fragment_id: None,
519            row_id_index: None,
520            definition,
521            conflict_behavior,
522            version_column_indices,
523            // Refresh staging tables are consumed by the refresh merge path with
524            // `iter_keyed_row_with_vnode` range scans. They are not read by
525            // fixed-prefix lookups, so prefix SST filters would only add write cost.
526            read_prefix_len_hint: 0,
527            version,
528            created_at_epoch,
529            initialized_at_epoch,
530            create_type,
531            stream_job_status,
532            description,
533            created_at_cluster_version,
534            initialized_at_cluster_version,
535            cdc_table_id,
536            vnode_count,
537            webhook_info,
538            job_id,
539            engine,
540            clean_watermark_index_in_pk,
541            clean_watermark_indices,
542            refreshable: false,
543            vector_index_info,
544            cdc_table_type,
545        }
546    }
547
548    /// The refresh progress table is used to track refresh operation progress.
549    /// Simplified Schema: vnode (i32), `current_pos`... (variable PK from upstream),
550    /// `is_completed` (bool), `processed_rows` (i64)
551    fn derive_refresh_progress_table_catalog(table: TableCatalog) -> TableCatalog {
552        tracing::debug!(
553            table_name = %table.name,
554            "Creating refresh progress table for refreshable table"
555        );
556
557        // Define the simplified schema for the refresh progress table
558        // Schema: | vnode | current_pos... | is_completed | processed_rows |
559        let mut columns = vec![ColumnCatalog {
560            column_desc: risingwave_common::catalog::ColumnDesc::named(
561                "vnode",
562                0.into(),
563                DataType::Int16,
564            ),
565            is_hidden: false,
566        }];
567
568        // Add current_pos columns (mirror upstream table's primary key)
569        let mut col_index = 1;
570        for pk_col in &table.pk {
571            let upstream_col = &table.columns[pk_col.column_index];
572            columns.push(ColumnCatalog {
573                column_desc: risingwave_common::catalog::ColumnDesc::named(
574                    format!("pos_{}", upstream_col.name()),
575                    col_index.into(),
576                    upstream_col.data_type().clone(),
577                ),
578                is_hidden: false,
579            });
580            col_index += 1;
581        }
582
583        // Add metadata columns
584        for (name, data_type) in [
585            ("is_completed", DataType::Boolean),
586            ("processed_rows", DataType::Int64),
587        ] {
588            columns.push(ColumnCatalog {
589                column_desc: risingwave_common::catalog::ColumnDesc::named(
590                    name,
591                    col_index.into(),
592                    data_type,
593                ),
594                is_hidden: false,
595            });
596            col_index += 1;
597        }
598
599        let mut builder = TableCatalogBuilder::default();
600
601        // Add all columns to builder
602        for column in &columns {
603            builder.add_column(&(&column.column_desc).into());
604        }
605
606        // Primary key is vnode (column 0)
607        builder.add_order_column(0, OrderType::ascending());
608        builder.set_vnode_col_idx(0);
609        builder.set_value_indices((0..columns.len()).collect());
610        builder.set_dist_key_in_pk(vec![0]);
611
612        builder.build(vec![0], 1)
613    }
614
615    /// Get a reference to the stream materialize's table.
616    #[must_use]
617    pub fn table(&self) -> &TableCatalog {
618        &self.table
619    }
620
621    /// Get a reference to the stream materialize's staging table.
622    #[must_use]
623    pub fn staging_table(&self) -> Option<&TableCatalog> {
624        self.staging_table.as_ref()
625    }
626
627    /// Get a reference to the stream materialize's refresh progress table.
628    #[must_use]
629    pub fn refresh_progress_table(&self) -> Option<&TableCatalog> {
630        self.refresh_progress_table.as_ref()
631    }
632
633    pub fn name(&self) -> &str {
634        self.table.name()
635    }
636}
637
638impl Distill for StreamMaterialize {
639    fn distill<'a>(&self) -> XmlNode<'a> {
640        let table = self.table();
641
642        let column_names = (table.columns.iter())
643            .map(|col| col.name_with_hidden().to_string())
644            .map(Pretty::from)
645            .collect();
646
647        let stream_key = (table.stream_key().iter())
648            .map(|&k| table.columns[k].name().to_owned())
649            .map(Pretty::from)
650            .collect();
651
652        let pk_columns = (table.pk.iter())
653            .map(|o| table.columns[o.column_index].name().to_owned())
654            .map(Pretty::from)
655            .collect();
656        let mut vec = Vec::with_capacity(5);
657        vec.push(("columns", Pretty::Array(column_names)));
658        vec.push(("stream_key", Pretty::Array(stream_key)));
659        vec.push(("pk_columns", Pretty::Array(pk_columns)));
660        let pk_conflict_behavior = self.table.conflict_behavior().debug_to_string();
661
662        vec.push(("pk_conflict", Pretty::from(pk_conflict_behavior)));
663
664        let watermark_columns = &self.base.watermark_columns();
665        if self.base.watermark_columns().n_indices() > 0 {
666            // TODO(rc): we ignore the watermark group info here, will be fixed it later
667            let watermark_column_names = watermark_columns
668                .indices()
669                .map(|i| table.columns()[i].name_with_hidden().to_string())
670                .map(Pretty::from)
671                .collect();
672            vec.push(("watermark_columns", Pretty::Array(watermark_column_names)));
673        };
674        childless_record("StreamMaterialize", vec)
675    }
676}
677
678impl PlanTreeNodeUnary<Stream> for StreamMaterialize {
679    fn input(&self) -> PlanRef {
680        self.input.clone()
681    }
682
683    fn clone_with_input(&self, input: PlanRef) -> Self {
684        let new = Self::new_with_staging_and_progress(
685            input,
686            self.table().clone(),
687            self.staging_table.clone(),
688            self.refresh_progress_table.clone(),
689        )
690        .unwrap();
691        new.base
692            .schema()
693            .fields
694            .iter()
695            .zip_eq_fast(self.base.schema().fields.iter())
696            .for_each(|(a, b)| {
697                assert_eq!(a.data_type, b.data_type);
698            });
699        assert_eq!(new.plan_base().stream_key(), self.plan_base().stream_key());
700        new
701    }
702}
703
704impl_plan_tree_node_for_unary! { Stream, StreamMaterialize }
705
706impl StreamNode for StreamMaterialize {
707    fn to_stream_prost_body(&self, state: &mut BuildFragmentGraphState) -> PbNodeBody {
708        use risingwave_pb::stream_plan::*;
709
710        tracing::debug!(
711            table_name = %self.table().name(),
712            refreshable = %self.table().refreshable,
713            has_staging_table = %self.staging_table.is_some(),
714            has_progress_table = %self.refresh_progress_table.is_some(),
715            staging_table_name = ?self.staging_table.as_ref().map(|t| (&t.id, &t.name)),
716            progress_table_name = ?self.refresh_progress_table.as_ref().map(|t| (&t.id, &t.name)),
717            "Converting StreamMaterialize to protobuf"
718        );
719
720        let staging_table_prost = self
721            .staging_table
722            .clone()
723            .map(|t| t.with_id(state.gen_table_id_wrapped()).to_prost());
724
725        let refresh_progress_table_prost = self
726            .refresh_progress_table
727            .clone()
728            .map(|t| t.with_id(state.gen_table_id_wrapped()).to_prost());
729
730        PbNodeBody::Materialize(Box::new(MaterializeNode {
731            // Do not fill `table` and `table_id` here to avoid duplication. It will be filled by
732            // meta service after global information is generated.
733            table_id: 0.into(),
734            table: None,
735            // Pass staging table catalog if available for refreshable tables
736            staging_table: staging_table_prost,
737            // Pass refresh progress table catalog if available for refreshable tables
738            refresh_progress_table: refresh_progress_table_prost,
739
740            column_orders: self
741                .table()
742                .pk()
743                .iter()
744                .copied()
745                .map(ColumnOrder::to_protobuf)
746                .collect(),
747
748            // Equavalency: `clean_watermark_indices` is set iff there's a TTL watermark column.
749            // See `StreamMaterialize::derive_table_catalog`.
750            cleaned_by_ttl_watermark: !self.table.clean_watermark_indices.is_empty(),
751        }))
752    }
753}
754
755impl ExprRewritable<Stream> for StreamMaterialize {}
756
757impl ExprVisitable for StreamMaterialize {}
risingwave_frontend/optimizer/plan_node/stream_materialize.rs

risingwave_frontend/optimizer/plan_node/
stream_materialize.rs
