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

risingwave_connector/sink/iceberg/
config.rs
