Repository: risingwavelabs/risingwave
File: src/frontend/src/optimizer/plan_node/stream_sink.rs (Rust)

Code before:
    /// Extract user-defined downstream pk columns from with options. Return the indices of the pk
    /// columns. An empty list of columns is not allowed.
    ///
    /// The format of `downstream_pk_str` should be 'col1,col2,...' (delimited by `,`) in order to
    /// get parsed.
    fn parse_downstream_pk(
        downstream_pk_str: &str,
        columns: &[ColumnCatalog],
    ) -> Result<Vec<usize>> {
        // If the user defines the downstream primary key, we find out their indices.
        let downstream_pk = downstream_pk_str.split(',').collect_vec();
        let mut downstream_pk_indices = Vec::with_capacity(downstream_pk.len());
        for key in downstream_pk {
            let trimmed_key = key.trim();
            if trimmed_key.is_empty() {
                continue;
            }
            downstream_pk_indices.push(find_column_idx_by_name(columns, trimmed_key)?);
        }
        if downstream_pk_indices.is_empty() {
            bail_invalid_input_syntax!(
                "Specified primary key should not be empty. \
                To use derived primary key, remove {DOWNSTREAM_PK_KEY} from WITH options instead."
            );
        }
        Ok(downstream_pk_indices)
    }

    /// The table schema is: | epoch | seq id | row op | sink columns |
    /// Pk is: | epoch | seq id |
    fn infer_kv_log_store_table_catalog(&self) -> TableCatalog {
        infer_kv_log_store_table_catalog_inner(&self.input, &self.sink_desc().columns)
    }
}

impl PlanTreeNodeUnary<Stream> for StreamSink {
    fn input(&self) -> PlanRef {
        self.input.clone()
    }

    fn clone_with_input(&self, input: PlanRef) -> Self {
        Self::new(input, self.sink_desc.clone(), self.log_store_type)
        // TODO(nanderstabel): Add assertions (assert_eq!)
    }
}

impl_plan_tree_node_for_unary! { Stream, StreamSink }

impl Distill for StreamSink {
    fn distill<'a>(&self) -> XmlNode<'a> {
        let sink_type = if self.sink_desc.sink_type.is_append_only() {

Review comment (suggestion):
This mapping logic conflates `StreamKind::Retract` and `StreamKind::Upsert` to `SinkType::Upsert`. Given the PR changes stream semantics, add a comment explaining why both stream kinds map to the same sink type and how this relates to the comment on lines 188-192.

Code after:
                            if syntax_legacy { "WITH" } else { "FORMAT ENCODE" }
                    ))
                        .into());
                    }
                }
                SinkType::ForceAppendOnly => unreachable!(),
                SinkType::Upsert => { /* always qualified */ }
                SinkType::Retract => {
                    if derived_stream_kind == StreamKind::Upsert {
                        bail_invalid_input_syntax!(
                            "The sink of upsert stream cannot be retract. \
                             Please create a materialized view or sink-into-table with this query before sinking it.",
                        );
                    }
                }
            }
            Ok(user_defined_sink_type)
        } else {
            // No specification at all, follow the optimizer's derivation.
            // This is also the case for sink-into-table.
            Ok(match derived_stream_kind {
                // We downgrade `Retract` to `Upsert` unless explicitly specified the type in options,
                // as it is well supported by most sinks and reduces the amount of data written.
                StreamKind::Retract | StreamKind::Upsert => SinkType::Upsert,
                StreamKind::AppendOnly => SinkType::AppendOnly,
            })
        }
    }

    /// Extract user-defined downstream pk columns from with options. Return the indices of the pk
    /// columns. An empty list of columns is not allowed.
    ///
    /// The format of `downstream_pk_str` should be 'col1,col2,...' (delimited by `,`) in order to
    /// get parsed.
    fn parse_downstream_pk(
        downstream_pk_str: &str,
        columns: &[ColumnCatalog],
    ) -> Result<Vec<usize>> {
        // If the user defines the downstream primary key, we find out their indices.
        let downstream_pk = downstream_pk_str.split(',').collect_vec();
        let mut downstream_pk_indices = Vec::with_capacity(downstream_pk.len());
        for key in downstream_pk {
            let trimmed_key = key.trim();
            if trimmed_key.is_empty() {
                continue;
            }
            downstream_pk_indices.push(find_column_idx_by_name(columns, trimmed_key)?);
        }
        if downstream_pk_indices.is_empty() {
            bail_invalid_input_syntax!(
                "Specified primary key should not be empty. \