1#[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
59pub 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 #[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 start(&mut self, request: SourceStart) -> Result<(), ConnectorError> {
476 let (config, position, _) = request.into_parts();
477 #[cfg(feature = "iceberg-core")]
478 let declared_schema = if config.get("_arrow_schema").is_some() {
479 let schema = config.arrow_schema().ok_or_else(|| {
480 ConnectorError::ConfigurationError(
481 "invalid Iceberg source '_arrow_schema' encoding".into(),
482 )
483 })?;
484 if schema.fields().is_empty() {
485 return Err(ConnectorError::ConfigurationError(
486 "Iceberg source '_arrow_schema' must declare at least one column".into(),
487 ));
488 }
489 Some(schema)
490 } else {
491 None
492 };
493 if !config.properties().is_empty() {
494 self.config = IcebergSourceConfig::from_config(&config)?;
495 }
496 super::iceberg::capabilities::validate_source(&self.config)?;
497 if matches!(&position, SourcePosition::Resume { .. })
498 && self.config.read_mode != IcebergReadMode::Append
499 {
500 return Err(ConnectorError::ConfigurationError(
501 "only Iceberg read.mode=append has a replay cursor".into(),
502 ));
503 }
504
505 #[cfg(feature = "iceberg-core")]
506 {
507 let recovered_cursor = match &position {
508 SourcePosition::Initial => None,
509 SourcePosition::Resume { checkpoint, .. } => {
510 Some(IcebergSourceCursorV1::from_checkpoint(checkpoint)?)
511 }
512 };
513 let catalog = super::iceberg_io::build_catalog_for_access_with_metrics(
514 &self.config.catalog,
515 &self.config.storage,
516 super::iceberg_io::CatalogAccess::Read,
517 Some(self.metrics.credential_refresh_failures.clone()),
518 )
519 .await?
520 .catalog;
521 let table = super::iceberg_io::load_table_with_timeout(
522 catalog.as_ref(),
523 &self.config.catalog.namespace,
524 &self.config.catalog.table_name,
525 self.config.catalog.request_timeout,
526 )
527 .await?;
528 let root_schema = if let Some(cursor) = recovered_cursor.as_ref() {
529 cursor.validate_binding(&self.config, &table)?;
530 append_lineage::validate_cursor_lineage(
531 &table,
532 cursor,
533 self.config.max_snapshots_per_poll,
534 )?;
535 cursor.retained_schema(&table)?
536 } else {
537 match Self::selected_snapshot(&table, &self.config)? {
538 Some(snapshot) => snapshot.schema(table.metadata()).map_err(|error| {
539 super::iceberg_scan::connector_scan_error(
540 "resolve initial Iceberg snapshot schema",
541 &error,
542 )
543 })?,
544 None => table.current_schema_ref(),
545 }
546 };
547 self.bind_read_schema(&root_schema, declared_schema)?;
548 self.catalog = Some(catalog);
549 let head = table
550 .metadata()
551 .snapshot_for_ref(&self.config.table_ref)
552 .map(|snapshot| (snapshot.snapshot_id(), snapshot.sequence_number()));
553 self.observe_table_head(head);
554 self.metrics
555 .observe_table(table.metadata(), &self.config.table_ref);
556 self.table = Some(table);
557
558 match position {
559 SourcePosition::Initial => self.start_initial_scan()?,
560 SourcePosition::Resume { .. } => {
561 self.install_cursor(recovered_cursor.ok_or_else(|| {
562 ConnectorError::Internal(
563 "Iceberg resume cursor was not retained during startup".into(),
564 )
565 })?)?;
566 }
567 }
568 self.state = ConnectorState::Running;
569 info!(
570 operation_class = "connector-open",
571 storage_provider = %self.config.storage.diagnostic_provider(&self.config.catalog.warehouse),
572 storage_endpoint_class = %self.config.storage.diagnostic_endpoint_class(&self.config.catalog.warehouse),
573 storage_auth_source = %self.config.storage.diagnostic_auth_source(&self.config.catalog.warehouse),
574 request_timeout_ms = self.config.storage.request_timeout.as_millis(),
575 success = true,
576 table = self.config.catalog.table_name,
577 namespace = self.config.catalog.namespace,
578 mode = %self.config.read_mode,
579 "iceberg source connected"
580 );
581 return Ok(());
582 }
583
584 #[cfg(not(feature = "iceberg-core"))]
585 {
586 self.state = ConnectorState::Failed;
587 Err(ConnectorError::FeatureUnsupported(
588 "Apache Iceberg requires the 'iceberg' feature".into(),
589 ))
590 }
591 }
592
593 async fn poll_batch(
594 &mut self,
595 max_records: usize,
596 ) -> Result<Option<SourceBatch>, ConnectorError> {
597 if max_records == 0 {
598 return Err(ConnectorError::ConfigurationError(
599 "Iceberg source max_records must be greater than zero".into(),
600 ));
601 }
602 #[cfg(feature = "iceberg-core")]
603 {
604 if self.state == ConnectorState::Failed {
605 return if self.replay_unit_in_progress {
606 Err(ConnectorError::TransactionError(PARTIAL_SCAN_ERROR.into()))
607 } else {
608 Err(ConnectorError::InvalidState {
609 expected: "Running".into(),
610 actual: self.state.to_string(),
611 })
612 };
613 }
614 loop {
615 if let Some(batch) = self.emit_pending(max_records)? {
616 return Ok(Some(batch));
617 }
618 if let Some(output) = self.next_scan_output().await? {
619 match output {
620 ScanOutput::Batch {
621 batch,
622 completed_cursor,
623 } => {
624 self.metrics.observe_batch(&batch);
625 self.pending = Some(PendingBatch {
626 batch,
627 completed_cursor,
628 });
629 }
630 ScanOutput::Cursor(cursor) => self.install_cursor(cursor)?,
631 ScanOutput::ReadMetrics {
632 files,
633 storage_bytes,
634 } => self.metrics.observe_completed_read(files, storage_bytes),
635 }
636 continue;
637 }
638 if self.bounded_snapshot_complete {
639 return Ok(None);
640 }
641 if self.config.read_mode == IcebergReadMode::Snapshot {
642 self.start_initial_scan()?;
643 if self.scan.is_some() {
644 continue;
645 }
646 return Ok(None);
647 }
648 if self.config.read_mode == IcebergReadMode::Append {
649 self.refresh_append().await?;
650 if self.scan.is_some() {
651 continue;
652 }
653 }
654 return Ok(None);
655 }
656 }
657 #[cfg(not(feature = "iceberg-core"))]
658 {
659 Err(ConnectorError::FeatureUnsupported(
660 "Apache Iceberg requires the 'iceberg' feature".into(),
661 ))
662 }
663 }
664
665 fn schema(&self) -> SchemaRef {
666 self.schema
667 .clone()
668 .unwrap_or_else(|| Arc::new(arrow_schema::Schema::empty()))
669 }
670
671 fn checkpoint(&self) -> SourceCheckpoint {
672 self.checkpoint.clone()
673 }
674
675 fn try_checkpoint(&self) -> Result<Option<SourceCheckpoint>, ConnectorError> {
676 #[cfg(feature = "iceberg-core")]
677 if self.replay_unit_in_progress {
678 return Ok(None);
679 }
680 Ok(Some(self.checkpoint()))
681 }
682
683 fn checkpoint_unavailable_policy(&self) -> SourceCheckpointUnavailablePolicy {
684 SourceCheckpointUnavailablePolicy::PollToReplayBoundary
685 }
686
687 fn contract(&self, config: &ConnectorConfig) -> Result<SourceContract, ConnectorError> {
688 let parsed = if config.properties().is_empty() {
689 self.config.clone()
690 } else {
691 IcebergSourceConfig::from_config(config)?
692 };
693 super::iceberg::capabilities::validate_source(&parsed)?;
694 Ok(SourceContract::new(
695 if parsed.read_mode == IcebergReadMode::Append {
696 SourceConsistency::Replayable
697 } else {
698 SourceConsistency::Ephemeral
699 },
700 SourceTopology::Singleton,
701 SourceInputMode::AppendOnly,
702 ))
703 }
704
705 async fn close(&mut self) -> Result<(), ConnectorError> {
706 #[cfg(feature = "iceberg-core")]
707 {
708 if let Some(scan) = self.scan.take() {
709 drop(scan.receiver);
710 scan.handle.abort();
711 let _ = scan.handle.await;
712 }
713 self.catalog = None;
714 self.table = None;
715 self.read_schema = None;
716 self.filter_predicate = None;
717 self.pending = None;
718 self.replay_unit_in_progress = false;
719 self.scan_failed = false;
720 }
721 self.state = ConnectorState::Closed;
722 Ok(())
723 }
724}
725
726#[cfg(test)]
727mod tests;