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    fn is_exactly_once(properties: &BTreeMap<String, String>) -> Result<bool> {
217        let Some(value) = properties.get("is_exactly_once") else {
218            return Ok(true);
219        };
220        value.parse::<bool>().map_err(|_| {
221            SinkError::Config(anyhow!(
222                "invalid value for `is_exactly_once`: expected `true` or `false`, got `{value}`"
223            ))
224        })
225    }
226
227    async fn validate(&self) -> Result<()> {
228        let catalog_kind = self.config.catalog_kind()?;
229        if matches!(catalog_kind, IcebergCatalogKind::Snowflake) {
230            bail!("Snowflake catalog only supports iceberg sources");
231        }
232
233        if matches!(catalog_kind, IcebergCatalogKind::Glue(_)) {
234            risingwave_common::license::Feature::IcebergSinkWithGlue
235                .check_available()
236                .map_err(|e| anyhow::anyhow!(e))?;
237        }
238
239        // Enforce merge-on-read for append-only tables
240        IcebergConfig::validate_append_only_write_mode(
241            &self.config.r#type,
242            self.config.write_mode,
243        )?;
244        validate_explicit_compaction_type(&self.config)?;
245        validate_compaction_option_compatibility(&self.config)?;
246
247        // VARIANT is not comparable, so it can never be an equality-delete key.
248        if self.config.r#type == SINK_TYPE_UPSERT
249            && !self.config.force_append_only
250            && let Some(pk_indices) = self
251                .param
252                .downstream_pk
253                .as_ref()
254                .filter(|pk| !pk.is_empty())
255        {
256            for &idx in pk_indices {
257                if let Some(column) = self.param.columns.get(idx)
258                    && column.data_type.contains_variant()
259                {
260                    bail!(
261                        "VARIANT column `{}` cannot be used as the primary key of an upsert iceberg sink",
262                        column.name
263                    );
264                }
265            }
266        }
267
268        let table = self.create_and_validate_table().await?;
269        self.config
270            .validate_manifest_rewrite_format(table.metadata().format_version())?;
271        Ok(())
272    }
273
274    fn support_schema_change() -> bool {
275        true
276    }
277
278    fn validate_alter_config_change(
279        config: &BTreeMap<String, String>,
280        alter_props: &BTreeMap<String, String>,
281    ) -> Result<()> {
282        let iceberg_config = IcebergConfig::from_btreemap(config.clone())?;
283        if iceberg_config.enable_pk_index {
284            // PK-index executors do not rebuild their writers on ConnectorPropsChange.
285            // Maintenance options are consumed by Meta, but accepting writer options here
286            // would leave the running writer and the persisted configuration out of sync.
287            // The ordinary sink's backlog and delete-file-count knobs also have no consumer
288            // in the PK-index path, which commits each checkpoint and selects by delete ratio.
289            for key in [
290                "commit_checkpoint_interval",
291                COMPACTION_MAX_SNAPSHOTS_NUM,
292                COMPACTION_DELETE_FILES_COUNT_THRESHOLD,
293                COMPACTION_WRITE_PARQUET_COMPRESSION,
294                COMPACTION_WRITE_PARQUET_MAX_ROW_GROUP_ROWS,
295                COMPACTION_WRITE_PARQUET_MAX_ROW_GROUP_BYTES,
296            ] {
297                if alter_props.contains_key(key) {
298                    bail!("altering `{key}` is not supported for Iceberg sinks with PK index");
299                }
300            }
301        }
302        let compaction_type_changed = alter_props.contains_key(COMPACTION_TYPE);
303        let compaction_options_changed = compaction_type_changed
304            || alter_props.contains_key(COMPACTION_SMALL_FILES_THRESHOLD_MB)
305            || alter_props.contains_key(COMPACTION_DELETE_FILES_COUNT_THRESHOLD);
306        let enabling_compaction = alter_props
307            .get(ENABLE_COMPACTION)
308            .is_some_and(|value| value.eq_ignore_ascii_case("true"));
309
310        if compaction_options_changed || enabling_compaction {
311            let validate_explicit_type = compaction_type_changed
312                || (enabling_compaction
313                    && iceberg_config.write_mode == IcebergWriteMode::MergeOnRead);
314
315            // Persisted COW types are legacy-only and must not block compaction activation.
316            if validate_explicit_type {
317                validate_explicit_compaction_type(&iceberg_config)?;
318            }
319            validate_compaction_option_compatibility(&iceberg_config)?;
320        }
321
322        Self::validate_alter_config(config)
323    }
324
325    fn validate_alter_config(config: &BTreeMap<String, String>) -> Result<()> {
326        let iceberg_config = IcebergConfig::from_btreemap(config.clone())?;
327
328        // Validate compaction interval
329        if let Some(compaction_interval) = iceberg_config.compaction_interval_sec {
330            if iceberg_config.enable_compaction && compaction_interval == 0 {
331                bail!(
332                    "`compaction-interval-sec` must be greater than 0 when `enable-compaction` is true"
333                );
334            }
335
336            tracing::info!(
337                "Alter config compaction_interval set to {} seconds",
338                compaction_interval
339            );
340        }
341
342        // Validate max snapshots
343        if let Some(max_snapshots) = iceberg_config.max_snapshots_num_before_compaction
344            && max_snapshots < 1
345        {
346            bail!(
347                "`compaction.max_snapshots_num` must be greater than 0, got: {}",
348                max_snapshots
349            );
350        }
351
352        // Validate target file size
353        if let Some(target_file_size_mb) = iceberg_config.target_file_size_mb
354            && target_file_size_mb == 0
355        {
356            bail!("`compaction.target_file_size_mb` must be greater than 0");
357        }
358
359        // Validate parquet max row group rows
360        if let Some(max_row_group_rows) = iceberg_config.write_parquet_max_row_group_rows
361            && max_row_group_rows == 0
362        {
363            bail!("`compaction.write_parquet_max_row_group_rows` must be greater than 0");
364        }
365
366        // Validate parquet max row group bytes
367        if let Some(max_row_group_bytes) = iceberg_config.write_parquet_max_row_group_bytes
368            && max_row_group_bytes == 0
369        {
370            bail!("`compaction.write_parquet_max_row_group_bytes` must be greater than 0");
371        }
372
373        // Validate parquet compression codec
374        if let Some(ref compression) = iceberg_config.write_parquet_compression {
375            let valid_codecs = [
376                "uncompressed",
377                "snappy",
378                "gzip",
379                "lzo",
380                "brotli",
381                "lz4",
382                "zstd",
383            ];
384            if !valid_codecs.contains(&compression.to_lowercase().as_str()) {
385                bail!(
386                    "`compaction.write_parquet_compression` must be one of {:?}, got: {}",
387                    valid_codecs,
388                    compression
389                );
390            }
391        }
392
393        Ok(())
394    }
395
396    async fn new_log_sinker(&self, writer_param: SinkWriterParam) -> Result<Self::LogSinker> {
397        let writer = IcebergSinkWriter::new(
398            self.config.clone(),
399            self.param.clone(),
400            writer_param.clone(),
401            self.upsert_primary_key_column_names.clone(),
402        );
403
404        let commit_checkpoint_interval =
405            NonZeroU64::new(self.config.commit_checkpoint_interval).expect(
406                "commit_checkpoint_interval should be greater than 0, and it should be checked in config validation",
407            );
408        let log_sinker = CoordinatedLogSinker::new(
409            &writer_param,
410            self.param.clone(),
411            writer,
412            commit_checkpoint_interval,
413        )
414        .await?;
415
416        Ok(log_sinker)
417    }
418
419    fn is_coordinated_sink(&self) -> bool {
420        true
421    }
422
423    async fn new_coordinator(
424        &self,
425        iceberg_compact_stat_sender: Option<UnboundedSender<IcebergSinkCompactionUpdate>>,
426    ) -> Result<SinkCommitCoordinator> {
427        let catalog = self.config.create_catalog().await?;
428        let table = self.create_and_validate_table().await?;
429        let coordinator = IcebergSinkCommitter {
430            catalog,
431            table,
432            last_commit_epoch: 0,
433            sink_id: self.param.sink_id,
434            config: self.config.clone(),
435            param: self.param.clone(),
436            commit_retry_num: self.config.commit_retry_num,
437            iceberg_compact_stat_sender,
438        };
439        if Self::is_exactly_once(&self.param.properties)? {
440            Ok(SinkCommitCoordinator::TwoPhase(Box::new(coordinator)))
441        } else {
442            Ok(SinkCommitCoordinator::SinglePhase(Box::new(coordinator)))
443        }
444    }
445}
risingwave_connector/sink/iceberg/mod.rs

risingwave_connector/sink/iceberg/
mod.rs
