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

risingwave_connector/sink/iceberg/
config.rs
