risingwave_connector/sink/iceberg/
mod.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
15#[cfg(test)]
16mod test;
17
18mod commit;
19pub mod commit_retry;
20mod config;
21mod create_table;
22mod engine_options;
23mod metadata;
24#[cfg(any(test, madsim))]
25pub mod mock_v3_catalog_registry;
26mod position_delete;
27mod prometheus;
28mod writer;
29
30use std::collections::BTreeMap;
31use std::fmt::Debug;
32use std::num::NonZeroU64;
33
34use anyhow::{Context, anyhow};
35pub use commit::*;
36pub use config::*;
37pub use create_table::*;
38pub use engine_options::*;
39use iceberg::table::Table;
40pub use metadata::*;
41pub use position_delete::*;
42use risingwave_common::bail;
43use risingwave_common::license::Feature;
44use tokio::sync::mpsc::UnboundedSender;
45pub use writer::*;
46
47use super::{
48    GLOBAL_SINK_METRICS, SINK_TYPE_APPEND_ONLY, SINK_TYPE_OPTION, SINK_TYPE_UPSERT, Sink,
49    SinkError, SinkWriterParam,
50};
51use crate::connector_common::{IcebergCatalogKind, IcebergSinkCompactionUpdate};
52use crate::enforce_secret::EnforceSecret;
53use crate::sink::coordinate::CoordinatedLogSinker;
54use crate::sink::{Result, SinkCommitCoordinator, SinkParam};
55
56pub const ICEBERG_SINK: &str = "iceberg";
57
58pub struct IcebergSink {
59    pub config: IcebergConfig,
60    param: SinkParam,
61    // In upsert mode, it is never None or empty.
62    upsert_primary_key_column_names: Option<Vec<String>>,
63}
64
65impl EnforceSecret for IcebergSink {
66    fn enforce_secret<'a>(
67        prop_iter: impl Iterator<Item = &'a str>,
68    ) -> crate::error::ConnectorResult<()> {
69        for prop in prop_iter {
70            IcebergConfig::enforce_one(prop)?;
71        }
72        Ok(())
73    }
74}
75
76impl TryFrom<SinkParam> for IcebergSink {
77    type Error = SinkError;
78
79    fn try_from(param: SinkParam) -> std::result::Result<Self, Self::Error> {
80        let config = IcebergConfig::from_btreemap(param.properties.clone())?;
81        IcebergSink::new(config, param)
82    }
83}
84
85impl Debug for IcebergSink {
86    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
87        f.debug_struct("IcebergSink")
88            .field("config", &self.config)
89            .finish()
90    }
91}
92
93fn validate_explicit_compaction_type(config: &IcebergConfig) -> Result<()> {
94    let Some(compaction_type) = config.compaction_type else {
95        return Ok(());
96    };
97
98    if config.write_mode == IcebergWriteMode::CopyOnWrite {
99        bail!(
100            "`compaction.type` must not be set when `write_mode` is `copy-on-write`; \
101             copy-on-write selects its compaction policy automatically"
102        );
103    }
104
105    if !matches!(compaction_type, CompactionType::Full) {
106        Feature::IcebergCompaction
107            .check_available()
108            .map_err(|e| anyhow!(e))?;
109    }
110
111    Ok(())
112}
113
114fn validate_compaction_option_compatibility(config: &IcebergConfig) -> Result<()> {
115    // Value ranges are validated by `IcebergConfig::from_btreemap`. This only rejects
116    // options that are incompatible with the selected merge-on-read strategy.
117    // COW ignores legacy persisted types, so merge-on-read type-option compatibility does not apply.
118    if config.write_mode == IcebergWriteMode::CopyOnWrite {
119        return Ok(());
120    }
121
122    let Some(compaction_type) = config.compaction_type else {
123        return Ok(());
124    };
125
126    let unsupported_option = match compaction_type {
127        // Auto uses both selection thresholds.
128        CompactionType::Auto => None,
129        // Keep accepting strategy-specific thresholds that Full ignores.
130        CompactionType::Full => None,
131        CompactionType::SmallFiles => config
132            .delete_files_count_threshold
133            .is_some()
134            .then_some(COMPACTION_DELETE_FILES_COUNT_THRESHOLD),
135        CompactionType::FilesWithDelete => config
136            .small_files_threshold_mb
137            .is_some()
138            .then_some(COMPACTION_SMALL_FILES_THRESHOLD_MB),
139    };
140    if let Some(option) = unsupported_option {
141        bail!(
142            "`{option}` is not supported for '{}' compaction type",
143            compaction_type.as_str()
144        );
145    }
146
147    Ok(())
148}
149
150impl IcebergSink {
151    pub async fn create_and_validate_table(&self) -> Result<Table> {
152        create_and_validate_table_impl(&self.config, &self.param).await
153    }
154
155    /// Returns `true` if this call created the table, `false` if it already existed.
156    pub async fn create_table_if_not_exists(&self) -> Result<bool> {
157        create_table_if_not_exists_impl(&self.config, &self.param).await
158    }
159
160    pub fn new(config: IcebergConfig, param: SinkParam) -> Result<Self> {
161        if let Some(order_key) = &config.order_key {
162            validate_order_key_columns(
163                order_key,
164                param.columns.iter().map(|column| column.name.as_str()),
165            )
166            .context("invalid order_key")
167            .map_err(SinkError::Config)?;
168        }
169
170        let upsert_primary_key_column_names =
171            if config.r#type == SINK_TYPE_UPSERT && !config.force_append_only {
172                let pk_indices = param
173                    .downstream_pk
174                    .as_ref()
175                    .filter(|pk| !pk.is_empty())
176                    .ok_or_else(|| {
177                        SinkError::Config(anyhow!(
178                            "primary key must be specified for upsert iceberg sink"
179                        ))
180                    })?;
181                Some(
182                    pk_indices
183                        .iter()
184                        .map(|&idx| {
185                            param
186                                .columns
187                                .get(idx)
188                                .map(|column| column.name.clone())
189                                .ok_or_else(|| {
190                                    SinkError::Config(anyhow!(
191                                        "primary key column index {} out of range in sink schema",
192                                        idx
193                                    ))
194                                })
195                        })
196                        .collect::<Result<Vec<_>>>()?,
197                )
198            } else {
199                None
200            };
201        Ok(Self {
202            config,
203            param,
204            upsert_primary_key_column_names,
205        })
206    }
207}
208
209impl Sink for IcebergSink {
210    type LogSinker = CoordinatedLogSinker<IcebergSinkWriter>;
211
212    const SINK_NAME: &'static str = ICEBERG_SINK;
213
214    crate::impl_validate_sink_unknown_fields!();
215
216    async fn validate(&self) -> Result<()> {
217        let catalog_kind = self.config.catalog_kind()?;
218        if matches!(catalog_kind, IcebergCatalogKind::Snowflake) {
219            bail!("Snowflake catalog only supports iceberg sources");
220        }
221
222        if matches!(catalog_kind, IcebergCatalogKind::Glue(_)) {
223            risingwave_common::license::Feature::IcebergSinkWithGlue
224                .check_available()
225                .map_err(|e| anyhow::anyhow!(e))?;
226        }
227
228        // Enforce merge-on-read for append-only tables
229        IcebergConfig::validate_append_only_write_mode(
230            &self.config.r#type,
231            self.config.write_mode,
232        )?;
233        validate_explicit_compaction_type(&self.config)?;
234        validate_compaction_option_compatibility(&self.config)?;
235
236        let table = self.create_and_validate_table().await?;
237        self.config
238            .validate_manifest_rewrite_format(table.metadata().format_version())?;
239        Ok(())
240    }
241
242    fn support_schema_change() -> bool {
243        true
244    }
245
246    fn validate_alter_config_change(
247        config: &BTreeMap<String, String>,
248        alter_props: &BTreeMap<String, String>,
249    ) -> Result<()> {
250        let compaction_type_changed = alter_props.contains_key(COMPACTION_TYPE);
251        let compaction_options_changed = compaction_type_changed
252            || alter_props.contains_key(COMPACTION_SMALL_FILES_THRESHOLD_MB)
253            || alter_props.contains_key(COMPACTION_DELETE_FILES_COUNT_THRESHOLD);
254        let enabling_compaction = alter_props
255            .get(ENABLE_COMPACTION)
256            .is_some_and(|value| value.eq_ignore_ascii_case("true"));
257
258        if compaction_options_changed || enabling_compaction {
259            let iceberg_config = IcebergConfig::from_btreemap(config.clone())?;
260            let validate_explicit_type = compaction_type_changed
261                || (enabling_compaction
262                    && iceberg_config.write_mode == IcebergWriteMode::MergeOnRead);
263
264            // Persisted COW types are legacy-only and must not block compaction activation.
265            if validate_explicit_type {
266                validate_explicit_compaction_type(&iceberg_config)?;
267            }
268            validate_compaction_option_compatibility(&iceberg_config)?;
269        }
270
271        Self::validate_alter_config(config)
272    }
273
274    fn validate_alter_config(config: &BTreeMap<String, String>) -> Result<()> {
275        let iceberg_config = IcebergConfig::from_btreemap(config.clone())?;
276
277        // Validate compaction interval
278        if let Some(compaction_interval) = iceberg_config.compaction_interval_sec {
279            if iceberg_config.enable_compaction && compaction_interval == 0 {
280                bail!(
281                    "`compaction-interval-sec` must be greater than 0 when `enable-compaction` is true"
282                );
283            }
284
285            tracing::info!(
286                "Alter config compaction_interval set to {} seconds",
287                compaction_interval
288            );
289        }
290
291        // Validate max snapshots
292        if let Some(max_snapshots) = iceberg_config.max_snapshots_num_before_compaction
293            && max_snapshots < 1
294        {
295            bail!(
296                "`compaction.max_snapshots_num` must be greater than 0, got: {}",
297                max_snapshots
298            );
299        }
300
301        // Validate target file size
302        if let Some(target_file_size_mb) = iceberg_config.target_file_size_mb
303            && target_file_size_mb == 0
304        {
305            bail!("`compaction.target_file_size_mb` must be greater than 0");
306        }
307
308        // Validate parquet max row group rows
309        if let Some(max_row_group_rows) = iceberg_config.write_parquet_max_row_group_rows
310            && max_row_group_rows == 0
311        {
312            bail!("`compaction.write_parquet_max_row_group_rows` must be greater than 0");
313        }
314
315        // Validate parquet max row group bytes
316        if let Some(max_row_group_bytes) = iceberg_config.write_parquet_max_row_group_bytes
317            && max_row_group_bytes == 0
318        {
319            bail!("`compaction.write_parquet_max_row_group_bytes` must be greater than 0");
320        }
321
322        // Validate parquet compression codec
323        if let Some(ref compression) = iceberg_config.write_parquet_compression {
324            let valid_codecs = [
325                "uncompressed",
326                "snappy",
327                "gzip",
328                "lzo",
329                "brotli",
330                "lz4",
331                "zstd",
332            ];
333            if !valid_codecs.contains(&compression.to_lowercase().as_str()) {
334                bail!(
335                    "`compaction.write_parquet_compression` must be one of {:?}, got: {}",
336                    valid_codecs,
337                    compression
338                );
339            }
340        }
341
342        Ok(())
343    }
344
345    async fn new_log_sinker(&self, writer_param: SinkWriterParam) -> Result<Self::LogSinker> {
346        let writer = IcebergSinkWriter::new(
347            self.config.clone(),
348            self.param.clone(),
349            writer_param.clone(),
350            self.upsert_primary_key_column_names.clone(),
351        );
352
353        let commit_checkpoint_interval =
354            NonZeroU64::new(self.config.commit_checkpoint_interval).expect(
355                "commit_checkpoint_interval should be greater than 0, and it should be checked in config validation",
356            );
357        let log_sinker = CoordinatedLogSinker::new(
358            &writer_param,
359            self.param.clone(),
360            writer,
361            commit_checkpoint_interval,
362        )
363        .await?;
364
365        Ok(log_sinker)
366    }
367
368    fn is_coordinated_sink(&self) -> bool {
369        true
370    }
371
372    async fn new_coordinator(
373        &self,
374        iceberg_compact_stat_sender: Option<UnboundedSender<IcebergSinkCompactionUpdate>>,
375    ) -> Result<SinkCommitCoordinator> {
376        let catalog = self.config.create_catalog().await?;
377        let table = self.create_and_validate_table().await?;
378        let coordinator = IcebergSinkCommitter {
379            catalog,
380            table,
381            last_commit_epoch: 0,
382            sink_id: self.param.sink_id,
383            config: self.config.clone(),
384            param: self.param.clone(),
385            commit_retry_num: self.config.commit_retry_num,
386            iceberg_compact_stat_sender,
387        };
388        if self.config.is_exactly_once.unwrap_or_default() {
389            Ok(SinkCommitCoordinator::TwoPhase(Box::new(coordinator)))
390        } else {
391            Ok(SinkCommitCoordinator::SinglePhase(Box::new(coordinator)))
392        }
393    }
394}
risingwave_connector/sink/iceberg/mod.rs

risingwave_connector/sink/iceberg/
mod.rs
