risingwave_connector/sink/catalog/
desc.rs

1// Copyright 2023 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::collections::BTreeMap;
16
17use itertools::Itertools;
18use risingwave_common::catalog::{
19    ColumnCatalog, ConnectionId, CreateType, DatabaseId, SchemaId, StreamJobStatus, TableId, UserId,
20};
21use risingwave_common::util::sort_util::ColumnOrder;
22use risingwave_pb::secret::PbSecretRef;
23use risingwave_pb::stream_plan::PbSinkDesc;
24
25use super::{SinkCatalog, SinkFormatDesc, SinkId, SinkType};
26
27#[derive(Debug, Clone, PartialEq, Eq, Hash)]
28pub struct SinkDesc {
29    /// Id of the sink. For debug now.
30    pub id: SinkId,
31
32    /// Name of the sink. For debug now.
33    pub name: String,
34
35    /// Full SQL definition of the sink. For debug now.
36    pub definition: String,
37
38    /// All columns of the sink. Note that this is NOT sorted by columnId in the vector.
39    pub columns: Vec<ColumnCatalog>,
40
41    /// Primary keys of the sink. Derived by the frontend.
42    pub plan_pk: Vec<ColumnOrder>,
43
44    /// User-defined primary key indices for upsert sink, if any.
45    pub downstream_pk: Option<Vec<usize>>,
46
47    /// Distribution key indices of the sink. For example, if `distribution_key = [1, 2]`, then the
48    /// distribution keys will be `columns[1]` and `columns[2]`.
49    pub distribution_key: Vec<usize>,
50
51    /// The properties of the sink.
52    pub properties: BTreeMap<String, String>,
53
54    /// Secret ref
55    pub secret_refs: BTreeMap<String, PbSecretRef>,
56
57    // The append-only behavior of the physical sink connector. Frontend will determine `sink_type`
58    // based on both its own derivation on the append-only attribute and other user-specified
59    // options in `properties`.
60    pub sink_type: SinkType,
61
62    /// Whether to drop DELETE and convert UPDATE to INSERT in the sink executor.
63    pub ignore_delete: bool,
64
65    // The format and encode of the sink.
66    pub format_desc: Option<SinkFormatDesc>,
67
68    /// Name of the database
69    pub db_name: String,
70
71    /// Name of the "table" field for Debezium. If the sink is from table or mv,
72    /// it is the name of table/mv. Otherwise, it is the name of the sink.
73    pub sink_from_name: String,
74
75    /// Id of the target table for sink into table.
76    pub target_table: Option<TableId>,
77
78    /// See the same name field in `SinkWriterParam`.
79    pub extra_partition_col_idx: Option<usize>,
80
81    /// Whether the sink job should run in foreground or background.
82    pub create_type: CreateType,
83
84    pub is_exactly_once: Option<bool>,
85
86    pub auto_refresh_schema_from_table: Option<TableId>,
87}
88
89impl SinkDesc {
90    pub fn into_catalog(
91        self,
92        schema_id: SchemaId,
93        database_id: DatabaseId,
94        owner: UserId,
95        connection_id: Option<ConnectionId>,
96    ) -> SinkCatalog {
97        SinkCatalog {
98            id: self.id,
99            schema_id,
100            database_id,
101            name: self.name,
102            definition: self.definition,
103            columns: self.columns,
104            plan_pk: self.plan_pk,
105            downstream_pk: self.downstream_pk,
106            distribution_key: self.distribution_key,
107            owner,
108            properties: self.properties,
109            secret_refs: self.secret_refs,
110            sink_type: self.sink_type,
111            ignore_delete: self.ignore_delete,
112            format_desc: self.format_desc,
113            connection_id,
114            created_at_epoch: None,
115            initialized_at_epoch: None,
116            db_name: self.db_name,
117            sink_from_name: self.sink_from_name,
118            auto_refresh_schema_from_table: self.auto_refresh_schema_from_table,
119            target_table: self.target_table,
120            created_at_cluster_version: None,
121            initialized_at_cluster_version: None,
122            create_type: self.create_type,
123            original_target_columns: vec![],
124            stream_job_status: StreamJobStatus::Creating,
125        }
126    }
127
128    pub fn to_proto(&self) -> PbSinkDesc {
129        PbSinkDesc {
130            id: self.id,
131            name: self.name.clone(),
132            definition: self.definition.clone(),
133            column_catalogs: self
134                .columns
135                .iter()
136                .map(|column| column.to_protobuf())
137                .collect_vec(),
138            plan_pk: self.plan_pk.iter().map(|k| k.to_protobuf()).collect_vec(),
139            downstream_pk: (self.downstream_pk.as_ref())
140                .map_or_else(Vec::new, |pk| pk.iter().map(|idx| *idx as _).collect_vec()),
141            distribution_key: self.distribution_key.iter().map(|k| *k as _).collect_vec(),
142            properties: self.properties.clone().into_iter().collect(),
143            sink_type: self.sink_type.to_proto() as i32,
144            raw_ignore_delete: self.ignore_delete,
145            format_desc: self.format_desc.as_ref().map(|f| f.to_proto()),
146            db_name: self.db_name.clone(),
147            sink_from_name: self.sink_from_name.clone(),
148            target_table: self.target_table.map(|table_id| table_id.as_raw_id()),
149            extra_partition_col_idx: self.extra_partition_col_idx.map(|idx| idx as u64),
150            secret_refs: self.secret_refs.clone(),
151        }
152    }
153}
risingwave_connector/sink/catalog/desc.rs

risingwave_connector/sink/catalog/
desc.rs
