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

risingwave_connector/sink/iceberg/
config.rs
