laminar_connectors/lakehouse/iceberg_io/
append.rs

1use iceberg::table::Table;
2use iceberg::transaction::{ApplyTransactionAction, Transaction};
3use iceberg::Catalog;
4
5use crate::error::ConnectorError;
6
7use super::external_error_summary;
8
9pub(crate) enum GeneratedAppendError {
10    Preflight(ConnectorError),
11    Commit(iceberg::Error),
12}
13
14#[derive(Clone, Copy)]
15enum AppendPathProvenance {
16    Arbitrary,
17    WriterGenerated,
18}
19
20impl AppendPathProvenance {
21    const fn check_duplicates(self) -> bool {
22        matches!(self, Self::Arbitrary)
23    }
24}
25
26/// Append `data_files` in one Iceberg transaction.
27///
28/// # Errors
29/// Returns [`ConnectorError::OutcomeUnknown`] when the catalog does not acknowledge the commit.
30pub async fn commit_data_files_append(
31    table: &Table,
32    catalog: &dyn Catalog,
33    data_files: Vec<iceberg::spec::DataFile>,
34) -> Result<Table, ConnectorError> {
35    commit_append(table, catalog, data_files, AppendPathProvenance::Arbitrary)
36        .await
37        .map_err(|error| iceberg_commit_error(&error))
38}
39
40pub(crate) async fn commit_generated_data_files_append(
41    table: &Table,
42    catalog: &dyn Catalog,
43    data_files: Vec<iceberg::spec::DataFile>,
44    deadline: tokio::time::Instant,
45) -> Result<Table, GeneratedAppendError> {
46    // INVARIANT: callers use writer-generated paths in a unique deployment/epoch namespace.
47    super::super::iceberg_scan::preflight_current_snapshot_manifest_list(table, deadline)
48        .await
49        .map_err(GeneratedAppendError::Preflight)?;
50    commit_append(
51        table,
52        catalog,
53        data_files,
54        AppendPathProvenance::WriterGenerated,
55    )
56    .await
57    .map_err(GeneratedAppendError::Commit)
58}
59
60async fn commit_append(
61    table: &Table,
62    catalog: &dyn Catalog,
63    data_files: Vec<iceberg::spec::DataFile>,
64    path_provenance: AppendPathProvenance,
65) -> iceberg::Result<Table> {
66    let tx = Transaction::new(table);
67    let tx = if data_files.is_empty() {
68        tx
69    } else {
70        tx.fast_append()
71            .with_check_duplicate(path_provenance.check_duplicates())
72            .add_data_files(data_files)
73            .apply(tx)?
74    };
75    tx.commit(catalog).await
76}
77
78pub(crate) fn iceberg_commit_error(error: &iceberg::Error) -> ConnectorError {
79    use iceberg::ErrorKind;
80
81    let kind = error.kind();
82    let retryable = error.retryable();
83    let summary = external_error_summary(error);
84    match kind {
85        ErrorKind::CatalogCommitConflicts => {
86            ConnectorError::WriteError(format!("Iceberg catalog commit conflict ({summary})"))
87        }
88        ErrorKind::PreconditionFailed
89        | ErrorKind::DataInvalid
90        | ErrorKind::NamespaceAlreadyExists
91        | ErrorKind::TableAlreadyExists
92        | ErrorKind::NamespaceNotFound
93        | ErrorKind::TableNotFound
94        | ErrorKind::FeatureUnsupported => {
95            ConnectorError::TransactionError(format!("Iceberg catalog rejected commit ({summary})"))
96        }
97        ErrorKind::Unexpected => ConnectorError::outcome_unknown(
98            format!("Iceberg catalog commit may have applied ({summary})"),
99            retryable,
100        ),
101        _ => ConnectorError::outcome_unknown(
102            format!("Iceberg catalog returned an unclassified commit failure ({summary})"),
103            retryable,
104        ),
105    }
106}
laminar_connectors/lakehouse/iceberg_io/append.rs

laminar_connectors/lakehouse/iceberg_io/
append.rs
