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

risingwave_connector/sink/iceberg/
config.rs
