laminar_connectors/lakehouse/iceberg_source/
cursor.rs

1use iceberg::spec::Snapshot;
2use iceberg::table::Table;
3use serde::{Deserialize, Serialize};
4use std::fmt;
5
6use crate::checkpoint::SourceCheckpoint;
7use crate::error::ConnectorError;
8use crate::lakehouse::iceberg_config::{stable_catalog_identity, IcebergSourceConfig};
9
10const CURSOR_KEY: &str = "iceberg.cursor";
11const CURSOR_VERSION: u8 = 1;
12const MAX_CURSOR_BYTES: usize = 16 * 1024;
13
14/// Origin represented by an Iceberg source cursor.
15#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
16#[serde(rename_all = "snake_case")]
17pub enum IcebergSourceCursorOriginV1 {
18    /// A concrete Iceberg snapshot.
19    #[default]
20    Snapshot,
21    /// The durable table state before its first snapshot.
22    EmptyTable,
23}
24
25impl IcebergSourceCursorOriginV1 {
26    fn is_snapshot(&self) -> bool {
27        self == &Self::Snapshot
28    }
29}
30
31/// Replay position for an Iceberg snapshot lineage.
32#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
33#[serde(deny_unknown_fields)]
34pub struct IcebergSourceCursorV1 {
35    /// Serialized cursor version.
36    pub version: u8,
37    /// Hash of non-secret catalog and storage identity.
38    pub catalog_identity: String,
39    /// Iceberg table UUID.
40    pub table_uuid: String,
41    /// Fully qualified namespace and table identifier.
42    pub table_identifier: String,
43    /// Iceberg ref followed by this source.
44    pub table_ref: String,
45    /// Durable origin of the replay position.
46    #[serde(
47        default,
48        skip_serializing_if = "IcebergSourceCursorOriginV1::is_snapshot"
49    )]
50    pub origin: IcebergSourceCursorOriginV1,
51    /// Last completely emitted snapshot.
52    pub snapshot_id: i64,
53    /// Sequence number of that snapshot.
54    pub sequence_number: i64,
55    /// Schema that binds the declared output columns to Iceberg field IDs.
56    pub read_schema_id: i32,
57    /// Metadata file observed when the cursor was created.
58    pub metadata_location: String,
59}
60
61impl fmt::Debug for IcebergSourceCursorV1 {
62    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
63        let table_identifier =
64            crate::security::sanitize_identity_value("table_identifier", &self.table_identifier);
65        let table_ref = crate::security::sanitize_identity_value("table_ref", &self.table_ref);
66        let metadata_location =
67            crate::security::sanitize_identity_value("metadata_location", &self.metadata_location);
68        formatter
69            .debug_struct("IcebergSourceCursorV1")
70            .field("version", &self.version)
71            .field("catalog_identity", &self.catalog_identity)
72            .field("table_uuid", &self.table_uuid)
73            .field("table_identifier", &table_identifier)
74            .field("table_ref", &table_ref)
75            .field("origin", &self.origin)
76            .field("snapshot_id", &self.snapshot_id)
77            .field("sequence_number", &self.sequence_number)
78            .field("read_schema_id", &self.read_schema_id)
79            .field("metadata_location", &metadata_location)
80            .finish()
81    }
82}
83
84impl IcebergSourceCursorV1 {
85    pub(super) fn from_snapshot(
86        config: &IcebergSourceConfig,
87        table: &Table,
88        snapshot: &Snapshot,
89        read_schema_id: i32,
90    ) -> Result<Self, ConnectorError> {
91        let cursor = Self {
92            version: CURSOR_VERSION,
93            catalog_identity: stable_catalog_identity(&config.catalog, &config.storage),
94            table_uuid: table.metadata().uuid().to_string(),
95            table_identifier: table_identifier(config),
96            table_ref: config.table_ref.clone(),
97            origin: IcebergSourceCursorOriginV1::Snapshot,
98            snapshot_id: snapshot.snapshot_id(),
99            sequence_number: snapshot.sequence_number(),
100            read_schema_id,
101            metadata_location: durable_metadata_location(table)?.to_string(),
102        };
103        cursor.encoded_payload()?;
104        Ok(cursor)
105    }
106
107    pub(super) fn from_empty_table(
108        config: &IcebergSourceConfig,
109        table: &Table,
110        read_schema_id: i32,
111    ) -> Result<Self, ConnectorError> {
112        let metadata_location = table
113            .metadata_location()
114            .filter(|location| !location.is_empty())
115            .ok_or_else(|| {
116                ConnectorError::FeatureUnsupported(
117                    "[LDB-ICEBERG-EMPTY-CURSOR-METADATA] append from an empty table requires a durable metadata location"
118                        .into(),
119                )
120            })?;
121        let cursor = Self {
122            version: CURSOR_VERSION,
123            catalog_identity: stable_catalog_identity(&config.catalog, &config.storage),
124            table_uuid: table.metadata().uuid().to_string(),
125            table_identifier: table_identifier(config),
126            table_ref: config.table_ref.clone(),
127            origin: IcebergSourceCursorOriginV1::EmptyTable,
128            snapshot_id: 0,
129            sequence_number: 0,
130            read_schema_id,
131            metadata_location: metadata_location.to_string(),
132        };
133        cursor.encoded_payload()?;
134        Ok(cursor)
135    }
136
137    pub(super) fn validate_binding(
138        &self,
139        config: &IcebergSourceConfig,
140        table: &Table,
141    ) -> Result<(), ConnectorError> {
142        if self.catalog_identity != stable_catalog_identity(&config.catalog, &config.storage) {
143            return Err(cursor_error(
144                "LDB-ICEBERG-CURSOR-CATALOG",
145                "checkpoint catalog identity differs from configured catalog",
146            ));
147        }
148        if self.table_uuid != table.metadata().uuid().to_string() {
149            return Err(cursor_error(
150                "LDB-ICEBERG-CURSOR-TABLE-UUID",
151                "checkpoint table UUID differs from loaded table",
152            ));
153        }
154        if self.table_identifier != table_identifier(config) {
155            return Err(cursor_error(
156                "LDB-ICEBERG-CURSOR-TABLE",
157                "checkpoint table identifier differs from configured table",
158            ));
159        }
160        if self.table_ref != config.table_ref {
161            return Err(cursor_error(
162                "LDB-ICEBERG-CURSOR-REF",
163                "checkpoint table ref differs from configured table ref",
164            ));
165        }
166        if self.is_empty_table() {
167            self.retained_schema(table)?;
168            return Ok(());
169        }
170        let snapshot = table
171            .metadata()
172            .snapshot_by_id(self.snapshot_id)
173            .ok_or_else(|| {
174                cursor_error(
175                    "LDB-ICEBERG-CURSOR-EXPIRED",
176                    "checkpoint snapshot is absent from retained table history",
177                )
178            })?;
179        if snapshot.sequence_number() != self.sequence_number {
180            return Err(cursor_error(
181                "LDB-ICEBERG-CURSOR-SEQUENCE",
182                "checkpoint snapshot sequence number differs from table metadata",
183            ));
184        }
185        self.retained_schema(table)?;
186        Ok(())
187    }
188
189    pub(super) fn retained_schema(
190        &self,
191        table: &Table,
192    ) -> Result<iceberg::spec::SchemaRef, ConnectorError> {
193        table
194            .metadata()
195            .schema_by_id(self.read_schema_id)
196            .cloned()
197            .ok_or_else(|| {
198                cursor_error(
199                    "LDB-ICEBERG-CURSOR-SCHEMA-EXPIRED",
200                    &format!(
201                        "checkpoint read schema {} is absent from retained table metadata",
202                        self.read_schema_id
203                    ),
204                )
205            })
206    }
207
208    /// Encodes this cursor as a connector checkpoint.
209    ///
210    /// # Errors
211    ///
212    /// Returns a serialization error if the cursor cannot be encoded.
213    pub fn to_checkpoint(&self) -> Result<SourceCheckpoint, ConnectorError> {
214        let encoded = self.encoded_payload()?;
215        let mut checkpoint = SourceCheckpoint::new();
216        checkpoint.set_offset(CURSOR_KEY, encoded);
217        checkpoint.set_metadata("connector_type", "iceberg");
218        Ok(checkpoint)
219    }
220
221    /// Decodes a versioned cursor from a connector checkpoint.
222    ///
223    /// # Errors
224    ///
225    /// Returns a stable configuration error for missing or unknown versions.
226    pub fn from_checkpoint(checkpoint: &SourceCheckpoint) -> Result<Self, ConnectorError> {
227        if checkpoint.get_metadata("connector_type") != Some("iceberg") {
228            return Err(cursor_error(
229                "LDB-ICEBERG-CURSOR-CONNECTOR",
230                "resume checkpoint is not bound to the Iceberg connector",
231            ));
232        }
233        let encoded = checkpoint.get_offset(CURSOR_KEY).ok_or_else(|| {
234            cursor_error(
235                "LDB-ICEBERG-CURSOR-MISSING",
236                "resume checkpoint has no versioned Iceberg cursor",
237            )
238        })?;
239        if encoded.len() > MAX_CURSOR_BYTES {
240            return Err(cursor_error(
241                "LDB-ICEBERG-CURSOR-SIZE",
242                "Iceberg cursor exceeds its fixed durability bound",
243            ));
244        }
245        let cursor: Self = serde_json::from_str(encoded).map_err(|_| {
246            cursor_error(
247                "LDB-ICEBERG-CURSOR-DECODE",
248                "Iceberg cursor is not valid versioned JSON",
249            )
250        })?;
251        cursor.validate_payload()?;
252        Ok(cursor)
253    }
254
255    fn encoded_payload(&self) -> Result<String, ConnectorError> {
256        self.validate_payload()?;
257        let encoded =
258            serde_json::to_string(self).map_err(|error| ConnectorError::Serde(error.into()))?;
259        if encoded.len() > MAX_CURSOR_BYTES {
260            return Err(cursor_error(
261                "LDB-ICEBERG-CURSOR-SIZE",
262                "Iceberg cursor exceeds its fixed durability bound",
263            ));
264        }
265        Ok(encoded)
266    }
267
268    fn validate_payload(&self) -> Result<(), ConnectorError> {
269        if self.version != CURSOR_VERSION {
270            return Err(cursor_error(
271                "LDB-ICEBERG-CURSOR-VERSION",
272                &format!("unsupported cursor version {}", self.version),
273            ));
274        }
275        if self.is_empty_table() && (self.snapshot_id != 0 || self.sequence_number != 0) {
276            return Err(cursor_error(
277                "LDB-ICEBERG-CURSOR-EMPTY-POSITION",
278                "empty-table cursor contains a snapshot or sequence position",
279            ));
280        }
281        if self.is_empty_table() && self.table_ref != "main" {
282            return Err(cursor_error(
283                "LDB-ICEBERG-CURSOR-EMPTY-REF",
284                "empty-table cursor must follow the main ref",
285            ));
286        }
287        if self.catalog_identity.len() != 64
288            || !self
289                .catalog_identity
290                .bytes()
291                .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
292        {
293            return Err(cursor_error(
294                "LDB-ICEBERG-CURSOR-CATALOG-IDENTITY",
295                "cursor catalog identity is not a canonical SHA-256 digest",
296            ));
297        }
298        let table_uuid = uuid::Uuid::parse_str(&self.table_uuid).map_err(|_| {
299            cursor_error(
300                "LDB-ICEBERG-CURSOR-TABLE-UUID-FORMAT",
301                "cursor table UUID is not canonical",
302            )
303        })?;
304        if table_uuid.is_nil() || table_uuid.to_string() != self.table_uuid {
305            return Err(cursor_error(
306                "LDB-ICEBERG-CURSOR-TABLE-UUID-FORMAT",
307                "cursor table UUID is not canonical",
308            ));
309        }
310        for (code, label, value) in [
311            (
312                "LDB-ICEBERG-CURSOR-TABLE-FORMAT",
313                "table identifier",
314                self.table_identifier.as_str(),
315            ),
316            (
317                "LDB-ICEBERG-CURSOR-REF-FORMAT",
318                "table ref",
319                self.table_ref.as_str(),
320            ),
321        ] {
322            if value.is_empty() || value.chars().any(char::is_control) {
323                return Err(cursor_error(
324                    code,
325                    &format!("cursor {label} is empty or contains control characters"),
326                ));
327            }
328        }
329        if self.metadata_location.is_empty()
330            || self.metadata_location.chars().any(char::is_control)
331            || crate::security::value_contains_uri_secret(&self.metadata_location, false)
332        {
333            return Err(cursor_error(
334                "LDB-ICEBERG-CURSOR-METADATA-LOCATION",
335                "cursor metadata location is not safe for durable state",
336            ));
337        }
338        Ok(())
339    }
340
341    pub(super) fn is_empty_table(&self) -> bool {
342        self.origin == IcebergSourceCursorOriginV1::EmptyTable
343    }
344}
345
346fn durable_metadata_location(table: &Table) -> Result<&str, ConnectorError> {
347    table
348        .metadata_location()
349        .filter(|location| !location.is_empty())
350        .ok_or_else(|| {
351            cursor_error(
352                "LDB-ICEBERG-CURSOR-METADATA-LOCATION",
353                "loaded table has no durable metadata location",
354            )
355        })
356}
357
358fn cursor_error(code: &str, detail: &str) -> ConnectorError {
359    ConnectorError::ConfigurationError(format!("[{code}] {detail}"))
360}
361
362fn table_identifier(config: &IcebergSourceConfig) -> String {
363    format!("{}.{}", config.catalog.namespace, config.catalog.table_name)
364}
365
366#[cfg(test)]
367mod tests {
368    use super::*;
369
370    fn cursor() -> IcebergSourceCursorV1 {
371        IcebergSourceCursorV1 {
372            version: 1,
373            catalog_identity: "0".repeat(64),
374            table_uuid: "018f0f9d-7b2f-7a61-b72d-f4be1c7f43e1".into(),
375            table_identifier: "ns.table".into(),
376            table_ref: "main".into(),
377            origin: IcebergSourceCursorOriginV1::Snapshot,
378            snapshot_id: 42,
379            sequence_number: 7,
380            read_schema_id: 3,
381            metadata_location: "s3://bucket/metadata/v7.json".into(),
382        }
383    }
384
385    fn unchecked_checkpoint(cursor: &IcebergSourceCursorV1) -> SourceCheckpoint {
386        let mut checkpoint = SourceCheckpoint::new();
387        checkpoint.set_offset(CURSOR_KEY, serde_json::to_string(cursor).unwrap());
388        checkpoint.set_metadata("connector_type", "iceberg");
389        checkpoint
390    }
391
392    #[test]
393    fn checkpoint_round_trip_is_versioned() {
394        let cursor = cursor();
395        let checkpoint = cursor.to_checkpoint().unwrap();
396        assert_eq!(
397            IcebergSourceCursorV1::from_checkpoint(&checkpoint).unwrap(),
398            cursor
399        );
400    }
401
402    #[test]
403    fn empty_table_position_is_explicit_and_version_compatible() {
404        let regular = cursor();
405        let regular_json = serde_json::to_value(&regular).unwrap();
406        assert!(regular_json.get("origin").is_none());
407        assert_eq!(
408            IcebergSourceCursorV1::from_checkpoint(&unchecked_checkpoint(&regular)).unwrap(),
409            regular
410        );
411
412        let mut empty = cursor();
413        empty.origin = IcebergSourceCursorOriginV1::EmptyTable;
414        empty.snapshot_id = 0;
415        empty.sequence_number = 0;
416        assert_eq!(
417            IcebergSourceCursorV1::from_checkpoint(&empty.to_checkpoint().unwrap()).unwrap(),
418            empty
419        );
420
421        empty.snapshot_id = 1;
422        let error = IcebergSourceCursorV1::from_checkpoint(&unchecked_checkpoint(&empty))
423            .expect_err("empty position must not conceal a snapshot");
424        assert!(error
425            .to_string()
426            .contains("LDB-ICEBERG-CURSOR-EMPTY-POSITION"));
427
428        empty.snapshot_id = 0;
429        empty.table_ref = "audit".into();
430        let error = IcebergSourceCursorV1::from_checkpoint(&unchecked_checkpoint(&empty))
431            .expect_err("an empty cursor cannot invent a branch origin");
432        assert!(error.to_string().contains("LDB-ICEBERG-CURSOR-EMPTY-REF"));
433    }
434
435    #[test]
436    fn unknown_cursor_version_is_rejected() {
437        let mut cursor = cursor();
438        cursor.version = 2;
439        let error = IcebergSourceCursorV1::from_checkpoint(&unchecked_checkpoint(&cursor))
440            .expect_err("future cursor version must fail closed");
441        assert!(error.to_string().contains("LDB-ICEBERG-CURSOR-VERSION"));
442    }
443
444    #[test]
445    fn cursor_shape_and_size_fail_closed_without_echoing_payloads() {
446        let mut wrong_connector = cursor().to_checkpoint().unwrap();
447        wrong_connector.set_metadata("connector_type", "kafka");
448        assert!(IcebergSourceCursorV1::from_checkpoint(&wrong_connector)
449            .unwrap_err()
450            .to_string()
451            .contains("LDB-ICEBERG-CURSOR-CONNECTOR"));
452
453        let mut oversized = SourceCheckpoint::new();
454        oversized.set_metadata("connector_type", "iceberg");
455        oversized.set_offset(CURSOR_KEY, "x".repeat(MAX_CURSOR_BYTES + 1));
456        assert!(IcebergSourceCursorV1::from_checkpoint(&oversized)
457            .unwrap_err()
458            .to_string()
459            .contains("LDB-ICEBERG-CURSOR-SIZE"));
460
461        let mut encoded = serde_json::to_value(cursor()).unwrap();
462        encoded["oauth_client_secret=do-not-echo"] = serde_json::Value::Bool(true);
463        let mut unknown = SourceCheckpoint::new();
464        unknown.set_metadata("connector_type", "iceberg");
465        unknown.set_offset(CURSOR_KEY, serde_json::to_string(&encoded).unwrap());
466        let message = IcebergSourceCursorV1::from_checkpoint(&unknown)
467            .unwrap_err()
468            .to_string();
469        assert!(message.contains("LDB-ICEBERG-CURSOR-DECODE"));
470        assert!(!message.contains("do-not-echo"));
471    }
472
473    #[test]
474    fn credential_bearing_metadata_location_is_never_durable_or_echoed() {
475        let mut cursor = cursor();
476        cursor.metadata_location = "s3://catalog-user:do-not-echo@bucket/metadata/v7.json".into();
477        let debug = format!("{cursor:?}");
478        assert!(!debug.contains("catalog-user"));
479        assert!(!debug.contains("do-not-echo"));
480        let message = cursor.to_checkpoint().unwrap_err().to_string();
481        assert!(message.contains("LDB-ICEBERG-CURSOR-METADATA-LOCATION"));
482        assert!(!message.contains("do-not-echo"));
483
484        cursor.metadata_location.clear();
485        assert!(cursor
486            .to_checkpoint()
487            .unwrap_err()
488            .to_string()
489            .contains("LDB-ICEBERG-CURSOR-METADATA-LOCATION"));
490
491        let message = IcebergSourceCursorV1::from_checkpoint(&unchecked_checkpoint(&cursor))
492            .unwrap_err()
493            .to_string();
494        assert!(message.contains("LDB-ICEBERG-CURSOR-METADATA-LOCATION"));
495        assert!(!message.contains("do-not-echo"));
496    }
497
498    #[test]
499    fn noncanonical_cursor_identities_are_rejected() {
500        let mut invalid_catalog = cursor();
501        invalid_catalog.catalog_identity = "A".repeat(64);
502        assert!(
503            IcebergSourceCursorV1::from_checkpoint(&unchecked_checkpoint(&invalid_catalog))
504                .unwrap_err()
505                .to_string()
506                .contains("LDB-ICEBERG-CURSOR-CATALOG-IDENTITY")
507        );
508
509        let mut invalid_uuid = cursor();
510        invalid_uuid.table_uuid = uuid::Uuid::nil().to_string();
511        assert!(
512            IcebergSourceCursorV1::from_checkpoint(&unchecked_checkpoint(&invalid_uuid))
513                .unwrap_err()
514                .to_string()
515                .contains("LDB-ICEBERG-CURSOR-TABLE-UUID-FORMAT")
516        );
517    }
518
519    #[tokio::test]
520    async fn table_uuid_and_ref_mismatches_are_rejected() {
521        use crate::config::ConnectorConfig;
522        use crate::lakehouse::iceberg::test_support::{append_rows, create_test_table};
523
524        let fixture = create_test_table(false).await;
525        let (table, _) = append_rows(&fixture, &fixture.table, 1, &[(1, Some("a"))]).await;
526        let mut raw = ConnectorConfig::new("iceberg");
527        raw.set("catalog.uri", fixture.config.catalog.catalog_uri.clone());
528        raw.set(
529            "catalog.warehouse",
530            fixture.config.catalog.warehouse.clone(),
531        );
532        raw.set("namespace", "test");
533        raw.set("table.name", "events");
534        let config = IcebergSourceConfig::from_config(&raw).unwrap();
535        let snapshot = table.metadata().current_snapshot().unwrap();
536        let cursor = IcebergSourceCursorV1::from_snapshot(
537            &config,
538            &table,
539            snapshot,
540            table.metadata().current_schema_id(),
541        )
542        .unwrap();
543
544        let mut wrong_uuid = cursor.clone();
545        wrong_uuid.table_uuid = uuid::Uuid::now_v7().to_string();
546        assert!(wrong_uuid
547            .validate_binding(&config, &table)
548            .unwrap_err()
549            .to_string()
550            .contains("LDB-ICEBERG-CURSOR-TABLE-UUID"));
551
552        let mut wrong_ref = cursor;
553        wrong_ref.table_ref = "audit".into();
554        assert!(wrong_ref
555            .validate_binding(&config, &table)
556            .unwrap_err()
557            .to_string()
558            .contains("LDB-ICEBERG-CURSOR-REF"));
559
560        let mut missing_schema = IcebergSourceCursorV1::from_snapshot(
561            &config,
562            &table,
563            snapshot,
564            table.metadata().current_schema_id(),
565        )
566        .unwrap();
567        missing_schema.read_schema_id = i32::MAX;
568        assert!(missing_schema
569            .validate_binding(&config, &table)
570            .unwrap_err()
571            .to_string()
572            .contains("LDB-ICEBERG-CURSOR-SCHEMA-EXPIRED"));
573    }
574
575    #[tokio::test]
576    async fn snapshot_cursor_requires_durable_metadata_location() {
577        use crate::config::ConnectorConfig;
578        use crate::lakehouse::iceberg::test_support::{append_rows, create_test_table};
579
580        let fixture = create_test_table(false).await;
581        let (table, _) = append_rows(&fixture, &fixture.table, 1, &[(1, Some("a"))]).await;
582        let mut raw = ConnectorConfig::new("iceberg");
583        raw.set("catalog.uri", fixture.config.catalog.catalog_uri.clone());
584        raw.set(
585            "catalog.warehouse",
586            fixture.config.catalog.warehouse.clone(),
587        );
588        raw.set("namespace", "test");
589        raw.set("table.name", "events");
590        let config = IcebergSourceConfig::from_config(&raw).unwrap();
591        let table_without_location = Table::builder()
592            .identifier(table.identifier().clone())
593            .metadata(table.metadata().clone())
594            .file_io(table.file_io().clone())
595            .runtime(iceberg::Runtime::try_current().unwrap())
596            .build()
597            .unwrap();
598        let snapshot = table_without_location
599            .metadata()
600            .current_snapshot()
601            .unwrap();
602        let error = IcebergSourceCursorV1::from_snapshot(
603            &config,
604            &table_without_location,
605            snapshot,
606            table_without_location.metadata().current_schema_id(),
607        )
608        .unwrap_err();
609
610        assert!(error
611            .to_string()
612            .contains("LDB-ICEBERG-CURSOR-METADATA-LOCATION"));
613
614        let error = IcebergSourceCursorV1::from_empty_table(
615            &config,
616            &table_without_location,
617            table_without_location.metadata().current_schema_id(),
618        )
619        .unwrap_err();
620        assert!(error
621            .to_string()
622            .contains("LDB-ICEBERG-EMPTY-CURSOR-METADATA"));
623    }
624}
laminar_connectors/lakehouse/iceberg_source/cursor.rs

laminar_connectors/lakehouse/iceberg_source/
cursor.rs
