laminar_connectors/lakehouse/iceberg/
coordinated_commit.rs

1use async_trait::async_trait;
2
3use crate::connector::{
4    CoordinatedCommitBatch, CoordinatedCommitContext, CoordinatedCommitCursor,
5    CoordinatedCommitNamespace, CoordinatedCommitter,
6};
7use crate::error::ConnectorError;
8
9use super::{publication, IcebergSink};
10
11#[async_trait]
12impl CoordinatedCommitter for IcebergSink {
13    async fn commit_aggregated(
14        &self,
15        batch: CoordinatedCommitBatch,
16        context: CoordinatedCommitContext,
17    ) -> Result<(), ConnectorError> {
18        let catalog = self
19            .catalog
20            .as_ref()
21            .ok_or_else(|| ConnectorError::InvalidState {
22                expected: "open Iceberg sink".into(),
23                actual: "catalog is not initialized".into(),
24            })?;
25        let pending = publication::unresolved_publication(&self.config, &batch)?;
26        {
27            let mut unresolved = self.unresolved_publication.lock();
28            if unresolved
29                .as_ref()
30                .is_some_and(|existing| existing != &pending)
31            {
32                return Err(ConnectorError::TransactionError(
33                    "Iceberg has a different unresolved publication; only that exact cut may be reconciled"
34                        .into(),
35                ));
36            }
37            *unresolved = Some(pending.clone());
38        }
39        let result = publication::publish_coordinated(
40            catalog,
41            &self.catalog_capabilities,
42            &self.catalog_session,
43            &self.config,
44            &batch,
45            context,
46            &self.metrics,
47        )
48        .await;
49        if result.is_ok()
50            || result
51                .as_ref()
52                .is_err_and(|error| !error.is_outcome_unknown())
53        {
54            let mut unresolved = self.unresolved_publication.lock();
55            if unresolved.as_ref() == Some(&pending) {
56                *unresolved = None;
57            }
58        }
59        result
60    }
61
62    async fn committed_cursor(
63        &self,
64        namespace: &CoordinatedCommitNamespace,
65    ) -> Result<Option<CoordinatedCommitCursor>, ConnectorError> {
66        namespace.validate()?;
67        let catalog = self
68            .catalog
69            .as_ref()
70            .ok_or_else(|| ConnectorError::InvalidState {
71                expected: "open Iceberg sink".into(),
72                actual: "catalog is not initialized".into(),
73            })?;
74        let deadline = crate::lakehouse::iceberg_io::checked_deadline(
75            self.config.catalog.request_timeout,
76            "catalog.request_timeout",
77        )?;
78        let pending = self.unresolved_publication.lock().clone();
79        let cursor = publication::read_committed_cursor(
80            catalog,
81            &self.config,
82            namespace,
83            deadline,
84            &self.metrics,
85            pending.as_ref(),
86        )
87        .await?;
88        let mut unresolved = self.unresolved_publication.lock();
89        if unresolved.as_ref().is_some_and(|pending| {
90            pending.external_key == namespace.external_key() && pending.reconciled_by(cursor)
91        }) {
92            *unresolved = None;
93        }
94        Ok(cursor)
95    }
96}
laminar_connectors/lakehouse/iceberg/coordinated_commit.rs

laminar_connectors/lakehouse/iceberg/
coordinated_commit.rs
