risingwave_connector/sink/iceberg/
config.rs

1// Copyright 2026 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, HashMap};
16use std::fmt::Debug;
17use std::sync::Arc;
18
19use anyhow::anyhow;
20use iceberg::spec::{FormatVersion, MAIN_BRANCH};
21use iceberg::table::Table;
22use iceberg::transaction::{MANIFEST_MIN_MERGE_COUNT_DEFAULT, MANIFEST_TARGET_SIZE_BYTES_DEFAULT};
23use iceberg::{Catalog, TableIdent};
24use parquet::basic::Compression;
25use serde::de::{self, Deserializer, Visitor};
26use serde::{Deserialize, Serialize};
27use serde_with::{DisplayFromStr, serde_as};
28use with_options::WithOptions;
29
30use super::{SINK_TYPE_APPEND_ONLY, SINK_TYPE_OPTION, SINK_TYPE_UPSERT, SinkError};
31use crate::connector_common::{
32    IcebergCatalogKind, IcebergCommon, IcebergTableIdentifier, ResolvedIcebergCatalogConfig,
33    iceberg_java_catalog_props_from_options,
34};
35use crate::enforce_secret::EnforceSecret;
36use crate::sink::Result;
37use crate::sink::decouple_checkpoint_log_sink::iceberg_default_commit_checkpoint_interval;
38use crate::{deserialize_bool_from_string, deserialize_optional_string_seq_from_string};
39
40pub const ICEBERG_COW_BRANCH: &str = "ingestion";
41
42pub const ICEBERG_WRITE_MODE_MERGE_ON_READ: &str = "merge-on-read";
43pub const ICEBERG_WRITE_MODE_COPY_ON_WRITE: &str = "copy-on-write";
44pub const ICEBERG_COMPACTION_TYPE_FULL: &str = "full";
45pub const ICEBERG_COMPACTION_TYPE_SMALL_FILES: &str = "small-files";
46pub const ICEBERG_COMPACTION_TYPE_FILES_WITH_DELETE: &str = "files-with-delete";
47
48pub const PARTITION_DATA_ID_START: i32 = 1000;
49
50#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Default, Serialize, Deserialize)]
51#[serde(rename_all = "kebab-case")]
52pub enum IcebergWriteMode {
53    #[default]
54    MergeOnRead,
55    CopyOnWrite,
56}
57
58impl IcebergWriteMode {
59    pub fn as_str(self) -> &'static str {
60        match self {
61            IcebergWriteMode::MergeOnRead => ICEBERG_WRITE_MODE_MERGE_ON_READ,
62            IcebergWriteMode::CopyOnWrite => ICEBERG_WRITE_MODE_COPY_ON_WRITE,
63        }
64    }
65}
66
67impl std::str::FromStr for IcebergWriteMode {
68    type Err = SinkError;
69
70    fn from_str(s: &str) -> std::result::Result<Self, Self::Err> {
71        match s {
72            ICEBERG_WRITE_MODE_MERGE_ON_READ => Ok(IcebergWriteMode::MergeOnRead),
73            ICEBERG_WRITE_MODE_COPY_ON_WRITE => Ok(IcebergWriteMode::CopyOnWrite),
74            _ => Err(SinkError::Config(anyhow!(format!(
75                "invalid write_mode: {}, must be one of: {}, {}",
76                s, ICEBERG_WRITE_MODE_MERGE_ON_READ, ICEBERG_WRITE_MODE_COPY_ON_WRITE
77            )))),
78        }
79    }
80}
81
82impl TryFrom<&str> for IcebergWriteMode {
83    type Error = <Self as std::str::FromStr>::Err;
84
85    fn try_from(value: &str) -> std::result::Result<Self, Self::Error> {
86        value.parse()
87    }
88}
89
90impl TryFrom<String> for IcebergWriteMode {
91    type Error = <Self as std::str::FromStr>::Err;
92
93    fn try_from(value: String) -> std::result::Result<Self, Self::Error> {
94        value.as_str().parse()
95    }
96}
97
98impl std::fmt::Display for IcebergWriteMode {
99    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
100        f.write_str(self.as_str())
101    }
102}
103
104// Configuration constants
105pub const ENABLE_COMPACTION: &str = "enable_compaction";
106pub const COMPACTION_INTERVAL_SEC: &str = "compaction_interval_sec";
107pub const ENABLE_SNAPSHOT_EXPIRATION: &str = "enable_snapshot_expiration";
108pub const WRITE_MODE: &str = "write_mode";
109pub const FORMAT_VERSION: &str = "format_version";
110pub const SNAPSHOT_EXPIRATION_RETAIN_LAST: &str = "snapshot_expiration_retain_last";
111pub const SNAPSHOT_EXPIRATION_MAX_AGE_MILLIS: &str = "snapshot_expiration_max_age_millis";
112pub const SNAPSHOT_EXPIRATION_CLEAR_EXPIRED_FILES: &str = "snapshot_expiration_clear_expired_files";
113pub const SNAPSHOT_EXPIRATION_CLEAR_EXPIRED_META_DATA: &str =
114    "snapshot_expiration_clear_expired_meta_data";
115pub const ENABLE_MANIFEST_REWRITE: &str = "enable_manifest_rewrite";
116pub const MANIFEST_REWRITE_TARGET_SIZE_BYTES: &str = "manifest_rewrite_target_size_bytes";
117pub const MANIFEST_REWRITE_MIN_COUNT_TO_MERGE: &str = "manifest_rewrite_min_count_to_merge";
118pub const COMPACTION_MAX_SNAPSHOTS_NUM: &str = "compaction.max_snapshots_num";
119
120pub const COMPACTION_SMALL_FILES_THRESHOLD_MB: &str = "compaction.small_files_threshold_mb";
121
122pub const COMPACTION_DELETE_FILES_COUNT_THRESHOLD: &str = "compaction.delete_files_count_threshold";
123
124pub const COMPACTION_TRIGGER_SNAPSHOT_COUNT: &str = "compaction.trigger_snapshot_count";
125
126pub const COMPACTION_TARGET_FILE_SIZE_MB: &str = "compaction.target_file_size_mb";
127
128pub const COMPACTION_TYPE: &str = "compaction.type";
129
130pub const COMPACTION_WRITE_PARQUET_COMPRESSION: &str = "compaction.write_parquet_compression";
131pub const COMPACTION_WRITE_PARQUET_MAX_ROW_GROUP_ROWS: &str =
132    "compaction.write_parquet_max_row_group_rows";
133pub const COMPACTION_WRITE_PARQUET_MAX_ROW_GROUP_BYTES: &str =
134    "compaction.write_parquet_max_row_group_bytes";
135pub const ORDER_KEY: &str = "order_key";
136pub const DEFAULT_COMPACTION_MAX_SNAPSHOTS_NUM: usize = 1000;
137pub const ICEBERG_DEFAULT_WRITE_PARQUET_MAX_ROW_GROUP_BYTES: usize = 128 * 1024 * 1024;
138pub const ENABLE_PK_INDEX: &str = "enable_pk_index";
139
140pub const PARQUET_CREATED_BY: &str = concat!("risingwave version ", env!("CARGO_PKG_VERSION"));
141
142fn default_commit_retry_num() -> u32 {
143    8
144}
145
146fn default_iceberg_write_mode() -> IcebergWriteMode {
147    IcebergWriteMode::MergeOnRead
148}
149
150fn default_iceberg_format_version() -> FormatVersion {
151    FormatVersion::V2
152}
153
154fn default_true() -> bool {
155    true
156}
157
158fn default_some_true() -> Option<bool> {
159    Some(true)
160}
161
162fn parse_format_version_str(value: &str) -> std::result::Result<FormatVersion, String> {
163    let parsed = value
164        .trim()
165        .parse::<u8>()
166        .map_err(|_| "`format-version` must be one of 1, 2, or 3".to_owned())?;
167    match parsed {
168        1 => Ok(FormatVersion::V1),
169        2 => Ok(FormatVersion::V2),
170        3 => Ok(FormatVersion::V3),
171        _ => Err("`format-version` must be one of 1, 2, or 3".to_owned()),
172    }
173}
174
175fn deserialize_format_version<'de, D>(
176    deserializer: D,
177) -> std::result::Result<FormatVersion, D::Error>
178where
179    D: Deserializer<'de>,
180{
181    struct FormatVersionVisitor;
182
183    impl<'de> Visitor<'de> for FormatVersionVisitor {
184        type Value = FormatVersion;
185
186        fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
187            formatter.write_str("format-version as 1, 2, or 3")
188        }
189
190        fn visit_u64<E>(self, value: u64) -> std::result::Result<Self::Value, E>
191        where
192            E: de::Error,
193        {
194            let value = u8::try_from(value)
195                .map_err(|_| E::custom("`format-version` must be one of 1, 2, or 3"))?;
196            parse_format_version_str(&value.to_string()).map_err(E::custom)
197        }
198
199        fn visit_i64<E>(self, value: i64) -> std::result::Result<Self::Value, E>
200        where
201            E: de::Error,
202        {
203            let value = u8::try_from(value)
204                .map_err(|_| E::custom("`format-version` must be one of 1, 2, or 3"))?;
205            parse_format_version_str(&value.to_string()).map_err(E::custom)
206        }
207
208        fn visit_str<E>(self, value: &str) -> std::result::Result<Self::Value, E>
209        where
210            E: de::Error,
211        {
212            parse_format_version_str(value).map_err(E::custom)
213        }
214
215        fn visit_string<E>(self, value: String) -> std::result::Result<Self::Value, E>
216        where
217            E: de::Error,
218        {
219            self.visit_str(&value)
220        }
221    }
222
223    deserializer.deserialize_any(FormatVersionVisitor)
224}
225
226/// Compaction type for Iceberg sink
227#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
228#[serde(rename_all = "kebab-case")]
229pub enum CompactionType {
230    /// Full compaction - rewrites all data files
231    #[default]
232    Full,
233    /// Small files compaction - only compact small files
234    SmallFiles,
235    /// Files with delete compaction - only compact files that have associated delete files
236    FilesWithDelete,
237}
238
239impl CompactionType {
240    pub fn as_str(&self) -> &'static str {
241        match self {
242            CompactionType::Full => ICEBERG_COMPACTION_TYPE_FULL,
243            CompactionType::SmallFiles => ICEBERG_COMPACTION_TYPE_SMALL_FILES,
244            CompactionType::FilesWithDelete => ICEBERG_COMPACTION_TYPE_FILES_WITH_DELETE,
245        }
246    }
247}
248
249impl std::str::FromStr for CompactionType {
250    type Err = SinkError;
251
252    fn from_str(s: &str) -> std::result::Result<Self, Self::Err> {
253        match s {
254            ICEBERG_COMPACTION_TYPE_FULL => Ok(CompactionType::Full),
255            ICEBERG_COMPACTION_TYPE_SMALL_FILES => Ok(CompactionType::SmallFiles),
256            ICEBERG_COMPACTION_TYPE_FILES_WITH_DELETE => Ok(CompactionType::FilesWithDelete),
257            _ => Err(SinkError::Config(anyhow!(format!(
258                "invalid compaction_type: {}, must be one of: {}, {}, {}",
259                s,
260                ICEBERG_COMPACTION_TYPE_FULL,
261                ICEBERG_COMPACTION_TYPE_SMALL_FILES,
262                ICEBERG_COMPACTION_TYPE_FILES_WITH_DELETE
263            )))),
264        }
265    }
266}
267
268impl TryFrom<&str> for CompactionType {
269    type Error = <Self as std::str::FromStr>::Err;
270
271    fn try_from(value: &str) -> std::result::Result<Self, Self::Error> {
272        value.parse()
273    }
274}
275
276impl TryFrom<String> for CompactionType {
277    type Error = <Self as std::str::FromStr>::Err;
278
279    fn try_from(value: String) -> std::result::Result<Self, Self::Error> {
280        value.as_str().parse()
281    }
282}
283
284impl std::fmt::Display for CompactionType {
285    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
286        write!(f, "{}", self.as_str())
287    }
288}
289
290#[serde_as]
291#[derive(Debug, Clone, PartialEq, Eq, Deserialize, WithOptions)]
292pub struct IcebergConfig {
293    pub r#type: String, // accept "append-only" or "upsert"
294
295    #[serde(default, deserialize_with = "deserialize_bool_from_string")]
296    pub force_append_only: bool,
297
298    #[serde(flatten)]
299    pub(crate) common: IcebergCommon,
300
301    #[serde(flatten)]
302    pub(crate) table: IcebergTableIdentifier,
303
304    #[serde(
305        rename = "primary_key",
306        default,
307        deserialize_with = "deserialize_optional_string_seq_from_string"
308    )]
309    pub primary_key: Option<Vec<String>>,
310
311    // Props for java catalog props.
312    #[serde(skip)]
313    pub java_catalog_props: HashMap<String, String>,
314
315    #[serde(default)]
316    #[with_option(iceberg_engine)]
317    pub partition_by: Option<String>,
318
319    #[serde(default)]
320    #[with_option(iceberg_engine)]
321    pub order_key: Option<String>,
322
323    /// Commit every n(>0) checkpoints, default is 60.
324    #[serde(default = "iceberg_default_commit_checkpoint_interval")]
325    #[serde_as(as = "DisplayFromStr")]
326    #[with_option(allow_alter_on_fly, iceberg_engine)]
327    pub commit_checkpoint_interval: u64,
328
329    #[serde(default, deserialize_with = "deserialize_bool_from_string")]
330    pub create_table_if_not_exists: bool,
331
332    /// Whether it is `exactly_once`, the default is true.
333    #[serde(default = "default_some_true")]
334    #[serde_as(as = "Option<DisplayFromStr>")]
335    pub is_exactly_once: Option<bool>,
336    // Retry commit num when iceberg commit fail. default is 8.
337    // # TODO
338    // Iceberg table may store the retry commit num in table meta.
339    // We should try to find and use that as default commit retry num first.
340    #[serde(default = "default_commit_retry_num")]
341    pub commit_retry_num: u32,
342
343    /// Whether to enable iceberg compaction.
344    #[serde(
345        rename = "enable_compaction",
346        default,
347        deserialize_with = "deserialize_bool_from_string"
348    )]
349    #[with_option(allow_alter_on_fly, iceberg_engine)]
350    pub enable_compaction: bool,
351
352    /// The interval of iceberg compaction
353    #[serde(rename = "compaction_interval_sec", default)]
354    #[serde_as(as = "Option<DisplayFromStr>")]
355    #[with_option(allow_alter_on_fly, iceberg_engine)]
356    pub compaction_interval_sec: Option<u64>,
357
358    /// Whether to enable iceberg expired snapshots.
359    #[serde(
360        rename = "enable_snapshot_expiration",
361        default = "default_true",
362        deserialize_with = "deserialize_bool_from_string"
363    )]
364    #[with_option(allow_alter_on_fly, iceberg_engine)]
365    pub enable_snapshot_expiration: bool,
366
367    /// The iceberg write mode, can be `merge-on-read` or `copy-on-write`.
368    #[serde(rename = "write_mode", default = "default_iceberg_write_mode")]
369    #[with_option(iceberg_engine)]
370    pub write_mode: IcebergWriteMode,
371
372    /// Iceberg format version for table creation.
373    #[serde(
374        rename = "format_version",
375        default = "default_iceberg_format_version",
376        deserialize_with = "deserialize_format_version"
377    )]
378    #[with_option(iceberg_engine)]
379    pub format_version: FormatVersion,
380
381    /// The maximum age (in milliseconds) for snapshots before they expire
382    /// For example, if set to 3600000, snapshots older than 1 hour will be expired
383    #[serde(rename = "snapshot_expiration_max_age_millis", default)]
384    #[serde_as(as = "Option<DisplayFromStr>")]
385    #[with_option(allow_alter_on_fly, iceberg_engine)]
386    pub snapshot_expiration_max_age_millis: Option<i64>,
387
388    /// The number of snapshots to retain
389    #[serde(rename = "snapshot_expiration_retain_last", default)]
390    #[serde_as(as = "Option<DisplayFromStr>")]
391    #[with_option(allow_alter_on_fly, iceberg_engine)]
392    pub snapshot_expiration_retain_last: Option<i32>,
393
394    #[serde(
395        rename = "snapshot_expiration_clear_expired_files",
396        default = "default_true",
397        deserialize_with = "deserialize_bool_from_string"
398    )]
399    #[with_option(allow_alter_on_fly, iceberg_engine)]
400    pub snapshot_expiration_clear_expired_files: bool,
401
402    #[serde(
403        rename = "snapshot_expiration_clear_expired_meta_data",
404        default = "default_true",
405        deserialize_with = "deserialize_bool_from_string"
406    )]
407    #[with_option(allow_alter_on_fly, iceberg_engine)]
408    pub snapshot_expiration_clear_expired_meta_data: bool,
409
410    /// Whether to periodically rewrite fragmented Iceberg data manifests.
411    #[serde(
412        rename = "enable_manifest_rewrite",
413        default,
414        deserialize_with = "deserialize_bool_from_string"
415    )]
416    #[with_option(allow_alter_on_fly, iceberg_engine)]
417    pub enable_manifest_rewrite: bool,
418
419    /// Target size in bytes for rewritten Iceberg data manifests.
420    #[serde(rename = "manifest_rewrite_target_size_bytes", default)]
421    #[serde_as(as = "Option<DisplayFromStr>")]
422    #[with_option(allow_alter_on_fly, iceberg_engine)]
423    pub manifest_rewrite_target_size_bytes: Option<u64>,
424
425    /// Minimum number of manifests required to rewrite an under-filled batch.
426    #[serde(rename = "manifest_rewrite_min_count_to_merge", default)]
427    #[serde_as(as = "Option<DisplayFromStr>")]
428    #[with_option(allow_alter_on_fly, iceberg_engine)]
429    pub manifest_rewrite_min_count_to_merge: Option<usize>,
430
431    /// The maximum number of snapshots allowed since the last rewrite operation
432    /// If set, sink will check snapshot count and wait if exceeded
433    /// If unset, defaults to 1000 only when compaction is enabled
434    #[serde(rename = "compaction.max_snapshots_num", default)]
435    #[serde_as(as = "Option<DisplayFromStr>")]
436    #[with_option(allow_alter_on_fly, iceberg_engine)]
437    pub max_snapshots_num_before_compaction: Option<usize>,
438
439    #[serde(rename = "compaction.small_files_threshold_mb", default)]
440    #[serde_as(as = "Option<DisplayFromStr>")]
441    #[with_option(allow_alter_on_fly, iceberg_engine)]
442    pub small_files_threshold_mb: Option<u64>,
443
444    #[serde(rename = "compaction.delete_files_count_threshold", default)]
445    #[serde_as(as = "Option<DisplayFromStr>")]
446    #[with_option(allow_alter_on_fly, iceberg_engine)]
447    pub delete_files_count_threshold: Option<usize>,
448
449    #[serde(rename = "compaction.trigger_snapshot_count", default)]
450    #[serde_as(as = "Option<DisplayFromStr>")]
451    #[with_option(allow_alter_on_fly, iceberg_engine)]
452    pub trigger_snapshot_count: Option<usize>,
453
454    #[serde(rename = "compaction.target_file_size_mb", default)]
455    #[serde_as(as = "Option<DisplayFromStr>")]
456    #[with_option(allow_alter_on_fly, iceberg_engine)]
457    pub target_file_size_mb: Option<u64>,
458
459    /// Compaction type: `full`, `small-files`, or `files-with-delete`
460    /// If not set, will default to `full`
461    #[serde(rename = "compaction.type", default)]
462    #[with_option(allow_alter_on_fly, iceberg_engine)]
463    pub compaction_type: Option<CompactionType>,
464
465    /// Parquet compression codec
466    /// Supported values: uncompressed, snappy, gzip, lzo, brotli, lz4, zstd
467    /// Default is zstd
468    #[serde(rename = "compaction.write_parquet_compression", default)]
469    #[with_option(allow_alter_on_fly, iceberg_engine)]
470    pub write_parquet_compression: Option<String>,
471
472    /// Deprecated: maximum number of rows in a Parquet row group.
473    /// Accepted for backward compatibility, but ignored by the writer.
474    #[serde(rename = "compaction.write_parquet_max_row_group_rows", default)]
475    #[serde_as(as = "Option<DisplayFromStr>")]
476    #[with_option(allow_alter_on_fly, iceberg_engine)]
477    pub write_parquet_max_row_group_rows: Option<usize>,
478
479    /// Maximum size of a Parquet row group in bytes
480    /// Default is 128 `MiB`, matching Iceberg defaults.
481    #[serde(rename = "compaction.write_parquet_max_row_group_bytes", default)]
482    #[serde_as(as = "Option<DisplayFromStr>")]
483    #[with_option(allow_alter_on_fly, iceberg_engine)]
484    pub write_parquet_max_row_group_bytes: Option<usize>,
485
486    /// Whether to enable PK index for upsert sink. Default is false.
487    /// For upsert iceberg sinks (V2/V3, merge-on-read): maintain a pk index and write
488    /// position deletes instead of equality deletes.
489    #[serde(
490        rename = "enable_pk_index",
491        default,
492        deserialize_with = "deserialize_bool_from_string"
493    )]
494    #[with_option(iceberg_engine)]
495    pub enable_pk_index: bool,
496
497    #[serde(flatten)]
498    pub unknown_fields: std::collections::HashMap<String, String>,
499}
500
501crate::impl_sink_unknown_fields!(IcebergConfig);
502
503impl EnforceSecret for IcebergConfig {
504    fn enforce_secret<'a>(
505        prop_iter: impl Iterator<Item = &'a str>,
506    ) -> crate::error::ConnectorResult<()> {
507        for prop in prop_iter {
508            IcebergCommon::enforce_one(prop)?;
509        }
510        Ok(())
511    }
512
513    fn enforce_one(prop: &str) -> crate::error::ConnectorResult<()> {
514        IcebergCommon::enforce_one(prop)
515    }
516}
517
518impl IcebergConfig {
519    /// Validate that append-only sinks use merge-on-read mode
520    /// Copy-on-write is strictly worse than merge-on-read for append-only workloads
521    pub fn validate_append_only_write_mode(
522        sink_type: &str,
523        write_mode: IcebergWriteMode,
524    ) -> Result<()> {
525        if sink_type == SINK_TYPE_APPEND_ONLY && write_mode == IcebergWriteMode::CopyOnWrite {
526            return Err(SinkError::Config(anyhow!(
527                "'copy-on-write' mode is not supported for append-only iceberg sink. \
528                 Please use 'merge-on-read' instead, which is strictly better for append-only workloads."
529            )));
530        }
531        Ok(())
532    }
533
534    pub(crate) fn validate_enable_pk_index(&self) -> Result<()> {
535        if !self.enable_pk_index {
536            return Ok(());
537        }
538
539        if self.r#type != SINK_TYPE_UPSERT {
540            return Err(SinkError::Config(anyhow!(
541                "`enable_pk_index` is only supported for upsert iceberg sink"
542            )));
543        }
544
545        if self.write_mode != IcebergWriteMode::MergeOnRead {
546            return Err(SinkError::Config(anyhow!(
547                "`enable_pk_index` is only supported for upsert iceberg sink with merge-on-read mode"
548            )));
549        }
550
551        if self.format_version < FormatVersion::V2 {
552            return Err(SinkError::Config(anyhow!(
553                "`enable_pk_index` is only supported for upsert iceberg sink with format version >= 2"
554            )));
555        }
556
557        if self.force_append_only {
558            return Err(SinkError::Config(anyhow!(
559                "`enable_pk_index` cannot be true when `force_append_only` is true"
560            )));
561        }
562
563        Ok(())
564    }
565
566    pub fn from_btreemap(values: BTreeMap<String, String>) -> Result<Self> {
567        let mut config =
568            serde_json::from_value::<IcebergConfig>(serde_json::to_value(&values).unwrap())
569                .map_err(|e| SinkError::Config(anyhow!(e)))?;
570
571        if config.enable_compaction && !values.contains_key(COMPACTION_MAX_SNAPSHOTS_NUM) {
572            config.max_snapshots_num_before_compaction = Some(DEFAULT_COMPACTION_MAX_SNAPSHOTS_NUM);
573        }
574
575        if config.r#type != SINK_TYPE_APPEND_ONLY && config.r#type != SINK_TYPE_UPSERT {
576            return Err(SinkError::Config(anyhow!(
577                "`{}` must be {}, or {}",
578                SINK_TYPE_OPTION,
579                SINK_TYPE_APPEND_ONLY,
580                SINK_TYPE_UPSERT
581            )));
582        }
583
584        if config.r#type == SINK_TYPE_UPSERT {
585            if let Some(primary_key) = &config.primary_key {
586                if primary_key.is_empty() {
587                    return Err(SinkError::Config(anyhow!(
588                        "`primary-key` must not be empty in {}",
589                        SINK_TYPE_UPSERT
590                    )));
591                }
592            } else if !config.enable_pk_index {
593                // When `enable_pk_index = true`, the planner auto-derives the iceberg pk
594                // from the upstream stream key, so the user does not need to spell it out
595                // in WITH options. The derived pk is written back into properties before
596                // this validation is consulted again at sink-construction time.
597                return Err(SinkError::Config(anyhow!(
598                    "Must set `primary-key` in {}",
599                    SINK_TYPE_UPSERT
600                )));
601            }
602        }
603
604        // Enforce merge-on-read for append-only sinks
605        Self::validate_append_only_write_mode(&config.r#type, config.write_mode)?;
606        config.validate_enable_pk_index()?;
607        config.validate_manifest_rewrite_format(config.format_version)?;
608
609        // All configs start with "catalog." will be treated as java configs.
610        config.java_catalog_props = iceberg_java_catalog_props_from_options(
611            values
612                .iter()
613                .map(|(key, value)| (key.as_str(), value.as_str())),
614        );
615        config
616            .unknown_fields
617            .retain(|key, _| key != "connector" && !key.starts_with("catalog."));
618
619        if config.commit_checkpoint_interval == 0 {
620            return Err(SinkError::Config(anyhow!(
621                "`commit-checkpoint-interval` must be greater than 0"
622            )));
623        }
624
625        if config.compaction_interval_sec == Some(0) {
626            return Err(SinkError::Config(anyhow!(
627                "`compaction_interval_sec` must be greater than 0"
628            )));
629        }
630
631        if config.trigger_snapshot_count == Some(0) {
632            return Err(SinkError::Config(anyhow!(
633                "`compaction.trigger_snapshot_count` must be greater than 0"
634            )));
635        }
636
637        if config.max_snapshots_num_before_compaction == Some(0) {
638            return Err(SinkError::Config(anyhow!(
639                "`compaction.max_snapshots_num` must be greater than 0"
640            )));
641        }
642
643        if config.small_files_threshold_mb == Some(0) {
644            return Err(SinkError::Config(anyhow!(
645                "`compaction.small_files_threshold_mb` must be greater than 0"
646            )));
647        }
648
649        if config.delete_files_count_threshold == Some(0) {
650            return Err(SinkError::Config(anyhow!(
651                "`compaction.delete_files_count_threshold` must be greater than 0"
652            )));
653        }
654
655        if config.target_file_size_mb == Some(0) {
656            return Err(SinkError::Config(anyhow!(
657                "`compaction.target_file_size_mb` must be greater than 0"
658            )));
659        }
660
661        if config.write_parquet_max_row_group_rows == Some(0) {
662            return Err(SinkError::Config(anyhow!(
663                "`compaction.write_parquet_max_row_group_rows` must be greater than 0"
664            )));
665        }
666
667        if config.write_parquet_max_row_group_bytes == Some(0) {
668            return Err(SinkError::Config(anyhow!(
669                "`compaction.write_parquet_max_row_group_bytes` must be greater than 0"
670            )));
671        }
672
673        if config.manifest_rewrite_target_size_bytes == Some(0) {
674            return Err(SinkError::Config(anyhow!(
675                "`manifest_rewrite_target_size_bytes` must be greater than 0"
676            )));
677        }
678
679        if config.manifest_rewrite_min_count_to_merge == Some(0) {
680            return Err(SinkError::Config(anyhow!(
681                "`manifest_rewrite_min_count_to_merge` must be greater than 0"
682            )));
683        }
684
685        // Validate table identifier (e.g., database.name should not contain dots)
686        config
687            .table
688            .validate()
689            .map_err(|e| SinkError::Config(anyhow!(e)))?;
690
691        if config.write_parquet_max_row_group_rows.is_some() {
692            tracing::warn!(
693                "`compaction.write_parquet_max_row_group_rows` is deprecated and ignored; use `compaction.write_parquet_max_row_group_bytes` instead"
694            );
695        }
696
697        Ok(config)
698    }
699
700    pub fn catalog_type(&self) -> &str {
701        self.common.catalog_type()
702    }
703
704    pub fn catalog_kind(&self) -> Result<IcebergCatalogKind> {
705        self.common
706            .resolve_catalog_kind()
707            .map_err(|err| SinkError::Config(anyhow!(err)))
708    }
709
710    fn resolved_catalog_config(&self) -> Result<ResolvedIcebergCatalogConfig<'_>> {
711        self.common
712            .resolve_catalog_config(self.java_catalog_props.clone())
713            .map_err(|err| SinkError::Config(anyhow!(err)))
714    }
715
716    pub async fn load_table(&self) -> Result<Table> {
717        #[cfg(any(test, madsim))]
718        if self.catalog_type() == "mock_v3" {
719            let catalog =
720                crate::sink::iceberg::mock_v3_catalog_registry::get().ok_or_else(|| {
721                    SinkError::Config(anyhow!(
722                        "mock_v3 catalog_type set but no mock catalog registered"
723                    ))
724                })?;
725            let table_id = self
726                .table
727                .to_table_ident()
728                .map_err(|e| SinkError::Config(anyhow!(e).context("Unable to parse table name")))?;
729            let table = catalog
730                .load_table(&table_id)
731                .await
732                .map_err(|e| SinkError::Config(anyhow!(e).context("Failed to load mock table")))?;
733            return Ok(table);
734        }
735        self.resolved_catalog_config()?
736            .load_table(&self.table)
737            .await
738            .map_err(Into::into)
739    }
740
741    pub async fn create_catalog(&self) -> Result<Arc<dyn Catalog>> {
742        #[cfg(any(test, madsim))]
743        if self.catalog_type() == "mock_v3" {
744            return Ok(
745                crate::sink::iceberg::mock_v3_catalog_registry::get().ok_or_else(|| {
746                    anyhow::anyhow!("mock_v3 catalog_type set but no mock catalog registered")
747                })?,
748            );
749        }
750        self.resolved_catalog_config()?
751            .create_catalog()
752            .await
753            .map_err(Into::into)
754    }
755
756    pub fn full_table_name(&self) -> Result<TableIdent> {
757        self.table.to_table_ident().map_err(Into::into)
758    }
759
760    pub fn catalog_name(&self) -> String {
761        self.common.catalog_name()
762    }
763
764    pub fn table_format_version(&self) -> FormatVersion {
765        self.format_version
766    }
767
768    pub fn validate_manifest_rewrite_format(&self, format_version: FormatVersion) -> Result<()> {
769        if self.enable_manifest_rewrite && format_version >= FormatVersion::V3 {
770            return Err(SinkError::Config(anyhow!(
771                "`enable_manifest_rewrite` is not supported for Iceberg format version 3 because rewrite manifests cannot preserve row lineage"
772            )));
773        }
774        Ok(())
775    }
776
777    pub fn compaction_interval_sec(&self) -> u64 {
778        // default to 1 hour
779        self.compaction_interval_sec.unwrap_or(3600)
780    }
781
782    /// Calculate the timestamp (in milliseconds) before which snapshots should be expired
783    /// Returns `current_time_ms` - `max_age_millis`
784    pub fn snapshot_expiration_timestamp_ms(&self, current_time_ms: i64) -> Option<i64> {
785        self.snapshot_expiration_max_age_millis
786            .map(|max_age_millis| current_time_ms - max_age_millis)
787    }
788
789    pub fn manifest_rewrite_target_size_bytes(&self) -> u64 {
790        self.manifest_rewrite_target_size_bytes
791            .unwrap_or(MANIFEST_TARGET_SIZE_BYTES_DEFAULT as u64)
792    }
793
794    pub fn manifest_rewrite_min_count_to_merge(&self) -> usize {
795        self.manifest_rewrite_min_count_to_merge
796            .unwrap_or(MANIFEST_MIN_MERGE_COUNT_DEFAULT as usize)
797    }
798
799    pub fn trigger_snapshot_count(&self) -> usize {
800        self.trigger_snapshot_count.unwrap_or(usize::MAX)
801    }
802
803    pub fn small_files_threshold_mb(&self) -> u64 {
804        self.small_files_threshold_mb.unwrap_or(64)
805    }
806
807    pub fn delete_files_count_threshold(&self) -> usize {
808        self.delete_files_count_threshold.unwrap_or(256)
809    }
810
811    pub fn target_file_size_mb(&self) -> u64 {
812        self.target_file_size_mb.unwrap_or(1024)
813    }
814
815    /// Get the compaction type as an enum
816    /// This method parses the string and returns the enum value
817    pub fn compaction_type(&self) -> CompactionType {
818        self.compaction_type.unwrap_or_default()
819    }
820
821    /// Get the parquet compression codec
822    /// Default is "zstd"
823    pub fn write_parquet_compression(&self) -> &str {
824        self.write_parquet_compression.as_deref().unwrap_or("zstd")
825    }
826
827    /// Get the maximum number of rows in a Parquet row group.
828    pub fn write_parquet_max_row_group_rows(&self) -> Option<usize> {
829        self.write_parquet_max_row_group_rows
830    }
831
832    /// Get the maximum size in bytes of a Parquet row group.
833    pub fn write_parquet_max_row_group_bytes(&self) -> Option<usize> {
834        self.write_parquet_max_row_group_bytes
835            .or(Some(ICEBERG_DEFAULT_WRITE_PARQUET_MAX_ROW_GROUP_BYTES))
836    }
837
838    /// Parse the compression codec string into Parquet Compression enum.
839    /// Invalid values fall back to SNAPPY.
840    pub fn get_parquet_compression(&self) -> Compression {
841        parse_parquet_compression(self.write_parquet_compression())
842    }
843}
844
845/// Parse compression codec string to Parquet Compression enum
846pub fn parse_parquet_compression(codec: &str) -> Compression {
847    match codec.to_lowercase().as_str() {
848        "uncompressed" => Compression::UNCOMPRESSED,
849        "snappy" => Compression::SNAPPY,
850        "gzip" => Compression::GZIP(Default::default()),
851        "lzo" => Compression::LZO,
852        "brotli" => Compression::BROTLI(Default::default()),
853        "lz4" => Compression::LZ4,
854        "zstd" => Compression::ZSTD(Default::default()),
855        _ => {
856            tracing::warn!(
857                "Unknown compression codec '{}', falling back to SNAPPY",
858                codec
859            );
860            Compression::SNAPPY
861        }
862    }
863}
864
865// Helper Functions
866
867pub fn commit_branch(sink_type: &str, write_mode: IcebergWriteMode) -> String {
868    if should_enable_iceberg_cow(sink_type, write_mode) {
869        ICEBERG_COW_BRANCH.to_owned()
870    } else {
871        MAIN_BRANCH.to_owned()
872    }
873}
874
875pub fn should_enable_iceberg_cow(sink_type: &str, write_mode: IcebergWriteMode) -> bool {
876    sink_type == SINK_TYPE_UPSERT && write_mode == IcebergWriteMode::CopyOnWrite
877}
878
879impl crate::with_options::WithOptions for IcebergWriteMode {}
880
881impl crate::with_options::WithOptions for FormatVersion {}
882
883impl crate::with_options::WithOptions for CompactionType {}
risingwave_connector/sink/iceberg/config.rs

risingwave_connector/sink/iceberg/
config.rs
