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 and commit intervals are consumed by Meta, but accepting
286            // writer options here would leave the writer and 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 selects by delete ratio.
289            for key in [
290                COMPACTION_MAX_SNAPSHOTS_NUM,
291                COMPACTION_DELETE_FILES_COUNT_THRESHOLD,
292                COMPACTION_WRITE_PARQUET_COMPRESSION,
293                COMPACTION_WRITE_PARQUET_MAX_ROW_GROUP_ROWS,
294                COMPACTION_WRITE_PARQUET_MAX_ROW_GROUP_BYTES,
295            ] {
296                if alter_props.contains_key(key) {
297                    bail!("altering `{key}` is not supported for Iceberg sinks with PK index");
298                }
299            }
300        }
301        let compaction_type_changed = alter_props.contains_key(COMPACTION_TYPE);
302        let compaction_options_changed = compaction_type_changed
303            || alter_props.contains_key(COMPACTION_SMALL_FILES_THRESHOLD_MB)
304            || alter_props.contains_key(COMPACTION_DELETE_FILES_COUNT_THRESHOLD);
305        let enabling_compaction = alter_props
306            .get(ENABLE_COMPACTION)
307            .is_some_and(|value| value.eq_ignore_ascii_case("true"));
308
309        if compaction_options_changed || enabling_compaction {
310            let validate_explicit_type = compaction_type_changed
311                || (enabling_compaction
312                    && iceberg_config.write_mode == IcebergWriteMode::MergeOnRead);
313
314            // Persisted COW types are legacy-only and must not block compaction activation.
315            if validate_explicit_type {
316                validate_explicit_compaction_type(&iceberg_config)?;
317            }
318            validate_compaction_option_compatibility(&iceberg_config)?;
319        }
320
321        Self::validate_alter_config(config)
322    }
323
324    fn validate_alter_config(config: &BTreeMap<String, String>) -> Result<()> {
325        let iceberg_config = IcebergConfig::from_btreemap(config.clone())?;
326
327        // Validate compaction interval
328        if let Some(compaction_interval) = iceberg_config.compaction_interval_sec {
329            if iceberg_config.enable_compaction && compaction_interval == 0 {
330                bail!(
331                    "`compaction-interval-sec` must be greater than 0 when `enable-compaction` is true"
332                );
333            }
334
335            tracing::info!(
336                "Alter config compaction_interval set to {} seconds",
337                compaction_interval
338            );
339        }
340
341        // Validate max snapshots
342        if let Some(max_snapshots) = iceberg_config.max_snapshots_num_before_compaction
343            && max_snapshots < 1
344        {
345            bail!(
346                "`compaction.max_snapshots_num` must be greater than 0, got: {}",
347                max_snapshots
348            );
349        }
350
351        // Validate target file size
352        if let Some(target_file_size_mb) = iceberg_config.target_file_size_mb
353            && target_file_size_mb == 0
354        {
355            bail!("`compaction.target_file_size_mb` must be greater than 0");
356        }
357
358        // Validate parquet max row group rows
359        if let Some(max_row_group_rows) = iceberg_config.write_parquet_max_row_group_rows
360            && max_row_group_rows == 0
361        {
362            bail!("`compaction.write_parquet_max_row_group_rows` must be greater than 0");
363        }
364
365        // Validate parquet max row group bytes
366        if let Some(max_row_group_bytes) = iceberg_config.write_parquet_max_row_group_bytes
367            && max_row_group_bytes == 0
368        {
369            bail!("`compaction.write_parquet_max_row_group_bytes` must be greater than 0");
370        }
371
372        // Validate parquet compression codec
373        if let Some(ref compression) = iceberg_config.write_parquet_compression {
374            let valid_codecs = [
375                "uncompressed",
376                "snappy",
377                "gzip",
378                "lzo",
379                "brotli",
380                "lz4",
381                "zstd",
382            ];
383            if !valid_codecs.contains(&compression.to_lowercase().as_str()) {
384                bail!(
385                    "`compaction.write_parquet_compression` must be one of {:?}, got: {}",
386                    valid_codecs,
387                    compression
388                );
389            }
390        }
391
392        Ok(())
393    }
394
395    async fn new_log_sinker(&self, writer_param: SinkWriterParam) -> Result<Self::LogSinker> {
396        let writer = IcebergSinkWriter::new(
397            self.config.clone(),
398            self.param.clone(),
399            writer_param.clone(),
400            self.upsert_primary_key_column_names.clone(),
401        );
402
403        let commit_checkpoint_interval =
404            NonZeroU64::new(self.config.commit_checkpoint_interval).expect(
405                "commit_checkpoint_interval should be greater than 0, and it should be checked in config validation",
406            );
407        let log_sinker = CoordinatedLogSinker::new(
408            &writer_param,
409            self.param.clone(),
410            writer,
411            commit_checkpoint_interval,
412        )
413        .await?;
414
415        Ok(log_sinker)
416    }
417
418    fn is_coordinated_sink(&self) -> bool {
419        true
420    }
421
422    async fn new_coordinator(
423        &self,
424        iceberg_compact_stat_sender: Option<UnboundedSender<IcebergSinkCompactionUpdate>>,
425    ) -> Result<SinkCommitCoordinator> {
426        let catalog = self.config.create_catalog().await?;
427        let table = self.create_and_validate_table().await?;
428        let coordinator = IcebergSinkCommitter {
429            catalog,
430            table,
431            last_commit_epoch: 0,
432            sink_id: self.param.sink_id,
433            config: self.config.clone(),
434            param: self.param.clone(),
435            commit_retry_num: self.config.commit_retry_num,
436            iceberg_compact_stat_sender,
437        };
438        if Self::is_exactly_once(&self.param.properties)? {
439            Ok(SinkCommitCoordinator::TwoPhase(Box::new(coordinator)))
440        } else {
441            Ok(SinkCommitCoordinator::SinglePhase(Box::new(coordinator)))
442        }
443    }
444}
risingwave_connector/sink/iceberg/mod.rs

risingwave_connector/sink/iceberg/
mod.rs
