laminar_connectors/lakehouse/iceberg/
mod.rs

1//! Apache Iceberg append sink connector.
2
3#[cfg(feature = "iceberg-core")]
4mod aborted_cleanup;
5#[cfg(feature = "iceberg-core")]
6mod artifact_inventory;
7pub(crate) mod capabilities;
8#[cfg(feature = "iceberg-core")]
9mod commit_cursor;
10#[cfg(feature = "iceberg-core")]
11mod coordinated_commit;
12#[cfg(feature = "iceberg-core")]
13mod descriptor;
14#[cfg(feature = "iceberg-core")]
15mod descriptor_batch;
16#[cfg(feature = "iceberg-core")]
17mod direct_publication;
18#[cfg(feature = "iceberg-core")]
19mod epoch_intent;
20#[cfg(feature = "iceberg-core")]
21mod epoch_writer;
22#[cfg(all(test, feature = "iceberg-core"))]
23mod fault_injection;
24#[cfg(feature = "iceberg-core")]
25mod file_finalizer;
26#[cfg(feature = "iceberg-core")]
27mod fingerprint;
28#[cfg(feature = "iceberg-core")]
29mod metrics;
30#[cfg(feature = "iceberg-core")]
31mod publication;
32#[cfg(all(test, feature = "iceberg-core"))]
33mod recovery_tests;
34#[cfg(feature = "iceberg-core")]
35mod schema_alignment;
36#[cfg(feature = "iceberg-core")]
37mod startup;
38#[cfg(all(test, feature = "iceberg-core"))]
39pub(crate) mod test_support;
40
41use std::sync::Arc;
42use std::time::Duration;
43
44use arrow_array::RecordBatch;
45use arrow_schema::SchemaRef;
46use async_trait::async_trait;
47#[cfg(feature = "iceberg-core")]
48use parking_lot::Mutex;
49
50use crate::config::{ConnectorConfig, ConnectorState};
51use crate::connector::{
52    DeliveryGuarantee, SinkConnector, SinkConsistency, SinkContract, SinkInputMode,
53    SinkRuntimeContext, SinkTopology, WriteResult,
54};
55use crate::error::ConnectorError;
56use crate::storage::StorageProvider;
57
58use super::iceberg_config::{IcebergSinkConfig, IcebergStorageType};
59#[cfg(feature = "iceberg-core")]
60use artifact_inventory::{PreparedEpochArtifacts, PreparedEpochCleanup};
61#[cfg(feature = "iceberg-core")]
62use descriptor::IcebergCommitDescriptorV1;
63#[cfg(feature = "iceberg-core")]
64use epoch_intent::IcebergEpochIntentV1;
65#[cfg(feature = "iceberg-core")]
66use epoch_writer::{EpochIdentity, IcebergEpochWriter};
67#[cfg(feature = "iceberg-core")]
68use metrics::IcebergMetrics;
69#[cfg(feature = "iceberg-core")]
70use publication::UnresolvedIcebergPublication;
71#[cfg(feature = "iceberg-core")]
72pub(in crate::lakehouse) use schema_alignment::SchemaAlignmentPlan;
73
74/// Apache Iceberg append sink.
75pub struct IcebergSink {
76    config: IcebergSinkConfig,
77    schema: Option<SchemaRef>,
78    state: ConnectorState,
79    runtime_context: Option<SinkRuntimeContext>,
80    #[cfg(feature = "iceberg-core")]
81    standalone_deployment_id: String,
82    #[cfg(feature = "iceberg-core")]
83    direct_epoch: u64,
84    #[cfg(feature = "iceberg-core")]
85    catalog: Option<Arc<dyn iceberg::Catalog>>,
86    #[cfg(feature = "iceberg-core")]
87    catalog_capabilities: super::iceberg_io::CatalogCapabilities,
88    #[cfg(feature = "iceberg-core")]
89    catalog_session: super::iceberg_io::CatalogSession,
90    #[cfg(feature = "iceberg-core")]
91    table: Option<iceberg::table::Table>,
92    #[cfg(feature = "iceberg-core")]
93    iceberg_arrow_schema: Option<SchemaRef>,
94    #[cfg(feature = "iceberg-core")]
95    alignment_plan: Option<SchemaAlignmentPlan>,
96    #[cfg(feature = "iceberg-core")]
97    active_epoch: Mutex<Option<IcebergEpochWriter>>,
98    #[cfg(feature = "iceberg-core")]
99    active_epoch_id: Option<u64>,
100    #[cfg(feature = "iceberg-core")]
101    admitted_epoch_intent: Option<IcebergEpochIntentV1>,
102    #[cfg(feature = "iceberg-core")]
103    prepared_epoch: Option<PreparedEpochArtifacts>,
104    #[cfg(feature = "iceberg-core")]
105    metrics: IcebergMetrics,
106    #[cfg(feature = "iceberg-core")]
107    unresolved_publication: Arc<Mutex<Option<UnresolvedIcebergPublication>>>,
108}
109
110impl IcebergSink {
111    /// Creates an Iceberg sink from validated connector configuration.
112    #[must_use]
113    pub fn new(config: IcebergSinkConfig, registry: Option<&prometheus::Registry>) -> Self {
114        #[cfg(not(feature = "iceberg-core"))]
115        let _ = registry;
116        Self {
117            config,
118            schema: None,
119            state: ConnectorState::Created,
120            runtime_context: None,
121            #[cfg(feature = "iceberg-core")]
122            standalone_deployment_id: uuid::Uuid::now_v7().to_string(),
123            #[cfg(feature = "iceberg-core")]
124            direct_epoch: 0,
125            #[cfg(feature = "iceberg-core")]
126            catalog: None,
127            #[cfg(feature = "iceberg-core")]
128            catalog_capabilities: super::iceberg_io::CatalogCapabilities::default(),
129            #[cfg(feature = "iceberg-core")]
130            catalog_session: super::iceberg_io::CatalogSession::default(),
131            #[cfg(feature = "iceberg-core")]
132            table: None,
133            #[cfg(feature = "iceberg-core")]
134            iceberg_arrow_schema: None,
135            #[cfg(feature = "iceberg-core")]
136            alignment_plan: None,
137            #[cfg(feature = "iceberg-core")]
138            active_epoch: Mutex::new(None),
139            #[cfg(feature = "iceberg-core")]
140            active_epoch_id: None,
141            #[cfg(feature = "iceberg-core")]
142            admitted_epoch_intent: None,
143            #[cfg(feature = "iceberg-core")]
144            prepared_epoch: None,
145            #[cfg(feature = "iceberg-core")]
146            metrics: IcebergMetrics::new(registry),
147            #[cfg(feature = "iceberg-core")]
148            unresolved_publication: Arc::new(Mutex::new(None)),
149        }
150    }
151
152    fn is_coordinated(&self) -> bool {
153        self.config.delivery_guarantee == DeliveryGuarantee::ExactlyOnce
154    }
155
156    #[cfg(feature = "iceberg-core")]
157    fn table(&self) -> Result<&iceberg::table::Table, ConnectorError> {
158        self.table
159            .as_ref()
160            .ok_or_else(|| ConnectorError::InvalidState {
161                expected: "open Iceberg sink".into(),
162                actual: "table is not loaded".into(),
163            })
164    }
165
166    #[cfg(feature = "iceberg-core")]
167    fn epoch_identity(&self, epoch: u64) -> Result<EpochIdentity, ConnectorError> {
168        if self.is_coordinated() {
169            let context =
170                self.runtime_context
171                    .as_ref()
172                    .ok_or_else(|| ConnectorError::InvalidState {
173                        expected: "checkpoint runtime identity bound before open".into(),
174                        actual: "runtime identity is absent".into(),
175                    })?;
176            return Ok(EpochIdentity {
177                deployment_id: context.deployment_id.clone(),
178                sink_id: context.sink_id.clone(),
179                participant_id: context.participant_id,
180                epoch,
181            });
182        }
183        Ok(EpochIdentity {
184            deployment_id: self.standalone_deployment_id.clone(),
185            sink_id: format!(
186                "{}.{}",
187                self.config.catalog.namespace, self.config.catalog.table_name
188            ),
189            participant_id: 1,
190            epoch,
191        })
192    }
193
194    #[cfg(feature = "iceberg-core")]
195    fn new_epoch_writer(&self, epoch: u64) -> Result<IcebergEpochWriter, ConnectorError> {
196        IcebergEpochWriter::new(
197            self.table()?,
198            &self.config,
199            &self.epoch_identity(epoch)?,
200            self.metrics.clone(),
201        )
202    }
203
204    #[cfg(feature = "iceberg-core")]
205    async fn load_current_table(&self) -> Result<iceberg::table::Table, ConnectorError> {
206        let catalog = self
207            .catalog
208            .as_ref()
209            .ok_or_else(|| ConnectorError::InvalidState {
210                expected: "open Iceberg sink".into(),
211                actual: "catalog is not initialized".into(),
212            })?;
213        super::iceberg_io::load_table_with_timeout(
214            catalog.as_ref(),
215            &self.config.catalog.namespace,
216            &self.config.catalog.table_name,
217            self.config.catalog.request_timeout,
218        )
219        .await
220    }
221
222    #[cfg(feature = "iceberg-core")]
223    async fn initialize_active_writer(&mut self) -> Result<(), ConnectorError> {
224        if self.active_epoch.get_mut().is_some() {
225            return Ok(());
226        }
227        let epoch = self
228            .active_epoch_id
229            .ok_or_else(|| ConnectorError::InvalidState {
230                expected: "an active Iceberg epoch".into(),
231                actual: "no epoch identity is active".into(),
232            })?;
233        let current = self.load_current_table().await?;
234        validate_epoch_table_refresh(self.table()?, &current)?;
235        if self.is_coordinated() {
236            self.admitted_epoch_intent
237                .as_ref()
238                .ok_or_else(|| {
239                    ConnectorError::Internal("Iceberg active epoch lost its artifact intent".into())
240                })?
241                .validate_writer(&current, &self.config, &self.epoch_identity(epoch)?)?;
242        }
243        self.table = Some(current);
244        let writer = self.new_epoch_writer(epoch)?;
245        *self.active_epoch.get_mut() = Some(writer);
246        Ok(())
247    }
248
249    #[cfg(feature = "iceberg-core")]
250    fn align_batch_to_iceberg_schema(
251        &self,
252        batch: &RecordBatch,
253    ) -> Result<RecordBatch, ConnectorError> {
254        self.alignment_plan
255            .as_ref()
256            .ok_or_else(|| ConnectorError::InvalidState {
257                expected: "open Iceberg sink with a schema alignment plan".into(),
258                actual: "Iceberg schema alignment is not initialized".into(),
259            })?
260            .align(batch)
261    }
262
263    #[cfg(feature = "iceberg-core")]
264    async fn publish_direct_epoch(&mut self) -> Result<(), ConnectorError> {
265        let writer = self.active_epoch.get_mut().take();
266        self.active_epoch_id = None;
267        let Some(writer) = writer else {
268            return Ok(());
269        };
270        let writer_table = self.table()?.clone();
271        let mut output = match writer.close().await {
272            Ok(output) => output,
273            Err(error) => {
274                self.state = ConnectorState::Failed;
275                return Err(error);
276            }
277        };
278        if output.data_files.is_empty() {
279            return Ok(());
280        }
281        let artifacts = std::mem::take(&mut output.artifacts);
282        let catalog = self
283            .catalog
284            .clone()
285            .ok_or_else(|| ConnectorError::InvalidState {
286                expected: "open Iceberg sink".into(),
287                actual: "catalog is not initialized".into(),
288            })?;
289        match direct_publication::publish_direct_append(
290            &self.config,
291            &catalog,
292            &self.catalog_capabilities,
293            &self.catalog_session,
294            &writer_table,
295            output.data_files,
296            &self.metrics,
297        )
298        .await
299        {
300            Ok(updated) => {
301                self.table = Some(updated);
302                Ok(())
303            }
304            Err(error) => {
305                self.state = ConnectorState::Failed;
306                if error.is_outcome_unknown() {
307                    return Err(error);
308                }
309                let cleanup = artifacts
310                    .cleanup_aborted(writer_table.file_io().clone(), self.metrics.clone())
311                    .await;
312                Err(match cleanup {
313                    Ok(()) => error,
314                    Err(cleanup) => ConnectorError::WriteError(format!(
315                        "{error}; exact artifact cleanup also failed: {cleanup}"
316                    )),
317                })
318            }
319        }
320    }
321}
322
323#[async_trait]
324impl SinkConnector for IcebergSink {
325    fn bind_runtime_context(&mut self, context: SinkRuntimeContext) -> Result<(), ConnectorError> {
326        if self.state != ConnectorState::Created {
327            return Err(ConnectorError::InvalidState {
328                expected: "created Iceberg sink before open".into(),
329                actual: self.state.to_string(),
330            });
331        }
332        let deployment = uuid::Uuid::parse_str(&context.deployment_id).map_err(|_| {
333            ConnectorError::ConfigurationError(
334                "Iceberg runtime deployment ID must be a canonical non-nil UUID".into(),
335            )
336        })?;
337        if self.is_coordinated() && deployment.get_version_num() != 7 {
338            return Err(ConnectorError::ConfigurationError(
339                "[LDB-ICEBERG-RUNTIME-DEPLOYMENT-VERSION] exactly-once Iceberg requires a UUIDv7 deployment identity"
340                    .into(),
341            ));
342        }
343        if deployment.is_nil()
344            || deployment.to_string() != context.deployment_id
345            || context.sink_id.is_empty()
346            || context.participant_id == 0
347        {
348            return Err(ConnectorError::ConfigurationError(
349                "Iceberg runtime identity contains an invalid deployment, sink, or participant"
350                    .into(),
351            ));
352        }
353        if let Some(bound) = &self.runtime_context {
354            if bound == &context {
355                return Ok(());
356            }
357            return Err(ConnectorError::InvalidState {
358                expected: "the runtime identity already bound to this Iceberg sink".into(),
359                actual: "a different runtime identity".into(),
360            });
361        }
362        self.runtime_context = Some(context);
363        Ok(())
364    }
365
366    fn contract(&self, config: &ConnectorConfig) -> Result<SinkContract, ConnectorError> {
367        let config = IcebergSinkConfig::from_config(config)?;
368        capabilities::validate_sink(&config)?;
369        let shared_warehouse = config.storage.storage_type.map_or_else(
370            || StorageProvider::is_shared_uri(&config.catalog.warehouse),
371            |storage_type| {
372                matches!(
373                    storage_type,
374                    IcebergStorageType::S3 | IcebergStorageType::Gcs | IcebergStorageType::Azure
375                )
376            },
377        );
378        let consistency = if config.delivery_guarantee == DeliveryGuarantee::ExactlyOnce {
379            SinkConsistency::CheckpointCommittable
380        } else {
381            SinkConsistency::DurableAtLeastOnce
382        };
383        let contract = SinkContract::new(
384            consistency,
385            if shared_warehouse {
386                SinkTopology::MultiWriter
387            } else {
388                SinkTopology::Singleton
389            },
390            SinkInputMode::AppendOnly,
391        );
392        Ok(
393            if consistency == SinkConsistency::CheckpointCommittable
394                && capabilities::cluster_exact_append_certified(&config)
395            {
396                contract.with_cluster_exact_delivery_certification()
397            } else {
398                contract
399            },
400        )
401    }
402
403    async fn resolve_schema(
404        &mut self,
405        config: &ConnectorConfig,
406        input: SchemaRef,
407    ) -> Result<crate::schema::resolution::SchemaBinding, ConnectorError> {
408        #[cfg(feature = "iceberg-core")]
409        {
410            super::schema_resolution::iceberg_sink(config, input).await
411        }
412        #[cfg(not(feature = "iceberg-core"))]
413        {
414            let _ = (config, input);
415            Err(ConnectorError::FeatureUnsupported(
416                "Iceberg schema resolution requires the iceberg feature".into(),
417            ))
418        }
419    }
420
421    async fn prepare_schema(
422        &mut self,
423        config: &ConnectorConfig,
424        binding: &mut crate::schema::resolution::SchemaBinding,
425    ) -> Result<(), ConnectorError> {
426        #[cfg(feature = "iceberg-core")]
427        {
428            super::schema_resolution::prepare_iceberg(config, binding).await
429        }
430        #[cfg(not(feature = "iceberg-core"))]
431        {
432            let _ = (config, binding);
433            Err(ConnectorError::FeatureUnsupported(
434                "Iceberg preparation requires the iceberg feature".into(),
435            ))
436        }
437    }
438
439    async fn open(&mut self, config: &ConnectorConfig) -> Result<(), ConnectorError> {
440        if !config.properties().is_empty() {
441            self.config = IcebergSinkConfig::from_config(config)?;
442        }
443        capabilities::validate_sink(&self.config)?;
444        if self.is_coordinated() && self.runtime_context.is_none() {
445            return Err(ConnectorError::ConfigurationError(
446                "exactly-once Iceberg append requires checkpoint runtime identity".into(),
447            ));
448        }
449
450        #[cfg(feature = "iceberg-core")]
451        {
452            self.open_target(config).await
453        }
454
455        #[cfg(not(feature = "iceberg-core"))]
456        {
457            self.state = ConnectorState::Failed;
458            Err(ConnectorError::FeatureUnsupported(
459                "Apache Iceberg requires the 'iceberg' feature".into(),
460            ))
461        }
462    }
463
464    async fn write_batch(&mut self, batch: &RecordBatch) -> Result<WriteResult, ConnectorError> {
465        if self.state != ConnectorState::Running {
466            return Err(ConnectorError::InvalidState {
467                expected: "running Iceberg sink".into(),
468                actual: self.state.to_string(),
469            });
470        }
471        if batch.num_rows() == 0 {
472            return Ok(WriteResult::new(0, 0));
473        }
474        #[cfg(feature = "iceberg-core")]
475        {
476            if self.schema.is_none() {
477                self.schema = Some(batch.schema());
478            }
479            if self.active_epoch_id.is_none() {
480                if self.is_coordinated() {
481                    return Err(ConnectorError::InvalidState {
482                        expected: "begin_epoch before an exactly-once Iceberg write".into(),
483                        actual: "no active epoch".into(),
484                    });
485                }
486                self.direct_epoch = self.direct_epoch.checked_add(1).ok_or_else(|| {
487                    ConnectorError::WriteError("Iceberg direct epoch ID overflow".into())
488                })?;
489                self.active_epoch_id = Some(self.direct_epoch);
490            }
491            self.initialize_active_writer().await?;
492            let aligned = self.align_batch_to_iceberg_schema(batch)?;
493            let bytes = aligned.get_array_memory_size();
494            self.active_epoch
495                .get_mut()
496                .as_mut()
497                .ok_or_else(|| ConnectorError::Internal("Iceberg epoch writer disappeared".into()))?
498                .write(aligned)
499                .await?;
500            return Ok(WriteResult::new(
501                batch.num_rows(),
502                u64::try_from(bytes).unwrap_or(u64::MAX),
503            ));
504        }
505
506        #[cfg(not(feature = "iceberg-core"))]
507        Err(ConnectorError::FeatureUnsupported(
508            "Apache Iceberg requires the 'iceberg' feature".into(),
509        ))
510    }
511
512    fn schema(&self) -> SchemaRef {
513        self.schema
514            .clone()
515            .unwrap_or_else(|| Arc::new(arrow_schema::Schema::empty()))
516    }
517
518    async fn checkpoint_artifact_intent(
519        &mut self,
520        epoch: u64,
521    ) -> Result<Option<Vec<u8>>, ConnectorError> {
522        #[cfg(feature = "iceberg-core")]
523        {
524            if !self.is_coordinated() {
525                return Ok(None);
526            }
527            self.ensure_no_unresolved_publication()?;
528            if self.active_epoch_id.is_some() || self.active_epoch.get_mut().is_some() {
529                return Err(ConnectorError::InvalidState {
530                    expected: "no active Iceberg epoch before artifact admission".into(),
531                    actual: "an Iceberg epoch is still active".into(),
532                });
533            }
534            let identity = self.epoch_identity(epoch)?;
535            if let Some(intent) = self.admitted_epoch_intent.as_ref() {
536                intent.validate_writer(self.table()?, &self.config, &identity)?;
537                return Ok(Some(intent.encode()?));
538            }
539            let intent = IcebergEpochIntentV1::capture(self.table()?, &self.config, &identity)?;
540            let payload = intent.encode()?;
541            self.admitted_epoch_intent = Some(intent);
542            return Ok(Some(payload));
543        }
544
545        #[cfg(not(feature = "iceberg-core"))]
546        {
547            let _ = epoch;
548            Ok(None)
549        }
550    }
551
552    async fn begin_epoch(&mut self, epoch: u64) -> Result<(), ConnectorError> {
553        #[cfg(not(feature = "iceberg-core"))]
554        let _ = epoch;
555        #[cfg(feature = "iceberg-core")]
556        {
557            if !self.is_coordinated() {
558                return Ok(());
559            }
560            self.ensure_no_unresolved_publication()?;
561            if self.active_epoch_id.is_some() || self.active_epoch.get_mut().is_some() {
562                return Err(ConnectorError::InvalidState {
563                    expected: "previous Iceberg epoch prepared or rolled back".into(),
564                    actual: "an epoch identity or writer is still active".into(),
565                });
566            }
567            let intent = self.admitted_epoch_intent.as_ref().ok_or_else(|| {
568                ConnectorError::InvalidState {
569                    expected: format!("durable Iceberg artifact intent for epoch {epoch}"),
570                    actual: "artifact intent was not captured before begin_epoch".into(),
571                }
572            })?;
573            intent.validate_writer(self.table()?, &self.config, &self.epoch_identity(epoch)?)?;
574            self.cleanup_prepared_epoch(PreparedEpochCleanup::Successor { next_epoch: epoch })
575                .await?;
576            self.active_epoch_id = Some(epoch);
577        }
578        Ok(())
579    }
580
581    async fn pre_commit(&mut self, epoch: u64) -> Result<Option<Vec<u8>>, ConnectorError> {
582        #[cfg(not(feature = "iceberg-core"))]
583        let _ = epoch;
584        #[cfg(feature = "iceberg-core")]
585        {
586            if !self.is_coordinated() {
587                self.flush().await?;
588                return Ok(None);
589            }
590            self.ensure_no_unresolved_publication()?;
591            if self.active_epoch_id != Some(epoch) {
592                return Err(ConnectorError::InvalidState {
593                    expected: format!("active Iceberg epoch {epoch}"),
594                    actual: format!("active epoch is {:?}", self.active_epoch_id),
595                });
596            }
597            if self.prepared_epoch.is_some() {
598                return Err(ConnectorError::InvalidState {
599                    expected: "no previously prepared Iceberg epoch".into(),
600                    actual: "prepared artifact ownership is still active".into(),
601                });
602            }
603            let started = std::time::Instant::now();
604            let writer = self.active_epoch.get_mut().take();
605            self.active_epoch_id = None;
606            let Some(writer) = writer else {
607                self.admitted_epoch_intent = None;
608                self.metrics
609                    .pre_commit_duration
610                    .observe(started.elapsed().as_secs_f64());
611                return Ok(None);
612            };
613            self.prepared_epoch = Some(PreparedEpochArtifacts::new(
614                epoch,
615                writer.artifact_tracker(),
616                &self.metrics,
617            ));
618            let output = writer.close().await?;
619            self.admitted_epoch_intent = None;
620            self.metrics
621                .pre_commit_duration
622                .observe(started.elapsed().as_secs_f64());
623            if output.data_files.is_empty() {
624                self.prepared_epoch = None;
625                self.metrics.pending_artifact_paths.set(0);
626                return Ok(None);
627            }
628            self.prepared_epoch
629                .as_ref()
630                .ok_or_else(|| {
631                    ConnectorError::Internal(
632                        "Iceberg prepared artifact ownership disappeared during writer close"
633                            .into(),
634                    )
635                })?
636                .seal(&output.artifacts, &self.metrics)?;
637            #[cfg(test)]
638            fault_injection::fail_if(fault_injection::IcebergFaultPoint::AfterFileClose)?;
639            let descriptor = IcebergCommitDescriptorV1::encode(
640                self.table()?,
641                &self.config,
642                &self.epoch_identity(epoch)?,
643                output,
644            )?;
645            #[cfg(test)]
646            fault_injection::fail_if(fault_injection::IcebergFaultPoint::AfterDescriptor)?;
647            self.prepared_epoch
648                .as_mut()
649                .ok_or_else(|| {
650                    ConnectorError::Internal(
651                        "Iceberg prepared artifact ownership disappeared before descriptor issue"
652                            .into(),
653                    )
654                })?
655                .mark_descriptor_issued();
656            return Ok(Some(descriptor));
657        }
658
659        #[cfg(not(feature = "iceberg-core"))]
660        Err(ConnectorError::FeatureUnsupported(
661            "Apache Iceberg requires the 'iceberg' feature".into(),
662        ))
663    }
664
665    async fn rollback_epoch(&mut self, epoch: u64) -> Result<(), ConnectorError> {
666        #[cfg(not(feature = "iceberg-core"))]
667        let _ = epoch;
668        #[cfg(feature = "iceberg-core")]
669        {
670            if self.active_epoch_id.is_some_and(|active| active != epoch) {
671                return Err(ConnectorError::InvalidState {
672                    expected: format!("rollback active Iceberg epoch {epoch}"),
673                    actual: format!("active epoch is {:?}", self.active_epoch_id),
674                });
675            }
676            let active_cleanup = self.discard_active_epoch().await;
677            let prepared_cleanup = self
678                .cleanup_prepared_epoch(PreparedEpochCleanup::Abort { epoch })
679                .await;
680            match (active_cleanup, prepared_cleanup) {
681                (Ok(()), Ok(())) => {}
682                (Err(error), Ok(())) | (Ok(()), Err(error)) => return Err(error),
683                (Err(active), Err(prepared)) => {
684                    return Err(ConnectorError::WriteError(format!(
685                        "{active}; prepared artifact cleanup also failed: {prepared}"
686                    )));
687                }
688            }
689            self.admitted_epoch_intent = None;
690        }
691        Ok(())
692    }
693
694    fn suggested_write_timeout(&self) -> Duration {
695        self.config.catalog.commit_timeout
696    }
697
698    fn flush_interval(&self) -> Duration {
699        self.config.max_flush_age.min(Duration::from_secs(5))
700    }
701
702    async fn flush(&mut self) -> Result<(), ConnectorError> {
703        if self.state != ConnectorState::Running {
704            return Err(ConnectorError::InvalidState {
705                expected: "running Iceberg sink".into(),
706                actual: self.state.to_string(),
707            });
708        }
709        #[cfg(feature = "iceberg-core")]
710        {
711            if self.is_coordinated() {
712                return Ok(());
713            }
714            return self.publish_direct_epoch().await;
715        }
716
717        #[cfg(not(feature = "iceberg-core"))]
718        Err(ConnectorError::FeatureUnsupported(
719            "Apache Iceberg requires the 'iceberg' feature".into(),
720        ))
721    }
722
723    async fn close(&mut self) -> Result<(), ConnectorError> {
724        #[cfg(feature = "iceberg-core")]
725        {
726            if self.is_coordinated() {
727                self.discard_active_epoch().await?;
728                self.cleanup_prepared_epoch(PreparedEpochCleanup::Close)
729                    .await?;
730                self.admitted_epoch_intent = None;
731            } else {
732                self.flush().await?;
733            }
734            self.catalog = None;
735            self.catalog_capabilities = super::iceberg_io::CatalogCapabilities::default();
736            self.catalog_session = super::iceberg_io::CatalogSession::default();
737            self.table = None;
738            self.iceberg_arrow_schema = None;
739            self.alignment_plan = None;
740            self.metrics.set_buffer(0, 0);
741            self.metrics.set_active_writers(0);
742            self.metrics.pending_artifact_paths.set(0);
743        }
744        self.state = ConnectorState::Closed;
745        Ok(())
746    }
747
748    #[cfg(feature = "iceberg-core")]
749    fn as_coordinated_committer(&self) -> Option<&dyn crate::connector::CoordinatedCommitter> {
750        self.is_coordinated()
751            .then_some(self as &dyn crate::connector::CoordinatedCommitter)
752    }
753
754    fn coordinated_abort_cleaner(
755        &self,
756    ) -> Option<Arc<dyn crate::connector::CoordinatedAbortCleaner>> {
757        #[cfg(feature = "iceberg-core")]
758        {
759            if !self.is_coordinated() {
760                return None;
761            }
762            let catalog = self.catalog.as_ref()?.clone();
763            Some(Arc::new(aborted_cleanup::IcebergAbortCleaner::new(
764                catalog,
765                self.config.clone(),
766                self.metrics.clone(),
767                Arc::clone(&self.unresolved_publication),
768            )))
769        }
770        #[cfg(not(feature = "iceberg-core"))]
771        None
772    }
773}
774
775#[cfg(feature = "iceberg-core")]
776fn validate_epoch_table_refresh(
777    expected: &iceberg::table::Table,
778    current: &iceberg::table::Table,
779) -> Result<(), ConnectorError> {
780    if expected.metadata().uuid() != current.metadata().uuid()
781        || expected.identifier() != current.identifier()
782        || expected.metadata().location() != current.metadata().location()
783    {
784        return Err(ConnectorError::TransactionError(
785            "[LDB-ICEBERG-TABLE-REPLACED] Iceberg table identity or location changed while the sink was running"
786                .into(),
787        ));
788    }
789    if expected.metadata().current_schema_id() != current.metadata().current_schema_id() {
790        return Err(ConnectorError::SchemaMismatch(
791            "[LDB-ICEBERG-SCHEMA-CHANGED] Iceberg schema changed while schema.evolution.mode=strict"
792                .into(),
793        ));
794    }
795    Ok(())
796}
797
798#[cfg(feature = "iceberg-core")]
799fn validate_direct_publication_table(
800    writer_table: &iceberg::table::Table,
801    current: &iceberg::table::Table,
802) -> Result<(), ConnectorError> {
803    validate_epoch_table_refresh(writer_table, current)?;
804    let writer = writer_table.metadata();
805    let current = current.metadata();
806    if writer.default_partition_spec_id() != current.default_partition_spec_id()
807        || writer.default_sort_order_id() != current.default_sort_order_id()
808        || writer.format_version() != current.format_version()
809    {
810        return Err(ConnectorError::TransactionError(
811            "[LDB-ICEBERG-DIRECT-LAYOUT-CHANGED] Iceberg partition spec, sort order, or format version changed while a direct epoch was open"
812                .into(),
813        ));
814    }
815    Ok(())
816}
817
818#[cfg(test)]
819mod tests;
laminar_connectors/lakehouse/iceberg/mod.rs

laminar_connectors/lakehouse/iceberg/
mod.rs
