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

risingwave_connector/sink/iceberg/
config.rs
