risingwave_frontend/handler/create_source/
additional_column.rs

1// Copyright 2024 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 super::*;
16
17// check the additional column compatibility with the format and encode
18fn check_additional_column_compatibility(
19    column_def: &IncludeOptionItem,
20    format_encode: Option<&FormatEncodeOptions>,
21) -> Result<()> {
22    // only allow header column have inner field
23    if column_def.inner_field.is_some()
24        && !column_def
25            .column_type
26            .real_value()
27            .eq_ignore_ascii_case("header")
28    {
29        return Err(RwError::from(ProtocolError(format!(
30            "Only header column can have inner field, but got {:?}",
31            column_def.column_type.real_value(),
32        ))));
33    }
34
35    // Payload column only allowed when encode is JSON
36    if let Some(schema) = format_encode
37        && column_def
38            .column_type
39            .real_value()
40            .eq_ignore_ascii_case("payload")
41        && !matches!(schema.row_encode, Encode::Json)
42    {
43        return Err(RwError::from(ProtocolError(format!(
44            "INCLUDE payload is only allowed when using ENCODE JSON, but got ENCODE {:?}",
45            schema.row_encode
46        ))));
47    }
48    Ok(())
49}
50
51/// add connector-spec columns to the end of column catalog
52pub fn handle_addition_columns(
53    format_encode: Option<&FormatEncodeOptions>,
54    with_properties: &BTreeMap<String, String>,
55    mut additional_columns: IncludeOption,
56    columns: &mut Vec<ColumnCatalog>,
57    is_cdc_backfill_table: bool,
58) -> Result<()> {
59    let connector_name = with_properties.get_connector().unwrap(); // there must be a connector in source
60
61    while let Some(item) = additional_columns.pop() {
62        check_additional_column_compatibility(&item, format_encode)?;
63
64        let data_type = item
65            .header_inner_expect_type
66            .map(|dt| bind_data_type(&dt))
67            .transpose()?;
68        if let Some(dt) = &data_type
69            && !matches!(dt, DataType::Bytea | DataType::Varchar)
70        {
71            return Err(
72                ErrorCode::BindError(format!("invalid additional column data type: {dt}")).into(),
73            );
74        }
75        let col = build_additional_column_desc(
76            ColumnId::placeholder(),
77            connector_name.as_str(),
78            item.column_type.real_value().as_str(),
79            item.column_alias.map(|alias| alias.real_value()),
80            item.inner_field.as_deref(),
81            data_type.as_ref(),
82            true,
83            is_cdc_backfill_table,
84        )?;
85        columns.push(ColumnCatalog::visible(col));
86    }
87
88    Ok(())
89}
90
91// Add a hidden column `_rw_kafka_timestamp` to each message from Kafka source.
92pub fn check_and_add_timestamp_column(
93    with_properties: &WithOptions,
94    columns: &mut Vec<ColumnCatalog>,
95) {
96    if with_properties.is_kafka_connector() {
97        if columns.iter().any(|col| {
98            matches!(
99                col.column_desc.additional_column.column_type,
100                Some(AdditionalColumnType::Timestamp(_))
101            )
102        }) {
103            // already has timestamp column, no need to add a new one
104            return;
105        }
106
107        // add a hidden column `_rw_kafka_timestamp` to each message from Kafka source
108        let col = build_additional_column_desc(
109            ColumnId::placeholder(),
110            KAFKA_CONNECTOR,
111            "timestamp",
112            Some(KAFKA_TIMESTAMP_COLUMN_NAME.to_owned()),
113            None,
114            None,
115            true,
116            false,
117        )
118        .unwrap();
119        columns.push(ColumnCatalog::hidden(col));
120    }
121}
risingwave_frontend/handler/create_source/additional_column.rs

risingwave_frontend/handler/create_source/
additional_column.rs
