laminar_connectors/lakehouse/iceberg_source/
mod.rs

1//! Replayable Apache Iceberg snapshot and append source.
2
3#[cfg(feature = "iceberg-core")]
4mod append_lineage;
5#[cfg(feature = "iceberg-core")]
6mod cursor;
7#[cfg(feature = "iceberg-core")]
8mod metrics;
9#[cfg(feature = "iceberg-core")]
10mod planner;
11#[cfg(feature = "iceberg-core")]
12mod read_schema;
13
14use std::sync::Arc;
15#[cfg(feature = "iceberg-core")]
16use std::time::Instant;
17
18#[cfg(feature = "iceberg-core")]
19use arrow_array::RecordBatch;
20use arrow_schema::SchemaRef;
21use async_trait::async_trait;
22#[cfg(feature = "iceberg-core")]
23use iceberg::expr::Predicate;
24#[cfg(feature = "iceberg-core")]
25use iceberg::spec::SnapshotRef;
26#[cfg(feature = "iceberg-core")]
27use tracing::info;
28
29use crate::checkpoint::SourceCheckpoint;
30use crate::config::{ConnectorConfig, ConnectorState};
31use crate::connector::{
32    SourceBatch, SourceCheckpointUnavailablePolicy, SourceConnector, SourceConsistency,
33    SourceContract, SourceInputMode, SourcePosition, SourceStart, SourceTopology,
34};
35use crate::error::ConnectorError;
36
37#[cfg(feature = "iceberg-core")]
38use super::iceberg_config::IcebergReadBootstrap;
39use super::iceberg_config::{IcebergReadMode, IcebergSourceConfig};
40#[cfg(feature = "iceberg-core")]
41use metrics::IcebergSourceMetrics;
42#[cfg(feature = "iceberg-core")]
43use planner::{ScanOutput, ScanTask};
44#[cfg(feature = "iceberg-core")]
45use read_schema::ReadSchemaBinding;
46
47#[cfg(feature = "iceberg-core")]
48pub use cursor::{IcebergSourceCursorOriginV1, IcebergSourceCursorV1};
49
50#[cfg(feature = "iceberg-core")]
51const PARTIAL_SCAN_ERROR: &str = "[LDB-ICEBERG-PARTIAL-SCAN] Iceberg scan failed after partial snapshot emission; recovery from the last durable cursor is required";
52
53#[cfg(feature = "iceberg-core")]
54struct PendingBatch {
55    batch: RecordBatch,
56    completed_cursor: Option<IcebergSourceCursorV1>,
57}
58
59/// Apache Iceberg source with bounded scan-to-ingestion backpressure.
60pub struct IcebergSource {
61    config: IcebergSourceConfig,
62    schema: Option<SchemaRef>,
63    state: ConnectorState,
64    checkpoint: SourceCheckpoint,
65    #[cfg(feature = "iceberg-core")]
66    cursor: Option<IcebergSourceCursorV1>,
67    #[cfg(feature = "iceberg-core")]
68    pending: Option<PendingBatch>,
69    #[cfg(feature = "iceberg-core")]
70    scan: Option<ScanTask>,
71    #[cfg(feature = "iceberg-core")]
72    replay_unit_in_progress: bool,
73    #[cfg(feature = "iceberg-core")]
74    scan_failed: bool,
75    #[cfg(feature = "iceberg-core")]
76    last_poll_time: Option<Instant>,
77    #[cfg(feature = "iceberg-core")]
78    bounded_snapshot_complete: bool,
79    #[cfg(feature = "iceberg-core")]
80    current_sequence_number: Option<i64>,
81    #[cfg(feature = "iceberg-core")]
82    pending_snapshots: usize,
83    #[cfg(feature = "iceberg-core")]
84    metrics: IcebergSourceMetrics,
85    #[cfg(feature = "iceberg-core")]
86    catalog: Option<Arc<dyn iceberg::Catalog>>,
87    #[cfg(feature = "iceberg-core")]
88    table: Option<iceberg::table::Table>,
89    #[cfg(feature = "iceberg-core")]
90    read_schema: Option<ReadSchemaBinding>,
91    #[cfg(feature = "iceberg-core")]
92    filter_predicate: Option<Predicate>,
93}
94
95impl IcebergSource {
96    /// Creates an Iceberg source from parsed configuration.
97    #[must_use]
98    pub fn new(config: IcebergSourceConfig, registry: Option<&prometheus::Registry>) -> Self {
99        #[cfg(not(feature = "iceberg-core"))]
100        let _ = registry;
101        let mut checkpoint = SourceCheckpoint::new();
102        checkpoint.set_metadata("connector_type", "iceberg");
103        Self {
104            config,
105            schema: None,
106            state: ConnectorState::Created,
107            checkpoint,
108            #[cfg(feature = "iceberg-core")]
109            cursor: None,
110            #[cfg(feature = "iceberg-core")]
111            pending: None,
112            #[cfg(feature = "iceberg-core")]
113            scan: None,
114            #[cfg(feature = "iceberg-core")]
115            replay_unit_in_progress: false,
116            #[cfg(feature = "iceberg-core")]
117            scan_failed: false,
118            #[cfg(feature = "iceberg-core")]
119            last_poll_time: None,
120            #[cfg(feature = "iceberg-core")]
121            bounded_snapshot_complete: false,
122            #[cfg(feature = "iceberg-core")]
123            current_sequence_number: None,
124            #[cfg(feature = "iceberg-core")]
125            pending_snapshots: 0,
126            #[cfg(feature = "iceberg-core")]
127            metrics: IcebergSourceMetrics::new(registry),
128            #[cfg(feature = "iceberg-core")]
129            catalog: None,
130            #[cfg(feature = "iceberg-core")]
131            table: None,
132            #[cfg(feature = "iceberg-core")]
133            read_schema: None,
134            #[cfg(feature = "iceberg-core")]
135            filter_predicate: None,
136        }
137    }
138
139    #[cfg(feature = "iceberg-core")]
140    fn bind_read_schema(
141        &mut self,
142        root: &iceberg::spec::Schema,
143        declared: Option<SchemaRef>,
144    ) -> Result<(), ConnectorError> {
145        let predicate = super::iceberg_scan::parse_and_bind_filter(
146            self.config.filter.as_deref(),
147            Arc::new(root.clone()),
148        )?;
149        let binding = ReadSchemaBinding::bind(root, &self.config.select_columns, declared)?;
150        self.schema = Some(binding.output_schema());
151        self.read_schema = Some(binding);
152        self.filter_predicate = predicate;
153        Ok(())
154    }
155
156    #[cfg(feature = "iceberg-core")]
157    fn read_schema(&self) -> Result<&ReadSchemaBinding, ConnectorError> {
158        self.read_schema
159            .as_ref()
160            .ok_or_else(|| ConnectorError::InvalidState {
161                expected: "Iceberg source bound to a retained snapshot schema".into(),
162                actual: "read schema is unavailable".into(),
163            })
164    }
165
166    #[cfg(feature = "iceberg-core")]
167    fn install_cursor(&mut self, cursor: IcebergSourceCursorV1) -> Result<(), ConnectorError> {
168        let checkpoint = cursor.to_checkpoint()?;
169        self.replay_unit_in_progress = false;
170        self.metrics.processed_snapshot_id.set(cursor.snapshot_id);
171        self.pending_snapshots = self.pending_snapshots.saturating_sub(1);
172        self.metrics
173            .snapshot_lag
174            .set(i64::try_from(self.pending_snapshots).unwrap_or(i64::MAX));
175        if let Some(current) = self.current_sequence_number {
176            self.metrics
177                .sequence_lag
178                .set(current.saturating_sub(cursor.sequence_number));
179        }
180        self.checkpoint = checkpoint;
181        self.cursor = Some(cursor);
182        Ok(())
183    }
184
185    #[cfg(feature = "iceberg-core")]
186    fn observe_table_head(&mut self, snapshot: Option<(i64, i64)>) {
187        if let Some((snapshot_id, sequence_number)) = snapshot {
188            self.metrics.current_snapshot_id.set(snapshot_id);
189            self.current_sequence_number = Some(sequence_number);
190            if let Some(cursor) = &self.cursor {
191                self.metrics
192                    .sequence_lag
193                    .set(sequence_number.saturating_sub(cursor.sequence_number));
194            }
195        }
196    }
197
198    #[cfg(feature = "iceberg-core")]
199    fn set_pending_snapshots(&mut self, pending: usize) {
200        self.pending_snapshots = pending;
201        self.metrics
202            .snapshot_lag
203            .set(i64::try_from(pending).unwrap_or(i64::MAX));
204    }
205
206    #[cfg(feature = "iceberg-core")]
207    fn emit_pending(&mut self, max_records: usize) -> Result<Option<SourceBatch>, ConnectorError> {
208        let Some(mut pending) = self.pending.take() else {
209            return Ok(None);
210        };
211        if pending.batch.num_rows() > max_records {
212            let emitted = pending.batch.slice(0, max_records);
213            pending.batch = pending
214                .batch
215                .slice(max_records, pending.batch.num_rows() - max_records);
216            self.pending = Some(pending);
217            self.replay_unit_in_progress = true;
218            return Ok(Some(
219                SourceBatch::new(emitted).with_checkpoint(self.checkpoint.clone()),
220            ));
221        }
222        if let Some(cursor) = pending.completed_cursor {
223            self.install_cursor(cursor)?;
224        } else {
225            self.replay_unit_in_progress = true;
226        }
227        Ok(Some(
228            SourceBatch::new(pending.batch).with_checkpoint(self.checkpoint.clone()),
229        ))
230    }
231
232    #[cfg(feature = "iceberg-core")]
233    fn selected_snapshot(
234        table: &iceberg::table::Table,
235        config: &IcebergSourceConfig,
236    ) -> Result<Option<SnapshotRef>, ConnectorError> {
237        if let Some(snapshot_id) = config.snapshot_id {
238            return table
239                .metadata()
240                .snapshot_by_id(snapshot_id)
241                .cloned()
242                .map(Some)
243                .ok_or_else(|| {
244                    ConnectorError::ReadError(format!(
245                        "[LDB-ICEBERG-SNAPSHOT-MISSING] snapshot {snapshot_id} does not exist"
246                    ))
247                });
248        }
249        if let Some(snapshot) = table.metadata().snapshot_for_ref(&config.table_ref) {
250            return Ok(Some(snapshot.clone()));
251        }
252        if config.table_ref == "main" && table.metadata().current_snapshot().is_none() {
253            return Ok(None);
254        }
255        Err(ConnectorError::ReadError(format!(
256            "[LDB-ICEBERG-REF-MISSING] table ref '{}' does not exist",
257            config.table_ref
258        )))
259    }
260
261    #[cfg(feature = "iceberg-core")]
262    fn start_initial_scan(&mut self) -> Result<(), ConnectorError> {
263        let table = self
264            .table
265            .as_ref()
266            .ok_or_else(|| ConnectorError::InvalidState {
267                expected: "loaded Iceberg table".into(),
268                actual: "table unavailable".into(),
269            })?
270            .clone();
271        let selected = Self::selected_snapshot(&table, &self.config)?;
272        if self.read_schema.is_none() {
273            let root_schema = match selected.as_ref() {
274                Some(snapshot) => snapshot.schema(table.metadata()).map_err(|error| {
275                    super::iceberg_scan::connector_scan_error(
276                        "resolve retained Iceberg snapshot schema",
277                        &error,
278                    )
279                })?,
280                None => table.current_schema_ref(),
281            };
282            self.bind_read_schema(&root_schema, None)?;
283        }
284        let Some(snapshot) = selected else {
285            if self.config.read_mode == IcebergReadMode::Snapshot {
286                self.bounded_snapshot_complete = true;
287                return Ok(());
288            }
289            let cursor = IcebergSourceCursorV1::from_empty_table(
290                &self.config,
291                &table,
292                self.read_schema()?.schema_id(),
293            )?;
294            return self.install_cursor(cursor);
295        };
296        let cursor = IcebergSourceCursorV1::from_snapshot(
297            &self.config,
298            &table,
299            &snapshot,
300            self.read_schema()?.schema_id(),
301        )?;
302        if self.config.read_mode == IcebergReadMode::Append {
303            append_lineage::validate_cursor_lineage(
304                &table,
305                &cursor,
306                self.config.max_snapshots_per_poll,
307            )?;
308        }
309        if self.config.read_mode == IcebergReadMode::Append
310            && self.config.bootstrap == IcebergReadBootstrap::None
311        {
312            return self.install_cursor(cursor);
313        }
314        self.set_pending_snapshots(1);
315        self.scan = Some(planner::full_snapshot_task(
316            table,
317            &self.config,
318            self.read_schema()?,
319            self.filter_predicate.clone(),
320            snapshot.snapshot_id(),
321        )?);
322        Ok(())
323    }
324
325    #[cfg(feature = "iceberg-core")]
326    async fn refresh_append(&mut self) -> Result<(), ConnectorError> {
327        if let Some(last_poll) = self.last_poll_time {
328            if last_poll.elapsed() < self.config.poll_interval {
329                return Ok(());
330            }
331        }
332        let poll_started = Instant::now();
333        let catalog = self
334            .catalog
335            .as_ref()
336            .ok_or_else(|| ConnectorError::InvalidState {
337                expected: "started Iceberg source".into(),
338                actual: "catalog unavailable".into(),
339            })?;
340        let table = super::iceberg_io::load_table_with_timeout(
341            catalog.as_ref(),
342            &self.config.catalog.namespace,
343            &self.config.catalog.table_name,
344            self.config.catalog.request_timeout,
345        )
346        .await?;
347        let head = table
348            .metadata()
349            .snapshot_for_ref(&self.config.table_ref)
350            .map(|snapshot| (snapshot.snapshot_id(), snapshot.sequence_number()));
351        self.observe_table_head(head);
352        self.metrics
353            .observe_table(table.metadata(), &self.config.table_ref);
354
355        if let Some(cursor) = self.cursor.as_ref() {
356            cursor.validate_binding(&self.config, &table)?;
357            let planning_started = Instant::now();
358            let deadline = super::iceberg_io::checked_deadline(
359                self.config.storage.request_timeout,
360                "storage.request_timeout",
361            )?;
362            let plans = tokio::time::timeout_at(
363                deadline,
364                append_lineage::plan_appends(&table, cursor, &self.config, deadline),
365            )
366            .await
367            .map_err(|_| {
368                ConnectorError::ReadError(
369                    "[LDB-ICEBERG-APPEND-PLANNING-TIMEOUT] append manifest planning exceeded storage.request_timeout"
370                        .into(),
371                )
372            })??;
373            self.metrics
374                .planning_duration
375                .observe(planning_started.elapsed().as_secs_f64());
376            let planned_files = plans.iter().fold(0_u64, |total, plan| {
377                total.saturating_add(u64::try_from(plan.added_files.len()).unwrap_or(u64::MAX))
378            });
379            let planned_manifests = plans.iter().fold(0_u64, |total, plan| {
380                total.saturating_add(u64::try_from(plan.manifest_count).unwrap_or(u64::MAX))
381            });
382            self.metrics.planned_files.inc_by(planned_files);
383            self.metrics.planned_manifests.inc_by(planned_manifests);
384            self.set_pending_snapshots(plans.len());
385            self.scan = planner::append_task(
386                table.clone(),
387                &self.config,
388                self.read_schema()?,
389                self.filter_predicate.clone(),
390                plans,
391            )?;
392        } else if let Some(snapshot) = Self::selected_snapshot(&table, &self.config)? {
393            if self.config.bootstrap == IcebergReadBootstrap::Initial {
394                self.scan = Some(planner::full_snapshot_task(
395                    table.clone(),
396                    &self.config,
397                    self.read_schema()?,
398                    self.filter_predicate.clone(),
399                    snapshot.snapshot_id(),
400                )?);
401            } else {
402                let cursor = IcebergSourceCursorV1::from_snapshot(
403                    &self.config,
404                    &table,
405                    &snapshot,
406                    self.read_schema()?.schema_id(),
407                )?;
408                self.install_cursor(cursor)?;
409            }
410        }
411        self.table = Some(table);
412        self.last_poll_time = Some(Instant::now());
413        self.metrics
414            .poll_duration
415            .observe(poll_started.elapsed().as_secs_f64());
416        Ok(())
417    }
418
419    #[cfg(feature = "iceberg-core")]
420    async fn finish_scan(&mut self) -> Result<(), ConnectorError> {
421        let Some(scan) = self.scan.take() else {
422            return Ok(());
423        };
424        let failed = std::mem::take(&mut self.scan_failed);
425        scan.handle.await.map_err(|error| {
426            ConnectorError::Internal(format!(
427                "Iceberg scan task terminated unexpectedly: {error}"
428            ))
429        })?;
430        self.metrics
431            .read_duration
432            .observe(scan.started_at.elapsed().as_secs_f64());
433        if self.config.read_mode == IcebergReadMode::Snapshot && !failed {
434            self.bounded_snapshot_complete = true;
435        }
436        Ok(())
437    }
438
439    #[cfg(feature = "iceberg-core")]
440    fn fail_partial_scan(&mut self) -> ConnectorError {
441        self.state = ConnectorState::Failed;
442        ConnectorError::TransactionError(PARTIAL_SCAN_ERROR.into())
443    }
444
445    #[cfg(feature = "iceberg-core")]
446    async fn next_scan_output(&mut self) -> Result<Option<ScanOutput>, ConnectorError> {
447        let result = match self.scan.as_mut() {
448            Some(scan) => scan.receiver.recv().await,
449            None => return Ok(None),
450        };
451        if let Some(result) = result {
452            match result {
453                Err(error) => {
454                    self.scan_failed = true;
455                    if self.replay_unit_in_progress {
456                        Err(self.fail_partial_scan())
457                    } else {
458                        Err(error)
459                    }
460                }
461                Ok(output) => Ok(Some(output)),
462            }
463        } else {
464            match self.finish_scan().await {
465                Ok(()) => Ok(None),
466                Err(_) if self.replay_unit_in_progress => Err(self.fail_partial_scan()),
467                Err(error) => Err(error),
468            }
469        }
470    }
471}
472
473#[async_trait]
474impl SourceConnector for IcebergSource {
475    async fn resolve_schema(
476        &mut self,
477        config: &ConnectorConfig,
478        explicit: Option<SchemaRef>,
479    ) -> Result<crate::schema::resolution::SchemaBinding, ConnectorError> {
480        #[cfg(feature = "iceberg-core")]
481        {
482            super::schema_resolution::iceberg_source(config, explicit).await
483        }
484        #[cfg(not(feature = "iceberg-core"))]
485        {
486            let _ = (config, explicit);
487            Err(ConnectorError::FeatureUnsupported(
488                "Iceberg schema discovery requires the iceberg feature".into(),
489            ))
490        }
491    }
492
493    async fn start(&mut self, request: SourceStart) -> Result<(), ConnectorError> {
494        let (config, position, _) = request.into_parts();
495        if matches!(&position, SourcePosition::Initialized { .. }) {
496            return Err(ConnectorError::ConfigurationError(
497                "Iceberg has no sealed topology startup contract".into(),
498            ));
499        }
500        #[cfg(feature = "iceberg-core")]
501        let declared_schema = if config.get("_arrow_schema").is_some() {
502            let schema = config.arrow_schema().ok_or_else(|| {
503                ConnectorError::ConfigurationError(
504                    "invalid Iceberg source '_arrow_schema' encoding".into(),
505                )
506            })?;
507            if schema.fields().is_empty() {
508                return Err(ConnectorError::ConfigurationError(
509                    "Iceberg source '_arrow_schema' must declare at least one column".into(),
510                ));
511            }
512            Some(schema)
513        } else {
514            None
515        };
516        if !config.properties().is_empty() {
517            self.config = IcebergSourceConfig::from_config(&config)?;
518        }
519        super::iceberg::capabilities::validate_source(&self.config)?;
520        if matches!(&position, SourcePosition::Resume { .. })
521            && self.config.read_mode != IcebergReadMode::Append
522        {
523            return Err(ConnectorError::ConfigurationError(
524                "only Iceberg read.mode=append has a replay cursor".into(),
525            ));
526        }
527
528        #[cfg(feature = "iceberg-core")]
529        {
530            let recovered_cursor = match &position {
531                SourcePosition::Initial => None,
532                SourcePosition::Initialized { .. } => {
533                    return Err(ConnectorError::ConfigurationError(
534                        "Iceberg has no sealed topology startup contract".into(),
535                    ));
536                }
537                SourcePosition::Resume { checkpoint, .. } => {
538                    Some(IcebergSourceCursorV1::from_checkpoint(checkpoint)?)
539                }
540            };
541            let catalog = super::iceberg_io::build_catalog_for_access_with_metrics(
542                &self.config.catalog,
543                &self.config.storage,
544                super::iceberg_io::CatalogAccess::Read,
545                Some(self.metrics.credential_refresh_failures.clone()),
546            )
547            .await?
548            .catalog;
549            let table = super::iceberg_io::load_table_with_timeout(
550                catalog.as_ref(),
551                &self.config.catalog.namespace,
552                &self.config.catalog.table_name,
553                self.config.catalog.request_timeout,
554            )
555            .await?;
556            let root_schema =
557                self.resolve_root_schema(&config, &table, recovered_cursor.as_ref())?;
558            self.bind_read_schema(&root_schema, declared_schema)?;
559            self.catalog = Some(catalog);
560            let head = table
561                .metadata()
562                .snapshot_for_ref(&self.config.table_ref)
563                .map(|snapshot| (snapshot.snapshot_id(), snapshot.sequence_number()));
564            self.observe_table_head(head);
565            self.metrics
566                .observe_table(table.metadata(), &self.config.table_ref);
567            self.table = Some(table);
568
569            match position {
570                SourcePosition::Initial => self.start_initial_scan()?,
571                SourcePosition::Initialized { .. } => {
572                    return Err(ConnectorError::ConfigurationError(
573                        "Iceberg has no sealed topology startup contract".into(),
574                    ));
575                }
576                SourcePosition::Resume { .. } => {
577                    self.install_cursor(recovered_cursor.ok_or_else(|| {
578                        ConnectorError::Internal(
579                            "Iceberg resume cursor was not retained during startup".into(),
580                        )
581                    })?)?;
582                }
583            }
584            self.state = ConnectorState::Running;
585            info!(
586                operation_class = "connector-open",
587                storage_provider = %self.config.storage.diagnostic_provider(&self.config.catalog.warehouse),
588                storage_endpoint_class = %self.config.storage.diagnostic_endpoint_class(&self.config.catalog.warehouse),
589                storage_auth_source = %self.config.storage.diagnostic_auth_source(&self.config.catalog.warehouse),
590                request_timeout_ms = self.config.storage.request_timeout.as_millis(),
591                success = true,
592                table = self.config.catalog.table_name,
593                namespace = self.config.catalog.namespace,
594                mode = %self.config.read_mode,
595                "iceberg source connected"
596            );
597            return Ok(());
598        }
599
600        #[cfg(not(feature = "iceberg-core"))]
601        {
602            self.state = ConnectorState::Failed;
603            Err(ConnectorError::FeatureUnsupported(
604                "Apache Iceberg requires the 'iceberg' feature".into(),
605            ))
606        }
607    }
608
609    async fn poll_batch(
610        &mut self,
611        max_records: usize,
612    ) -> Result<Option<SourceBatch>, ConnectorError> {
613        if max_records == 0 {
614            return Err(ConnectorError::ConfigurationError(
615                "Iceberg source max_records must be greater than zero".into(),
616            ));
617        }
618        #[cfg(feature = "iceberg-core")]
619        {
620            if self.state == ConnectorState::Failed {
621                return if self.replay_unit_in_progress {
622                    Err(ConnectorError::TransactionError(PARTIAL_SCAN_ERROR.into()))
623                } else {
624                    Err(ConnectorError::InvalidState {
625                        expected: "Running".into(),
626                        actual: self.state.to_string(),
627                    })
628                };
629            }
630            loop {
631                if let Some(batch) = self.emit_pending(max_records)? {
632                    return Ok(Some(batch));
633                }
634                if let Some(output) = self.next_scan_output().await? {
635                    match output {
636                        ScanOutput::Batch {
637                            batch,
638                            completed_cursor,
639                        } => {
640                            self.metrics.observe_batch(&batch);
641                            self.pending = Some(PendingBatch {
642                                batch,
643                                completed_cursor,
644                            });
645                        }
646                        ScanOutput::Cursor(cursor) => self.install_cursor(cursor)?,
647                        ScanOutput::ReadMetrics {
648                            files,
649                            storage_bytes,
650                        } => self.metrics.observe_completed_read(files, storage_bytes),
651                    }
652                    continue;
653                }
654                if self.bounded_snapshot_complete {
655                    return Ok(None);
656                }
657                if self.config.read_mode == IcebergReadMode::Snapshot {
658                    self.start_initial_scan()?;
659                    if self.scan.is_some() {
660                        continue;
661                    }
662                    return Ok(None);
663                }
664                if self.config.read_mode == IcebergReadMode::Append {
665                    self.refresh_append().await?;
666                    if self.scan.is_some() {
667                        continue;
668                    }
669                }
670                return Ok(None);
671            }
672        }
673        #[cfg(not(feature = "iceberg-core"))]
674        {
675            Err(ConnectorError::FeatureUnsupported(
676                "Apache Iceberg requires the 'iceberg' feature".into(),
677            ))
678        }
679    }
680
681    fn schema(&self) -> SchemaRef {
682        self.schema
683            .clone()
684            .unwrap_or_else(|| Arc::new(arrow_schema::Schema::empty()))
685    }
686
687    fn checkpoint(&self) -> SourceCheckpoint {
688        self.checkpoint.clone()
689    }
690
691    fn try_checkpoint(&self) -> Result<Option<SourceCheckpoint>, ConnectorError> {
692        #[cfg(feature = "iceberg-core")]
693        if self.replay_unit_in_progress {
694            return Ok(None);
695        }
696        Ok(Some(self.checkpoint()))
697    }
698
699    fn checkpoint_unavailable_policy(&self) -> SourceCheckpointUnavailablePolicy {
700        SourceCheckpointUnavailablePolicy::PollToReplayBoundary
701    }
702
703    fn contract(&self, config: &ConnectorConfig) -> Result<SourceContract, ConnectorError> {
704        let parsed = if config.properties().is_empty() {
705            self.config.clone()
706        } else {
707            IcebergSourceConfig::from_config(config)?
708        };
709        super::iceberg::capabilities::validate_source(&parsed)?;
710        Ok(SourceContract::new(
711            if parsed.read_mode == IcebergReadMode::Append {
712                SourceConsistency::Replayable
713            } else {
714                SourceConsistency::Ephemeral
715            },
716            SourceTopology::Singleton,
717            SourceInputMode::AppendOnly,
718        ))
719    }
720
721    async fn close(&mut self) -> Result<(), ConnectorError> {
722        #[cfg(feature = "iceberg-core")]
723        {
724            if let Some(scan) = self.scan.take() {
725                drop(scan.receiver);
726                scan.handle.abort();
727                let _ = scan.handle.await;
728            }
729            self.catalog = None;
730            self.table = None;
731            self.read_schema = None;
732            self.filter_predicate = None;
733            self.pending = None;
734            self.replay_unit_in_progress = false;
735            self.scan_failed = false;
736        }
737        self.state = ConnectorState::Closed;
738        Ok(())
739    }
740}
741
742#[cfg(test)]
743mod tests;
laminar_connectors/lakehouse/iceberg_source/mod.rs

laminar_connectors/lakehouse/iceberg_source/
mod.rs
