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

laminar_connectors/lakehouse/iceberg/
mod.rs
