From ff18a6d87553b09efd06a312e310fe156735a425 Mon Sep 17 00:00:00 2001 From: Swanand Mulay <73115739+swanandx@users.noreply.github.com> Date: Wed, 29 Jul 2026 19:19:44 +0530 Subject: [PATCH 1/5] [connectors] Implement Iceberg input follow mode Add `follow` and `snapshot_and_follow` modes to the Iceberg input connector. After the starting snapshot, the connector polls the catalog for new snapshots and, for each one, diffs it against its parent to find added and removed data files, ingesting added rows as inserts and removed rows as deletes. Only the manifests a snapshot added are read, so each step costs work proportional to the change, not to the table size. Follow mode requires a catalog: `metadata_location` points at a fixed snapshot and cannot observe new commits. currently handles copy-on-write only and rejects merge-on-read delete files with an actionable error; compaction snapshots (`operation = replace`) are skipped. Each followed snapshot forms one transaction and one checkpointable resume point, so a restart resumes after the last fully ingested snapshot. Signed-off-by: Swanand Mulay <73115739+swanandx@users.noreply.github.com> --- crates/adapters/Cargo.toml | 3 + crates/adapters/src/test.rs | 3 +- crates/adapters/src/test/iceberg.rs | 275 +++++++- crates/iceberg/src/input.rs | 619 ++++++++++++++++-- crates/iceberg/src/test/README.md | 33 + crates/iceberg/src/test/follow_table.py | 143 ++++ .../docs/connectors/sources/iceberg.md | 29 +- 7 files changed, 1049 insertions(+), 56 deletions(-) create mode 100644 crates/iceberg/src/test/follow_table.py diff --git a/crates/adapters/Cargo.toml b/crates/adapters/Cargo.toml index 576af7b0a4a..7a186317e0e 100644 --- a/crates/adapters/Cargo.toml +++ b/crates/adapters/Cargo.toml @@ -67,6 +67,9 @@ iceberg-tests-fs = [] iceberg-tests-glue = [] iceberg-tests-rest = [] iceberg-tests-s3tables = [] +# Follow-mode tests. Need a REST catalog and an S3 store both the writer +# (pyiceberg) and the connector can reach. See crates/iceberg/src/test/README.md. +iceberg-tests-follow = [] fips = ["rustls/fips"] bench-mode = [] with-postgres-cdc = ["etl", "etl-config", "etl-postgres"] diff --git a/crates/adapters/src/test.rs b/crates/adapters/src/test.rs index 4e64e0c3f46..a2819518fa5 100644 --- a/crates/adapters/src/test.rs +++ b/crates/adapters/src/test.rs @@ -50,7 +50,8 @@ mod datagen; feature = "iceberg-tests-fs", feature = "iceberg-tests-glue", feature = "iceberg-tests-rest", - feature = "iceberg-tests-s3tables" + feature = "iceberg-tests-s3tables", + feature = "iceberg-tests-follow" ) ))] mod iceberg; diff --git a/crates/adapters/src/test/iceberg.rs b/crates/adapters/src/test/iceberg.rs index cb5a31b7d1d..011ebc5cccb 100644 --- a/crates/adapters/src/test/iceberg.rs +++ b/crates/adapters/src/test/iceberg.rs @@ -7,7 +7,7 @@ use crate::{ use crossbeam::channel::Receiver; use dbsp::DBData; use feldera_sqllib::Variant; -#[cfg(feature = "iceberg-tests-fs")] +#[cfg(any(feature = "iceberg-tests-fs", feature = "iceberg-tests-follow"))] use feldera_sqllib::{ByteArray, F32, F64, Timestamp, TimestampTz}; use feldera_types::{ program_schema::Field, @@ -15,12 +15,25 @@ use feldera_types::{ }; use serde_json::json; -use std::{collections::HashMap, time::Instant}; +use std::collections::HashMap; +#[cfg(any( + feature = "iceberg-tests-fs", + feature = "iceberg-tests-glue", + feature = "iceberg-tests-rest", + feature = "iceberg-tests-s3tables" +))] +use std::time::Instant; use tempfile::NamedTempFile; +#[cfg(any( + feature = "iceberg-tests-fs", + feature = "iceberg-tests-glue", + feature = "iceberg-tests-rest", + feature = "iceberg-tests-s3tables" +))] use tracing::info; use tracing_subscriber::{EnvFilter, layer::SubscriberExt, util::SubscriberInitExt}; -#[cfg(feature = "iceberg-tests-fs")] +#[cfg(any(feature = "iceberg-tests-fs", feature = "iceberg-tests-follow"))] use std::io::Write; #[cfg(feature = "iceberg-tests-fs")] @@ -28,7 +41,8 @@ use super::IcebergSubsetTestStruct; #[cfg(any( feature = "iceberg-tests-fs", feature = "iceberg-tests-glue", - feature = "iceberg-tests-rest" + feature = "iceberg-tests-rest", + feature = "iceberg-tests-follow" ))] use super::IcebergTestStruct; #[cfg(feature = "iceberg-tests-s3tables")] @@ -46,7 +60,7 @@ fn init_logging() { .try_init(); } -#[cfg(feature = "iceberg-tests-fs")] +#[cfg(any(feature = "iceberg-tests-fs", feature = "iceberg-tests-follow"))] /// Store test dataset in an ndjson file fn data_to_ndjson(data: Vec) -> NamedTempFile { println!("delta_table_output_test: preparing input file"); @@ -93,6 +107,12 @@ fn iceberg_connector_metrics(pipeline: &Controller) -> HashMap { /// `config` is the connector's transport config as a JSON object. This function /// forces `mode = snapshot`. Returns the output file and the connector's custom /// metrics captured just before the pipeline is stopped. +#[cfg(any( + feature = "iceberg-tests-fs", + feature = "iceberg-tests-glue", + feature = "iceberg-tests-rest", + feature = "iceberg-tests-s3tables" +))] fn iceberg_snapshot_to_json( schema: &[Field], table_properties: &[(&str, &str)], @@ -224,7 +244,7 @@ where } /// Generate up to `max_records` _unique_ records. -#[cfg(feature = "iceberg-tests-fs")] +#[cfg(any(feature = "iceberg-tests-fs", feature = "iceberg-tests-follow"))] fn data(n_records: usize) -> Vec { let mut result = Vec::with_capacity(n_records); @@ -700,3 +720,246 @@ fn iceberg_rest_s3_input_test() { assert_eq!(zset.len(), 2000000); //assert_eq!(zset, expected_zset); } + +// --------------------------------------------------------------------------- +// Follow-mode tests (feature `iceberg-tests-follow`). +// +// These need a REST catalog and an S3 store that both the writer (pyiceberg) +// and the connector (iceberg-rust) can reach. Defaults target the local docker +// setup in crates/iceberg/src/test/README.md; override via FELDERA_ICEBERG_*. +// --------------------------------------------------------------------------- + +#[cfg(feature = "iceberg-tests-follow")] +fn env_or(key: &str, default: &str) -> String { + std::env::var(key).unwrap_or_else(|_| default.to_string()) +} + +/// Connector transport config for a REST-catalog table `mode` on the test store. +#[cfg(feature = "iceberg-tests-follow")] +fn rest_follow_config(table: &str, mode: &str) -> serde_json::Value { + json!({ + "mode": mode, + "catalog_type": "rest", + "rest.uri": env_or("FELDERA_ICEBERG_REST_URI", "http://localhost:8181"), + "rest.warehouse": env_or("FELDERA_ICEBERG_WAREHOUSE", "s3://test/iceberg-follow"), + "table_name": table, + "s3.endpoint": env_or("FELDERA_ICEBERG_S3_ENDPOINT", "http://localhost:9000"), + "s3.access-key-id": env_or("FELDERA_ICEBERG_S3_KEY", "minio"), + "s3.secret-access-key": env_or("FELDERA_ICEBERG_S3_SECRET", "miniopasswd"), + "s3.region": env_or("FELDERA_ICEBERG_S3_REGION", "us-east-1"), + // MinIO (and most non-AWS S3) serve path-style URLs; the opendal S3 + // backend defaults to virtual-host style, so opt out explicitly. + "s3.path-style-access": env_or("FELDERA_ICEBERG_S3_PATH_STYLE", "true"), + }) +} + +/// The follow connector config for `table` with an extra `key = value` option +/// merged in (e.g. a `snapshot_id` follow start point). +#[cfg(feature = "iceberg-tests-follow")] +fn rest_follow_config_with( + table: &str, + mode: &str, + key: &str, + value: serde_json::Value, +) -> serde_json::Value { + let mut config = rest_follow_config(table, mode); + config + .as_object_mut() + .expect("iceberg connector config must be a JSON object") + .insert(key.to_string(), value); + config +} + +/// Create (`op = "create"`) or append to (`op = "append"`) the REST-catalog test +/// table with `chunk`, producing a new snapshot. Returns the new snapshot's id so +/// a test can pin it as a follow start point. Shells out to the pyiceberg helper, +/// since iceberg-rust cannot write tables. +#[cfg(feature = "iceberg-tests-follow")] +fn follow_table_op(op: &str, table: &str, chunk: &[IcebergTestStruct]) -> i64 { + let ndjson = data_to_ndjson(chunk.to_vec()); + let script = "../iceberg/src/test/follow_table.py"; + let python = env_or("FELDERA_ICEBERG_PYTHON", "python3"); + let output = std::process::Command::new(python) + .arg(script) + .arg(format!("--op={op}")) + .arg(format!("--table={table}")) + .arg(format!("--json-file={}", ndjson.path().display())) + .output() + .unwrap_or_else(|e| panic!("failed to run '{script}': {e}")); + if !output.status.success() { + panic!( + "'{script} --op={op}' failed (status {}):\nstdout: {}\nstderr: {}", + output.status, + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr), + ); + } + // The writer prints the new snapshot id on its last stdout line. + let stdout = String::from_utf8_lossy(&output.stdout); + let last = stdout.trim().lines().last().unwrap_or_default(); + last.parse::() + .unwrap_or_else(|_| panic!("'{script} --op={op}' printed unexpected output: {last:?}")) +} + +/// Sum of the records the connector has ingested so far (snapshot phase plus +/// follow phase), read from its custom metrics. +#[cfg(feature = "iceberg-tests-follow")] +fn ingested_records(pipeline: &Controller) -> u64 { + let metrics = iceberg_connector_metrics(pipeline); + let get = |name: &str| metrics.get(name).copied().unwrap_or(0.0) as u64; + get("input_connector_iceberg_snapshot_records_total") + + get("input_connector_iceberg_follow_records_total") +} + +/// Run `body` against a running follow-mode pipeline, then stop it. The output +/// file receives `insert_delete` JSON. +#[cfg(feature = "iceberg-tests-follow")] +fn with_follow_pipeline(table: &str, mode: &str, body: F) +where + F: FnOnce(&Controller, &std::path::Path), +{ + with_follow_pipeline_cfg(rest_follow_config(table, mode), body) +} + +/// Like [`with_follow_pipeline`], but takes a fully built connector config so a +/// test can add options such as a `snapshot_id` follow start point. +#[cfg(feature = "iceberg-tests-follow")] +fn with_follow_pipeline_cfg(config: serde_json::Value, body: F) +where + F: FnOnce(&Controller, &std::path::Path), +{ + let json_file = NamedTempFile::new().unwrap(); + let (pipeline, err_receiver) = iceberg_input_pipeline::( + &IcebergTestStruct::schema_with_lateness(), + &[], + config, + &json_file.path().display().to_string(), + ); + pipeline.start(); + + // Surface connector errors promptly instead of hanging until a timeout. + let watch = err_receiver.clone(); + let guard = std::thread::spawn(move || { + if let Ok(msg) = watch.recv() { + panic!("follow pipeline reported an error: {msg}"); + } + }); + + body(&pipeline, json_file.path()); + + assert!(err_receiver.is_empty(), "follow pipeline reported errors"); + pipeline.stop().unwrap(); + drop(guard); +} + +/// `snapshot_and_follow`: the initial snapshot is read, then a second snapshot +/// committed before startup is caught up via follow. Every row lands once. +#[test] +#[cfg(feature = "iceberg-tests-follow")] +fn iceberg_rest_follow_snapshot_and_follow() { + use dbsp::trace::BatchReader; + + let all = data(10); + let ns = env_or("FELDERA_ICEBERG_NAMESPACE", "follow_ns"); + let table = &format!("{ns}.snapshot_and_follow"); + + // Two snapshots exist before the connector starts. + follow_table_op("create", table, &all[..5]); + follow_table_op("append", table, &all[5..]); + + with_follow_pipeline(table, "snapshot_and_follow", |pipeline, out_path| { + wait(|| ingested_records(pipeline) >= 10, 120_000) + .expect("timed out ingesting snapshot + follow"); + // Let the output connector flush the ingested rows. + wait(|| output_record_count(out_path) >= 10, 60_000).expect("timed out writing output"); + + let zset = output_zset(out_path); + assert_eq!(zset.len(), 10); + assert_eq!(zset, expected_zset(&all)); + }); +} + +/// `follow`: no initial snapshot; a snapshot committed after startup is tailed. +/// Rows present before the start snapshot are not ingested. +#[test] +#[cfg(feature = "iceberg-tests-follow")] +fn iceberg_rest_follow_live_append() { + use dbsp::trace::BatchReader; + + let all = data(10); + let ns = env_or("FELDERA_ICEBERG_NAMESPACE", "follow_ns"); + let table = &format!("{ns}.live_append"); + + // The connector starts following after this snapshot, so these rows are + // not ingested. + follow_table_op("create", table, &all[..5]); + + with_follow_pipeline(table, "follow", |pipeline, out_path| { + // Nothing to ingest until a new snapshot appears. + follow_table_op("append", table, &all[5..]); + + wait(|| ingested_records(pipeline) >= 5, 120_000) + .expect("timed out tailing the appended snapshot"); + wait(|| output_record_count(out_path) >= 5, 60_000).expect("timed out writing output"); + + let zset = output_zset(out_path); + assert_eq!(zset.len(), 5); + assert_eq!(zset, expected_zset(&all[5..])); + }); +} + +/// `follow` with an explicit `snapshot_id` start point. Three snapshots (A, B, +/// C) exist before the connector starts; following from B must ingest only C's +/// rows, never A's or B's, since `snapshots_after` walks the ancestry back to B. +#[test] +#[cfg(feature = "iceberg-tests-follow")] +fn iceberg_rest_follow_start_from_snapshot_id() { + use dbsp::trace::BatchReader; + + let all = data(15); + let ns = env_or("FELDERA_ICEBERG_NAMESPACE", "follow_ns"); + let table = &format!("{ns}.start_from_id"); + + follow_table_op("create", table, &all[..5]); // snapshot A + let snapshot_b = follow_table_op("append", table, &all[5..10]); // snapshot B + follow_table_op("append", table, &all[10..]); // snapshot C + + let config = rest_follow_config_with(table, "follow", "snapshot_id", json!(snapshot_b)); + with_follow_pipeline_cfg(config, |pipeline, out_path| { + wait(|| ingested_records(pipeline) >= 5, 120_000) + .expect("timed out following from the pinned snapshot"); + wait(|| output_record_count(out_path) >= 5, 60_000).expect("timed out writing output"); + + let zset = output_zset(out_path); + assert_eq!(zset.len(), 5); + assert_eq!(zset, expected_zset(&all[10..])); + }); +} + +/// Number of complete `insert_delete` records written to the output file so +/// far (one JSON value per line). +#[cfg(feature = "iceberg-tests-follow")] +fn output_record_count(path: &std::path::Path) -> usize { + std::fs::read(path) + .map(|bytes| bytes.iter().filter(|&&b| b == b'\n').count()) + .unwrap_or(0) +} + +/// The zset of `insert_delete` records currently in the output file. +#[cfg(feature = "iceberg-tests-follow")] +fn output_zset(path: &std::path::Path) -> dbsp::OrdZSet { + let mut file = std::fs::File::open(path).unwrap(); + file_to_zset::(&mut file) +} + +/// The all-`+1` zset the connector should produce for `data`. +#[cfg(feature = "iceberg-tests-follow")] +fn expected_zset(data: &[IcebergTestStruct]) -> dbsp::OrdZSet { + dbsp::OrdZSet::from_tuples( + (), + data.iter() + .cloned() + .map(|x| dbsp::utils::Tup2(dbsp::utils::Tup2(x, ()), 1)) + .collect(), + ) +} diff --git a/crates/iceberg/src/input.rs b/crates/iceberg/src/input.rs index 7c35e78b52d..91bac7db2db 100644 --- a/crates/iceberg/src/input.rs +++ b/crates/iceberg/src/input.rs @@ -33,13 +33,18 @@ use feldera_types::{ program_schema::{Field, Relation}, transport::iceberg::{IcebergCatalogType, IcebergReaderConfig, IcebergTransactionMode}, }; -use futures_util::StreamExt; +use futures_util::{stream, StreamExt}; use iceberg::CatalogBuilder; use iceberg::{ + arrow::ArrowReaderBuilder, io::{FileIO, FileIOBuilder, StorageFactory}, - spec::SnapshotRef, + scan::{FileScanTask, FileScanTaskStream}, + spec::{ + DataContentType, DataFile, ManifestStatus, NameMapping, Operation, + SchemaRef as IcebergSchemaRef, SnapshotRef, TableMetadata, DEFAULT_SCHEMA_NAME_MAPPING, + }, table::{StaticTable, Table as IcebergTable}, - Catalog, TableIdent, + Catalog, Runtime, TableIdent, }; use iceberg_catalog_glue::{ GlueCatalogBuilder, AWS_ACCESS_KEY_ID, AWS_PROFILE_NAME, AWS_REGION_NAME, @@ -186,7 +191,6 @@ enum IcebergPhase { } /// Prometheus-style metrics exported by the Iceberg input connector. -// TODO(#6165): follow-mode metrics land with follow mode. struct IcebergMetrics { /// Current phase of the connector (see [`IcebergPhase`]). phase: Atomic, @@ -199,6 +203,10 @@ struct IcebergMetrics { /// Sequence number of the ingested Iceberg snapshot; /// [`SEQUENCE_METRIC_UNSET`] until the snapshot has been read. last_ingested_sequence_number: AtomicU64, + /// Number of Iceberg snapshots ingested in follow mode. + follow_snapshots_total: AtomicU64, + /// Total records ingested in follow mode (inserts plus deletes). + follow_records_total: AtomicU64, } /// Sentinel stored in the sequence-number gauge before a value is available. @@ -212,6 +220,8 @@ impl IcebergMetrics { snapshot_records_total: AtomicU64::new(0), snapshot_transaction_starts: AtomicU64::new(0), last_ingested_sequence_number: AtomicU64::new(SEQUENCE_METRIC_UNSET), + follow_snapshots_total: AtomicU64::new(0), + follow_records_total: AtomicU64::new(0), } } @@ -269,6 +279,18 @@ impl ConnectorMetrics for IcebergMetrics { ValueType::Gauge, self.last_ingested_sequence_number_metric(), ), + ( + "input_connector_iceberg_follow_snapshots_total", + "Number of Iceberg snapshots ingested in follow mode.", + ValueType::Counter, + self.follow_snapshots_total.load(Ordering::Relaxed) as f64, + ), + ( + "input_connector_iceberg_follow_records_total", + "Total records ingested in follow mode (inserts plus deletes).", + ValueType::Counter, + self.follow_records_total.load(Ordering::Relaxed) as f64, + ), ] } } @@ -365,6 +387,16 @@ impl IcebergResumeInfo { } } + /// Resume point after fully ingesting follow snapshot `snapshot_id`: resume + /// by following the table after it. `eoi` is false; follow mode never ends. + fn follow(snapshot_id: i64) -> Self { + Self { + snapshot_id: Some(snapshot_id), + snapshot_timestamp: None, + eoi: false, + } + } + fn to_resume(&self) -> Resume { Resume::Seek { seek: serde_json::to_value(self).unwrap(), @@ -388,6 +420,85 @@ enum QueueEntry { Rollback, } +/// How often follow mode polls the catalog for new snapshots. +const FOLLOW_POLL_INTERVAL: std::time::Duration = std::time::Duration::from_secs(1); + +/// A data file a snapshot added or removed. `DataFile` omits its partition-spec +/// id, so carry it from the manifest: the Arrow reader needs it to fill in +/// identity-partition columns kept in partition metadata, not in the data file. +struct ChangedFile { + data_file: DataFile, + partition_spec_id: i32, +} + +/// One snapshot's changelog versus its predecessor: files it added (rows to +/// insert) and files it removed (rows to delete). +struct SnapshotDelta { + added: Vec, + removed: Vec, +} + +impl SnapshotDelta { + fn is_empty(&self) -> bool { + self.added.is_empty() && self.removed.is_empty() + } +} + +/// Snapshots committed after `after_id`, oldest first, found by walking the +/// parent chain back from the current snapshot. `None` returns the whole +/// ancestry. Errors if `after_id` is not an ancestor (it was expired or the +/// table was rolled back), since the connector can no longer follow it. +fn snapshots_after( + metadata: &TableMetadata, + after_id: Option, +) -> Result, AnyError> { + let mut chain = Vec::new(); + let mut current = metadata.current_snapshot().cloned(); + while let Some(snapshot) = current { + if Some(snapshot.snapshot_id()) == after_id { + chain.reverse(); + return Ok(chain); + } + let parent = snapshot.parent_snapshot_id(); + chain.push(snapshot); + current = parent.and_then(|id| metadata.snapshot_by_id(id).cloned()); + } + + match after_id { + // No pinned snapshot: the whole ancestry is new. + None => { + chain.reverse(); + Ok(chain) + } + Some(after_id) => bail!( + "pinned Iceberg snapshot {after_id} is no longer part of the table's current history; it may have been expired or the table may have been rolled back. Restart the connector to re-read from the current snapshot." + ), + } +} + +/// Whether a snapshot only rewrites files without changing table contents +/// (compaction / manifest rewrite, `operation = replace`). Follow mode skips +/// such snapshots: diffing them would produce insert/delete churn that cancels. +fn is_noop_operation(snapshot: &SnapshotRef) -> bool { + snapshot.summary().operation == Operation::Replace +} + +/// Parse the table's default name mapping (property +/// `schema.name-mapping.default`), used by the reader to resolve field ids for +/// Parquet files that store column names but not field ids. +fn table_name_mapping(table: &IcebergTable) -> Result>, AnyError> { + table + .metadata() + .properties() + .get(DEFAULT_SCHEMA_NAME_MAPPING) + .map(|json| serde_json::from_str::(json)) + .transpose() + .map(|mapping| mapping.map(Arc::new)) + .map_err(|e| { + anyhow!("error parsing the table's '{DEFAULT_SCHEMA_NAME_MAPPING}' property: {e}") + }) +} + /// Integrated input connector that reads from an Iceberg table. pub struct IcebergInputEndpoint { inner: Arc, @@ -458,8 +569,11 @@ impl IcebergInputReader { bail!("invalid Iceberg connector configuration: 'num_parsers' must be greater than 0"); } - if endpoint.config.follow() { - bail!("'{}' mode is not yet supported", endpoint.config.mode); + if endpoint.config.follow() && endpoint.config.catalog_type.is_none() { + bail!( + "'{}' mode requires an Iceberg catalog: set the 'catalog_type' property. The 'metadata_location' property points at a fixed table snapshot and cannot observe new commits.", + endpoint.config.mode + ); } // Register metrics here rather than at endpoint construction: the @@ -1002,6 +1116,375 @@ impl IcebergInputEndpointInner { ); } + /// Allocate a transaction for the next follow snapshot, or `None` when + /// `transaction_mode` is `none`. Each followed snapshot is one transaction. + fn allocate_follow_transaction(&self) -> Option> { + match self.config.transaction_mode { + IcebergTransactionMode::None => None, + IcebergTransactionMode::Snapshot => { + let index = self.transaction_index.fetch_add(1, Ordering::AcqRel); + Some(Some(format!("follow-{index}"))) + } + } + } + + /// Commit the current follow transaction (if any) and record a + /// checkpointable resume point after fully ingesting follow snapshot + /// `snapshot_id`. A restart resumes by following the table after it. + fn push_follow_boundary(&self, snapshot_id: i64) { + self.queue.push_entry( + InputQueueEntry::new_with_aux( + Utc::now(), + QueueEntry::ResumeInfo(Some(IcebergResumeInfo::follow(snapshot_id))), + ) + .with_commit_transaction(true), + Vec::new(), + ); + } + + /// Iceberg field ids for `used_columns`, in the same order, for projecting + /// the Arrow reader. Columns absent from `schema` are skipped. + fn project_field_ids(schema: &IcebergSchemaRef, used_columns: &[String]) -> Vec { + let name_to_id: std::collections::HashMap<&str, i32> = schema + .as_struct() + .fields() + .iter() + .map(|field| (field.name.as_str(), field.id)) + .collect(); + used_columns + .iter() + .filter_map(|name| name_to_id.get(name.as_str()).copied()) + .collect() + } + + /// Compute the data-file changelog of `snapshot` relative to its parent by + /// diffing manifests. Reads only the manifests this snapshot added, so the + /// cost is proportional to the change, not the table size. + /// + /// Rejects merge-on-read delete files (position or equality deletes) with a + /// clear error: v1 follow mode handles copy-on-write changes only. + async fn snapshot_delta( + &self, + table: &IcebergTable, + snapshot: &SnapshotRef, + ) -> Result { + let file_io = table.file_io(); + let metadata = table.metadata(); + let snapshot_id = snapshot.snapshot_id(); + + let manifest_list = snapshot + .load_manifest_list(file_io, metadata) + .await + .map_err(|e| anyhow!("error reading manifest list of snapshot {snapshot_id}: {e}"))?; + + let mut added = Vec::new(); + let mut removed = Vec::new(); + + for manifest_file in manifest_list.entries() { + // Only manifests this snapshot wrote carry its ADDED/DELETED + // entries; the rest are inherited unchanged. Skipping them keeps the + // diff proportional to the change. + if manifest_file.added_snapshot_id != snapshot_id { + continue; + } + + let manifest = manifest_file.load_manifest(file_io).await.map_err(|e| { + anyhow!( + "error reading manifest '{}': {e}", + manifest_file.manifest_path + ) + })?; + + for entry in manifest.entries() { + // Skip entries carried into this manifest from earlier snapshots + // (status EXISTING, attributed to a different snapshot id). + if entry.snapshot_id() != Some(snapshot_id) { + continue; + } + + if entry.content_type() != DataContentType::Data { + bail!( + "snapshot {snapshot_id} of the Iceberg table adds a merge-on-read delete file ('{}'); follow mode does not yet support merge-on-read deletes (position or equality delete files). Configure the writer to use copy-on-write, or track merge-on-read support in issue #6165.", + entry.file_path() + ); + } + + let changed = ChangedFile { + data_file: entry.data_file().clone(), + partition_spec_id: manifest_file.partition_spec_id, + }; + match entry.status() { + ManifestStatus::Added => added.push(changed), + ManifestStatus::Deleted => removed.push(changed), + ManifestStatus::Existing => {} + } + } + } + + Ok(SnapshotDelta { added, removed }) + } + + /// Build an Arrow record-batch stream over `files`, projecting `field_ids`. + /// + /// Uses the iceberg core reader so field-id projection, name mapping, and + /// identity-partition constants are handled correctly. `deletes` is empty: + /// merge-on-read is rejected earlier in [`snapshot_delta`]. + fn read_changed_files_stream( + &self, + table: &IcebergTable, + schema: &IcebergSchemaRef, + field_ids: &[i32], + name_mapping: &Option>, + files: &[ChangedFile], + ) -> Result>, AnyError> { + let metadata = table.metadata(); + let tasks: Vec> = files + .iter() + .map(|file| { + let data_file = &file.data_file; + Ok(FileScanTask { + file_size_in_bytes: data_file.file_size_in_bytes(), + start: 0, + length: data_file.file_size_in_bytes(), + record_count: Some(data_file.record_count()), + data_file_path: data_file.file_path().to_string(), + data_file_format: data_file.file_format(), + schema: schema.clone(), + project_field_ids: field_ids.to_vec(), + predicate: None, + deletes: vec![], + // Both `partition` and `partition_spec` must be set for the + // reader to materialize identity-partition column constants. + partition: Some(data_file.partition().clone()), + partition_spec: metadata + .partition_spec_by_id(file.partition_spec_id) + .cloned(), + name_mapping: name_mapping.clone(), + case_sensitive: false, + }) + }) + .collect(); + + let task_stream: FileScanTaskStream = stream::iter(tasks).boxed(); + + let runtime = Runtime::try_current() + .map_err(|e| anyhow!("no tokio runtime available for the Iceberg reader: {e}"))?; + let reader = ArrowReaderBuilder::new(table.file_io().clone(), runtime) + .with_data_file_concurrency_limit(self.config.num_parsers as usize) + .build(); + + let batch_stream = reader + .read(task_stream) + .map_err(|e| anyhow!("error starting Iceberg data-file read: {e}"))? + .stream() + .map(|batch| batch.map_err(|e| format!("error reading Iceberg data file: {e}"))) + .boxed(); + + Ok(batch_stream) + } + + /// Read `files` and push their rows to the circuit with `polarity`, wrapped + /// in `transaction`. Retries the whole read on transient failure, mirroring + /// [`execute_df`]. + #[allow(clippy::too_many_arguments)] + async fn push_changed_files( + &self, + table: &IcebergTable, + schema: &IcebergSchemaRef, + field_ids: &[i32], + name_mapping: &Option>, + files: &[ChangedFile], + polarity: bool, + transaction: Option>, + descr: &str, + input_stream: &mut dyn ArrowStream, + receiver: &mut Receiver, + ) -> Result { + if files.is_empty() { + return Ok(0); + } + + let max_retries = self.config.max_retries(); + let mut retry_count = 0; + loop { + let result = + match self.read_changed_files_stream(table, schema, field_ids, name_mapping, files) + { + Ok(stream) => self + .drain_batch_stream( + stream, + polarity, + transaction.clone(), + input_stream, + receiver, + ) + .await + .map_err(|e| anyhow!(e)), + Err(e) => Err(e), + }; + + match result { + Ok(total) => { + self.metrics + .follow_records_total + .fetch_add(total as u64, Ordering::Relaxed); + self.consumer + .update_connector_health(ConnectorHealth::healthy()); + return Ok(total); + } + Err(e) => { + // Commit any open transaction and mark a checkpointable + // boundary between retries; rows already queued are not + // rolled back, so a retry may re-emit them (at-least-once). + self.queue.push_entry( + InputQueueEntry::new_with_aux(Utc::now(), QueueEntry::Rollback) + .with_commit_transaction(true), + Vec::new(), + ); + + retry_count += 1; + if retry_count > max_retries { + let message = + format!("error reading {descr} after {retry_count} attempt(s): {e}"); + self.consumer + .update_connector_health(ConnectorHealth::unhealthy(&message)); + return Err(anyhow!(message)); + } + let backoff_delay = calculate_backoff_delay(retry_count - 1); + warn!( + "iceberg {}: error reading {descr}: {e}; retrying in {backoff_delay:?} (attempt {retry_count})", + &self.endpoint_name + ); + sleep(backoff_delay).await; + } + } + } + } + + /// Follow the table: poll for new snapshots after `start_snapshot_id` and + /// ingest each one's changes, until the worker task is canceled. + async fn follow_loop( + &self, + table: Arc, + used_columns: &[String], + start_snapshot_id: Option, + input_stream: &mut dyn ArrowStream, + receiver: &mut Receiver, + ) { + self.metrics.set_phase(IcebergPhase::Follow); + if let Err(e) = self + .follow_loop_inner( + table, + used_columns, + start_snapshot_id, + input_stream, + receiver, + ) + .await + { + self.consumer.error(true, e, Some("iceberg-follow")); + } + } + + async fn follow_loop_inner( + &self, + table: Arc, + used_columns: &[String], + start_snapshot_id: Option, + input_stream: &mut dyn ArrowStream, + receiver: &mut Receiver, + ) -> Result<(), AnyError> { + // Projection and name mapping are fixed by the table schema; compute + // them once. Schema evolution across followed snapshots is a v2 concern. + let schema = table.metadata().current_schema().clone(); + let field_ids = Self::project_field_ids(&schema, used_columns); + let name_mapping = table_name_mapping(&table)?; + + let mut last_snapshot_id = start_snapshot_id; + + loop { + wait_running(receiver).await; + + // Refresh the table to observe snapshots committed since the last + // poll. Retries transient catalog throttling. + let table = self + .open_table_with_retries() + .await + .map_err(|e| anyhow!("error refreshing the Iceberg table: {e}"))?; + + let new_snapshots = snapshots_after(table.metadata(), last_snapshot_id)?; + + if new_snapshots.is_empty() { + sleep(FOLLOW_POLL_INTERVAL).await; + continue; + } + + for snapshot in new_snapshots { + wait_running(receiver).await; + let snapshot_id = snapshot.snapshot_id(); + + // Compaction and other metadata-only rewrites carry no logical + // change; advance past them without reading. + if is_noop_operation(&snapshot) { + self.push_follow_boundary(snapshot_id); + last_snapshot_id = Some(snapshot_id); + continue; + } + + let delta = self.snapshot_delta(&table, &snapshot).await?; + + if !delta.is_empty() { + let transaction = self.allocate_follow_transaction(); + + // Inserts before deletes: a followed snapshot's added and + // removed files are disjoint, so ordering only affects + // intermediate state, not the final Z-set. + self.push_changed_files( + &table, + &schema, + &field_ids, + &name_mapping, + &delta.added, + true, + transaction.clone(), + &format!("snapshot {snapshot_id} inserts"), + input_stream, + receiver, + ) + .await?; + self.push_changed_files( + &table, + &schema, + &field_ids, + &name_mapping, + &delta.removed, + false, + transaction, + &format!("snapshot {snapshot_id} deletes"), + input_stream, + receiver, + ) + .await?; + } + + // Commit the snapshot's transaction and record the resume point. + self.push_follow_boundary(snapshot_id); + last_snapshot_id = Some(snapshot_id); + + self.metrics + .follow_snapshots_total + .fetch_add(1, Ordering::Relaxed); + self.metrics + .set_last_ingested_sequence_number(snapshot.sequence_number()); + info!( + "iceberg {}: ingested follow snapshot {snapshot_id} (sequence number {})", + &self.endpoint_name, + snapshot.sequence_number() + ); + } + } + } + async fn worker_task_inner( self: Arc, mut input_stream: Box, @@ -1092,25 +1575,43 @@ impl IcebergInputEndpointInner { ); } - // Terminal checkpoint boundary: the snapshot is fully ingested. - // Committing any in-progress transaction and recording the - // end-of-input state means a checkpoint taken after completion - // resumes straight into the eoi state, never re-reading the snapshot. - self.queue.push_entry( - InputQueueEntry::new_with_aux( - Utc::now(), - QueueEntry::ResumeInfo(Some(IcebergResumeInfo::eoi(snapshot_id))), - ) - .with_commit_transaction(true), - Vec::new(), - ); + if !self.config.follow() { + // Snapshot fully ingested. Commit any open transaction and + // record end-of-input, so a checkpoint here resumes into the eoi + // state without re-reading the snapshot. + self.queue.push_entry( + InputQueueEntry::new_with_aux( + Utc::now(), + QueueEntry::ResumeInfo(Some(IcebergResumeInfo::eoi(snapshot_id))), + ) + .with_commit_transaction(true), + Vec::new(), + ); + } else if let Some(snapshot_id) = snapshot_id { + // snapshot_and_follow: pin a follow resume point at the ingested + // snapshot, so a restart follows from it instead of re-reading. + self.push_follow_boundary(snapshot_id); + } } - // Snapshot-only connector: nothing follows the snapshot, so the - // connector is done once the snapshot has been read. - self.metrics.set_phase(IcebergPhase::Completed); - - self.consumer.eoi(); + if self.config.follow() { + // Follow until the worker task is canceled. `snapshot_id` is the + // start: the snapshot just ingested (snapshot_and_follow) or the + // resolved starting snapshot (follow-only). + self.follow_loop( + table.clone(), + &used_columns, + snapshot_id, + input_stream.as_mut(), + &mut receiver, + ) + .await; + } else { + // Snapshot-only connector: nothing follows the snapshot, so the + // connector is done once the snapshot has been read. + self.metrics.set_phase(IcebergPhase::Completed); + self.consumer.eoi(); + } } /// Open the table, retrying transient catalog throttling with backoff. @@ -1490,9 +1991,9 @@ impl IcebergInputEndpointInner { snapshot_id: Option, schema: &Relation, ) -> Result, ControllerError> { - if !self.config.snapshot() { - return Ok(Vec::new()); - } + // Compute `used_columns` for both modes: follow reads no initial + // snapshot but still projects changed files by it. The datafusion table + // and timestamp validation below run only when a snapshot is read. // Validate the filter before `config_referenced_columns` extracts // column names from it, so an invalid filter fails with a parse error @@ -1526,26 +2027,30 @@ impl IcebergInputEndpointInner { )); } - self.datafusion - .register_table("snapshot", Arc::new(provider)) - .map_err(|e| { - ControllerError::input_transport_error( + // Follow-only mode skips this: it reads changed files through the Arrow + // reader, not the datafusion `snapshot` table. + if self.config.snapshot() { + self.datafusion + .register_table("snapshot", Arc::new(provider)) + .map_err(|e| { + ControllerError::input_transport_error( + &self.endpoint_name, + true, + anyhow!("failed to register table snapshot with datafusion: {e}"), + ) + })?; + + if let Some(timestamp_column) = &self.config.timestamp_column { + validate_timestamp_column( &self.endpoint_name, - true, - anyhow!("failed to register table snapshot with datafusion: {e}"), + timestamp_column, + &self.datafusion, + schema, + "see Iceberg connector documentation for more details: https://docs.feldera.com/connectors/sources/iceberg" ) - })?; - - if let Some(timestamp_column) = &self.config.timestamp_column { - validate_timestamp_column( - &self.endpoint_name, - timestamp_column, - &self.datafusion, - schema, - "see Iceberg connector documentation for more details: https://docs.feldera.com/connectors/sources/iceberg" - ) - .await?; - }; + .await?; + }; + } Ok(used_columns) } @@ -1746,15 +2251,34 @@ impl IcebergInputEndpointInner { .fetch_add(1, Ordering::Relaxed); } - let mut stream = dataframe + let stream = dataframe .execute_stream() .await - .map_err(|e| format!("{e:?}"))?; + .map_err(|e| format!("{e:?}"))? + .map(|batch| batch.map_err(|e| format!("{e:?}"))) + .boxed(); // The dataframe compiled and started streaming: the connector is healthy. self.consumer .update_connector_health(ConnectorHealth::healthy()); + self.drain_batch_stream(stream, polarity, transaction, input_stream, receiver) + .await + } + + /// Parse a record-batch stream on a pool of `num_parsers` tasks and push the + /// resulting buffers to the input queue in enqueue order. + /// + /// Shared by the snapshot read path ([`execute_df_inner`]) and the follow + /// read path ([`push_changed_files`]). Blocks while the pipeline is paused. + async fn drain_batch_stream( + &self, + mut stream: stream::BoxStream<'static, Result>, + polarity: bool, + transaction: Option>, + input_stream: &mut dyn ArrowStream, + receiver: &mut Receiver, + ) -> Result { let mut num_batches = 0; let mut total_records = 0usize; @@ -1817,8 +2341,7 @@ impl IcebergInputEndpointInner { while let Some(batch) = stream.next().await { wait_running(receiver).await; - let batch = - batch.map_err(|e| format!("error retrieving batch {num_batches}: {e:?}"))?; + let batch = batch.map_err(|e| format!("error retrieving batch {num_batches}: {e}"))?; num_batches += 1; total_records += batch.num_rows(); job_queue.push_job((batch, timestamp)).await; diff --git a/crates/iceberg/src/test/README.md b/crates/iceberg/src/test/README.md index 6de4cef7142..2923fba169f 100644 --- a/crates/iceberg/src/test/README.md +++ b/crates/iceberg/src/test/README.md @@ -63,6 +63,39 @@ table (talk to leonid@feldera.com), for example via an SSO profile or exported a `export AWS_ACCESS_KEY_ID=... AWS_SECRET_ACCESS_KEY=... AWS_SESSION_TOKEN=...` * Run the following command in the `adapters` crate: `cargo test --features="iceberg-tests-s3tables" iceberg_s3tables_input_test` +## Follow-mode tests + +Follow mode needs a catalog both the writer (`pyiceberg`) and the connector +(`iceberg-rust`) can reach, so these tests use a REST catalog over an S3 store. +Unlike the snapshot tests, the writer commits several snapshots (the `create` +and `append` operations of `follow_table.py`) that the connector must tail. + +The default connection targets a local setup; override any of the +`FELDERA_ICEBERG_*` variables to point elsewhere. + +* Start an S3 store. The repo's `deploy/docker-compose.yml` already runs MinIO on + `localhost:9000` (`minio` / `miniopasswd`), with a `test` bucket. +* Start a REST catalog backed by that store: + + ``` + docker run -d --name iceberg-rest-follow --network deploy_default -p 8181:8181 \ + -e AWS_ACCESS_KEY_ID=minio -e AWS_SECRET_ACCESS_KEY=miniopasswd -e AWS_REGION=us-east-1 \ + -e CATALOG_WAREHOUSE=s3://test/iceberg-follow \ + -e CATALOG_IO__IMPL=org.apache.iceberg.aws.s3.S3FileIO \ + -e CATALOG_S3_ENDPOINT=http://deploy-minio-1:9000 \ + -e CATALOG_S3_PATH__STYLE__ACCESS=true \ + tabulario/iceberg-rest:1.6.0 + ``` + +* Run the tests in the `adapters` crate: + `cargo test --features="iceberg-tests-follow" iceberg_rest_follow` + +Overridable settings (defaults in parentheses): `FELDERA_ICEBERG_REST_URI` +(`http://localhost:8181`), `FELDERA_ICEBERG_WAREHOUSE` (`s3://test/iceberg-follow`), +`FELDERA_ICEBERG_S3_ENDPOINT` (`http://localhost:9000`), `FELDERA_ICEBERG_S3_KEY` +(`minio`), `FELDERA_ICEBERG_S3_SECRET` (`miniopasswd`), `FELDERA_ICEBERG_S3_REGION` +(`us-east-1`), `FELDERA_ICEBERG_S3_PATH_STYLE` (`true`). + # Running tests in CI Currently only Glue and FS-based tests run in CI. diff --git a/crates/iceberg/src/test/follow_table.py b/crates/iceberg/src/test/follow_table.py new file mode 100644 index 00000000000..7294916de7f --- /dev/null +++ b/crates/iceberg/src/test/follow_table.py @@ -0,0 +1,143 @@ +# Create and incrementally append to an Iceberg table through a REST catalog, +# for the connector's follow-mode tests. Each `append` produces a new snapshot +# the connector must pick up. +# +# The table schema matches `IcebergTestStruct` in the Rust tests, so the same +# `data()` generator and `file_to_zset` assertions apply. +# +# Connection settings come from the environment (defaults target the local +# docker setup in crates/iceberg/src/test/README.md): +# FELDERA_ICEBERG_REST_URI (default http://localhost:8181) +# FELDERA_ICEBERG_S3_ENDPOINT (default http://localhost:9000) +# FELDERA_ICEBERG_S3_KEY (default minio) +# FELDERA_ICEBERG_S3_SECRET (default miniopasswd) +# FELDERA_ICEBERG_S3_REGION (default us-east-1) + +import argparse +import os +from decimal import Decimal + +import pandas as pd +import pyarrow as pa +from pyiceberg.catalog.rest import RestCatalog +from pyiceberg.schema import Schema +from pyiceberg.partitioning import PartitionSpec, PartitionField +from pyiceberg.transforms import DayTransform +from pyiceberg.types import ( + BooleanType, + BinaryType, + DateType, + DoubleType, + DecimalType, + FloatType, + FixedType, + IntegerType, + LongType, + NestedField, + StringType, + TimeType, + TimestampType, + TimestamptzType, +) + +# Iceberg schema (matches `IcebergTestStruct`). +SCHEMA = Schema( + NestedField(1, "b", BooleanType(), required=True), + NestedField(2, "i", IntegerType(), required=True), + NestedField(3, "l", LongType(), required=True), + NestedField(4, "r", FloatType(), required=True), + NestedField(5, "d", DoubleType(), required=True), + NestedField(6, "dec", DecimalType(10, 3), required=True), + NestedField(7, "dt", DateType(), required=True), + NestedField(8, "tm", TimeType(), required=True), + NestedField(9, "ts", TimestampType(), required=True), + NestedField(10, "s", StringType(), required=True), + NestedField(11, "fixed", FixedType(5), required=True), + NestedField(12, "varbin", BinaryType(), required=True), + NestedField(13, "tstz", TimestamptzType(), required=True), +) + +ARROW_SCHEMA = pa.schema( + [ + pa.field("b", pa.bool_(), nullable=False), + pa.field("i", pa.int32(), nullable=False), + pa.field("l", pa.int64(), nullable=False), + pa.field("r", pa.float32(), nullable=False), + pa.field("d", pa.float64(), nullable=False), + pa.field("dec", pa.decimal128(10, 3), nullable=False), + pa.field("dt", pa.date32(), nullable=False), + pa.field("tm", pa.time64("us"), nullable=False), + pa.field("ts", pa.timestamp("us"), nullable=False), + pa.field("s", pa.string(), nullable=False), + pa.field("fixed", pa.binary(5), nullable=False), + pa.field("varbin", pa.binary(), nullable=False), + pa.field("tstz", pa.timestamp("us", tz="UTC"), nullable=False), + ] +) + +PARTITION_SPEC = PartitionSpec( + PartitionField(source_id=9, field_id=1000, transform=DayTransform(), name="date") +) + + +def catalog(): + return RestCatalog( + "follow", + **{ + "uri": os.getenv("FELDERA_ICEBERG_REST_URI", "http://localhost:8181"), + "s3.endpoint": os.getenv( + "FELDERA_ICEBERG_S3_ENDPOINT", "http://localhost:9000" + ), + "s3.access-key-id": os.getenv("FELDERA_ICEBERG_S3_KEY", "minio"), + "s3.secret-access-key": os.getenv("FELDERA_ICEBERG_S3_SECRET", "miniopasswd"), + "s3.region": os.getenv("FELDERA_ICEBERG_S3_REGION", "us-east-1"), + }, + ) + + +def arrow_chunk(json_file): + """Load an ndjson chunk (the format `data_to_ndjson` writes) into an Arrow + table matching the Iceberg schema.""" + df = pd.read_json(json_file, lines=True) + df["tm"] = pd.to_datetime(df["tm"]).dt.time + df["ts"] = pd.to_datetime(df["ts"]).astype("datetime64[us]") + df["tstz"] = pd.to_datetime(df["tstz"], utc=True).astype("datetime64[us, UTC]") + df["dt"] = pd.to_datetime(df["dt"]).dt.date + df["dec"] = df["dec"].apply(lambda x: Decimal(f"{x:.3f}")) + df["fixed"] = df["fixed"].apply(lambda x: bytes(x)) + df["varbin"] = df["varbin"].apply(lambda x: bytes(x)) + return pa.Table.from_pandas(df, schema=ARROW_SCHEMA) + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--op", choices=["create", "append"], required=True) + parser.add_argument("--table", required=True, help="table as 'namespace.name'") + parser.add_argument("--json-file", required=True, help="ndjson chunk to append") + args = parser.parse_args() + + cat = catalog() + namespace = args.table.split(".")[0] + + if args.op == "create": + try: + cat.create_namespace(namespace) + except Exception: + pass + try: + cat.drop_table(args.table) + except Exception: + pass + table = cat.create_table( + args.table, SCHEMA, partition_spec=PARTITION_SPEC + ) + else: + table = cat.load_table(args.table) + + table.append(arrow_chunk(args.json_file)) + # Print the current snapshot id so the caller can log progress. + print(table.metadata.current_snapshot_id) + + +if __name__ == "__main__": + main() diff --git a/docs.feldera.com/docs/connectors/sources/iceberg.md b/docs.feldera.com/docs/connectors/sources/iceberg.md index 230c2aa63c6..24df973b228 100644 --- a/docs.feldera.com/docs/connectors/sources/iceberg.md +++ b/docs.feldera.com/docs/connectors/sources/iceberg.md @@ -30,7 +30,7 @@ The Iceberg input connector supports [fault tolerance](/pipelines/fault-toleranc | Property | Type | Description | |-----------------------------|--------|---------------| -| `mode`* | enum | Table read mode. Currently, the only supported mode is `snapshot`, in which the connector reads a snapshot of the table and stops.| +| `mode`* | enum |

Table read mode. Supported values:

  • `snapshot` - read a snapshot of the table and stop.
  • `follow` - after the starting snapshot, continuously ingest new and deleted rows as the table commits new snapshots.
  • `snapshot_and_follow` - read a snapshot of the table, then switch to `follow` mode.

`follow` and `snapshot_and_follow` require an Iceberg catalog (set `catalog_type`); they cannot be used with `metadata_location`, which points at a fixed snapshot. See [Follow mode](#follow-mode) below.

| | `transaction_mode` | enum | Determines how the connector breaks up its input into transactions. Supported values are `none` (default) and `snapshot`. See [below](#transactions) for details. | | `timestamp_column` | string | Table column that serves as an event timestamp. When this option is specified, table rows are ingested in the timestamp order, respecting the [`LATENESS`](/sql/streaming#lateness-expressions) property of the column: each ingested row has a timestamp no more than `LATENESS` time units earlier than the most recent timestamp of any previously ingested row. See details [below](#ingesting-time-series-data-from-iceberg). | | `snapshot_filter` | string |

Optional row filter. When specified, only rows that satisfy the filter condition are included in the snapshot. The condition must be a valid SQL Boolean expression that can be used in the `where` clause of the `select * from snapshot where ..` query.

This option can be used to specify the range of event times to include in the snapshot, e.g.: `ts BETWEEN TIMESTAMP '2005-01-01 00:00:00' AND TIMESTAMP '2010-12-31 23:59:59'`.

@@ -169,6 +169,33 @@ declaration. Other columns of the Iceberg table are never read. In addition, when the table declaration sets the [`skip_unused_columns` property](/sql/grammar#ignoring-unused-columns), the connector skips declared columns that no view uses, provided they are nullable or have default values. +## Follow mode + +In `follow` and `snapshot_and_follow` modes the connector continuously ingests +changes committed to the table after its starting snapshot. It polls the catalog +for new snapshots and, for each one, diffs the snapshot against its predecessor +to find the data files it added and removed, ingesting added rows as inserts and +removed rows as deletes. Only the manifests a snapshot added are read, so the +cost of each step is proportional to the size of the change, not to the size of +the table. + +The starting snapshot is chosen the same way as in `snapshot` mode: by +`snapshot_id`, by `datetime`, or, when neither is set, the latest snapshot at the +time the connector starts. In `follow` mode the connector ingests only changes +committed after the starting snapshot; in `snapshot_and_follow` mode it first +reads the starting snapshot in full, then follows. + +Requirements and limitations: + +* **A catalog is required.** Set `catalog_type`; follow mode cannot be used with + `metadata_location`, which points at a fixed snapshot and cannot observe new + commits. +* **Copy-on-write only.** Follow mode reads copy-on-write changes. If a followed + snapshot adds a merge-on-read delete file (position or equality deletes), the + connector stops with an error. Configure the writer to use copy-on-write. +* **Compaction is skipped.** Snapshots that only rewrite files without changing + table contents (`operation = replace`) are recognized and skipped. + ## Transactions The Iceberg connector can be configured to automatically initiate [transactions](/pipelines/transactions) From b52f6ec6bb3198c13d0e4191f930f1e6e584128b Mon Sep 17 00:00:00 2001 From: feldera-bot Date: Wed, 29 Jul 2026 16:04:25 +0000 Subject: [PATCH 2/5] [ci] apply automatic fixes Signed-off-by: feldera-bot --- crates/iceberg/src/test/follow_table.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/crates/iceberg/src/test/follow_table.py b/crates/iceberg/src/test/follow_table.py index 7294916de7f..0415ce6cc4e 100644 --- a/crates/iceberg/src/test/follow_table.py +++ b/crates/iceberg/src/test/follow_table.py @@ -89,7 +89,9 @@ def catalog(): "FELDERA_ICEBERG_S3_ENDPOINT", "http://localhost:9000" ), "s3.access-key-id": os.getenv("FELDERA_ICEBERG_S3_KEY", "minio"), - "s3.secret-access-key": os.getenv("FELDERA_ICEBERG_S3_SECRET", "miniopasswd"), + "s3.secret-access-key": os.getenv( + "FELDERA_ICEBERG_S3_SECRET", "miniopasswd" + ), "s3.region": os.getenv("FELDERA_ICEBERG_S3_REGION", "us-east-1"), }, ) @@ -128,9 +130,7 @@ def main(): cat.drop_table(args.table) except Exception: pass - table = cat.create_table( - args.table, SCHEMA, partition_spec=PARTITION_SPEC - ) + table = cat.create_table(args.table, SCHEMA, partition_spec=PARTITION_SPEC) else: table = cat.load_table(args.table) From eb11c13ee5fcc685a7c19f680d7411aa1fa7a999 Mon Sep 17 00:00:00 2001 From: Swanand Mulay <73115739+swanandx@users.noreply.github.com> Date: Thu, 30 Jul 2026 06:43:14 +0530 Subject: [PATCH 3/5] [connectors] Add Iceberg follow transaction modes and end bound - catchup: group all commits seen in one poll into one transaction - always: one transaction per followed snapshot - end_snapshot_id: stop after ingesting the named snapshot Also recompute projection/name mapping per poll for schema evolution. Signed-off-by: Swanand Mulay <73115739+swanandx@users.noreply.github.com> --- crates/adapters/src/test/iceberg.rs | 216 ++++++++++- crates/feldera-types/src/transport/iceberg.rs | 38 +- crates/iceberg/src/input.rs | 363 ++++++++++++++---- .../docs/connectors/sources/iceberg.md | 29 +- 4 files changed, 542 insertions(+), 104 deletions(-) diff --git a/crates/adapters/src/test/iceberg.rs b/crates/adapters/src/test/iceberg.rs index 011ebc5cccb..5936584eac5 100644 --- a/crates/adapters/src/test/iceberg.rs +++ b/crates/adapters/src/test/iceberg.rs @@ -801,14 +801,21 @@ fn follow_table_op(op: &str, table: &str, chunk: &[IcebergTestStruct]) -> i64 { .unwrap_or_else(|_| panic!("'{script} --op={op}' printed unexpected output: {last:?}")) } +/// Read a single custom metric of the Iceberg connector, or `0.0` if absent. +#[cfg(feature = "iceberg-tests-follow")] +fn iceberg_metric(pipeline: &Controller, name: &str) -> f64 { + iceberg_connector_metrics(pipeline) + .get(name) + .copied() + .unwrap_or(0.0) +} + /// Sum of the records the connector has ingested so far (snapshot phase plus /// follow phase), read from its custom metrics. #[cfg(feature = "iceberg-tests-follow")] fn ingested_records(pipeline: &Controller) -> u64 { - let metrics = iceberg_connector_metrics(pipeline); - let get = |name: &str| metrics.get(name).copied().unwrap_or(0.0) as u64; - get("input_connector_iceberg_snapshot_records_total") - + get("input_connector_iceberg_follow_records_total") + (iceberg_metric(pipeline, "input_connector_iceberg_snapshot_records_total") + + iceberg_metric(pipeline, "input_connector_iceberg_follow_records_total")) as u64 } /// Run `body` against a running follow-mode pipeline, then stop it. The output @@ -936,6 +943,207 @@ fn iceberg_rest_follow_start_from_snapshot_id() { }); } +/// `follow`: `end_snapshot_id` stops the connector after fully ingesting the +/// named snapshot. Three snapshots (A, B, C) exist before startup; following +/// from A with `end_snapshot_id = B` must ingest only B's rows, reach the +/// completed phase, and never ingest C, even though C is in the same catch-up +/// batch as B. +#[test] +#[cfg(feature = "iceberg-tests-follow")] +fn iceberg_rest_follow_end_snapshot_id() { + use dbsp::trace::BatchReader; + + let all = data(15); + let ns = env_or("FELDERA_ICEBERG_NAMESPACE", "follow_ns"); + let table = &format!("{ns}.end_snapshot"); + + let snapshot_a = follow_table_op("create", table, &all[..5]); // snapshot A + let snapshot_b = follow_table_op("append", table, &all[5..10]); // snapshot B + follow_table_op("append", table, &all[10..]); // snapshot C + + let mut config = rest_follow_config_with(table, "follow", "snapshot_id", json!(snapshot_a)); + config + .as_object_mut() + .unwrap() + .insert("end_snapshot_id".to_string(), json!(snapshot_b)); + + with_follow_pipeline_cfg(config, |pipeline, out_path| { + // Phase 2 = completed: only set once the end snapshot is reached. + wait( + || iceberg_metric(pipeline, "input_connector_iceberg_phase") == 2.0, + 120_000, + ) + .expect("timed out reaching the end snapshot"); + wait(|| output_record_count(out_path) >= 5, 60_000).expect("timed out writing output"); + + // Stopped at B: C's rows are never ingested and exactly one follow + // snapshot was processed. + assert_eq!(ingested_records(pipeline), 5); + assert_eq!( + iceberg_metric(pipeline, "input_connector_iceberg_follow_snapshots_total"), + 1.0 + ); + let zset = output_zset(out_path); + assert_eq!(zset.len(), 5); + assert_eq!(zset, expected_zset(&all[5..10])); + }); +} + +/// `transaction_mode = always`: each followed snapshot forms its own Feldera +/// transaction. Following from A over two pre-committed snapshots (B, C) starts +/// two follow transactions. +#[test] +#[cfg(feature = "iceberg-tests-follow")] +fn iceberg_rest_follow_transaction_always() { + use dbsp::trace::BatchReader; + + let all = data(15); + let ns = env_or("FELDERA_ICEBERG_NAMESPACE", "follow_ns"); + let table = &format!("{ns}.txn_always"); + + let snapshot_a = follow_table_op("create", table, &all[..5]); // A + follow_table_op("append", table, &all[5..10]); // B + follow_table_op("append", table, &all[10..]); // C + + let mut config = rest_follow_config_with(table, "follow", "snapshot_id", json!(snapshot_a)); + config + .as_object_mut() + .unwrap() + .insert("transaction_mode".to_string(), json!("always")); + + with_follow_pipeline_cfg(config, |pipeline, out_path| { + wait(|| ingested_records(pipeline) >= 10, 120_000).expect("timed out following B and C"); + wait(|| output_record_count(out_path) >= 10, 60_000).expect("timed out writing output"); + + // One transaction per followed snapshot: B and C. (Reverting the + // per-snapshot allocation drops this below 2.) + assert_eq!( + iceberg_metric( + pipeline, + "input_connector_iceberg_follow_transaction_starts" + ), + 2.0 + ); + let zset = output_zset(out_path); + assert_eq!(zset.len(), 10); + assert_eq!(zset, expected_zset(&all[5..])); + }); +} + +/// `transaction_mode = catchup`: all snapshots caught up in one poll form a +/// single Feldera transaction. Following from A over two pre-committed snapshots +/// (B, C), both seen in the first poll, starts exactly one follow transaction. +#[test] +#[cfg(feature = "iceberg-tests-follow")] +fn iceberg_rest_follow_transaction_catchup() { + use dbsp::trace::BatchReader; + + let all = data(15); + let ns = env_or("FELDERA_ICEBERG_NAMESPACE", "follow_ns"); + let table = &format!("{ns}.txn_catchup"); + + let snapshot_a = follow_table_op("create", table, &all[..5]); // A + follow_table_op("append", table, &all[5..10]); // B + follow_table_op("append", table, &all[10..]); // C + + let mut config = rest_follow_config_with(table, "follow", "snapshot_id", json!(snapshot_a)); + config + .as_object_mut() + .unwrap() + .insert("transaction_mode".to_string(), json!("catchup")); + + with_follow_pipeline_cfg(config, |pipeline, out_path| { + wait(|| ingested_records(pipeline) >= 10, 120_000).expect("timed out following B and C"); + wait(|| output_record_count(out_path) >= 10, 60_000).expect("timed out writing output"); + + // B and C are committed before startup, so the first catch-up poll sees + // both and groups them into one transaction. (In `always` this is 2; + // that difference is the point of the mode.) + assert_eq!( + iceberg_metric( + pipeline, + "input_connector_iceberg_follow_transaction_starts" + ), + 1.0 + ); + // The catchup window closed once the batch committed, so the target + // gauge is back to its unset sentinel. + assert_eq!( + iceberg_metric( + pipeline, + "input_connector_iceberg_catchup_target_sequence_number" + ), + -1.0 + ); + let zset = output_zset(out_path); + assert_eq!(zset.len(), 10); + assert_eq!(zset, expected_zset(&all[5..])); + }); +} + +/// `transaction_mode = catchup` with `end_snapshot_id` landing mid-batch: the +/// catchup window opens targeting the batch's last snapshot (C), but the end +/// bound B is reached first. The open catchup transaction (holding B's rows) +/// must be committed at the end bound, C must never be read, and the target +/// gauge must reset. This is the one interaction the plain `end_snapshot_id` +/// test (transaction_mode `none`) does not exercise. +#[test] +#[cfg(feature = "iceberg-tests-follow")] +fn iceberg_rest_follow_transaction_catchup_end_snapshot_id() { + use dbsp::trace::BatchReader; + + let all = data(15); + let ns = env_or("FELDERA_ICEBERG_NAMESPACE", "follow_ns"); + let table = &format!("{ns}.txn_catchup_end"); + + let snapshot_a = follow_table_op("create", table, &all[..5]); // A + let snapshot_b = follow_table_op("append", table, &all[5..10]); // B + follow_table_op("append", table, &all[10..]); // C + + let mut config = rest_follow_config_with(table, "follow", "snapshot_id", json!(snapshot_a)); + { + let object = config.as_object_mut().unwrap(); + object.insert("transaction_mode".to_string(), json!("catchup")); + object.insert("end_snapshot_id".to_string(), json!(snapshot_b)); + } + + with_follow_pipeline_cfg(config, |pipeline, out_path| { + // Phase 2 = completed: only set once the end snapshot is reached. + wait( + || iceberg_metric(pipeline, "input_connector_iceberg_phase") == 2.0, + 120_000, + ) + .expect("timed out reaching the end snapshot"); + wait(|| output_record_count(out_path) >= 5, 60_000).expect("timed out writing output"); + + // Only B ingested; C never read even though it shared the catch-up batch. + assert_eq!(ingested_records(pipeline), 5); + assert_eq!( + iceberg_metric(pipeline, "input_connector_iceberg_follow_snapshots_total"), + 1.0 + ); + // The end bound committed the open catchup transaction: exactly one + // follow transaction started, and the target gauge reset on close. + assert_eq!( + iceberg_metric( + pipeline, + "input_connector_iceberg_follow_transaction_starts" + ), + 1.0 + ); + assert_eq!( + iceberg_metric( + pipeline, + "input_connector_iceberg_catchup_target_sequence_number" + ), + -1.0 + ); + let zset = output_zset(out_path); + assert_eq!(zset.len(), 5); + assert_eq!(zset, expected_zset(&all[5..10])); + }); +} + /// Number of complete `insert_delete` records written to the output file so /// far (one JSON value per line). #[cfg(feature = "iceberg-tests-follow")] diff --git a/crates/feldera-types/src/transport/iceberg.rs b/crates/feldera-types/src/transport/iceberg.rs index 9234879f2b6..8b1e2385821 100644 --- a/crates/feldera-types/src/transport/iceberg.rs +++ b/crates/feldera-types/src/transport/iceberg.rs @@ -52,16 +52,24 @@ pub enum IcebergCatalogType { /// /// Determines how the connector breaks up its input into Feldera transactions. /// -/// * `none` - the connector does not break up its input into transactions. -/// * `snapshot` - ingest the initial snapshot of the table in one or several transactions. +/// * `none` - the connector does not group its input into transactions. +/// * `snapshot` - ingest the initial snapshot in one or more transactions (see below). Changes +/// ingested afterward, in the follow phase, are not grouped into transactions. +/// * `catchup` - ingest the initial snapshot like `snapshot`. In the follow phase, the connector +/// groups all table commits that are already available into a single transaction: while catching +/// up on a backlog it ingests many commits per transaction, and once caught up it ingests about +/// one commit per transaction. Most efficient for backfill and steady-state following. +/// * `always` - ingest the initial snapshot like `snapshot`. In the follow phase, each table commit +/// is ingested in its own transaction. /// /// # How the table snapshot is ingested using transactions /// -/// When `transaction_mode` is set to `snapshot`, the connector ingests the snapshot in one -/// or several transactions, depending on `timestamp_column`. If `timestamp_column` is not set, -/// the whole snapshot is ingested in a single Feldera transaction. If `timestamp_column` is set, -/// the connector ingests the snapshot in a series of timestamp ranges of width equal to the -/// `LATENESS` attribute of the column, each range in a separate transaction. +/// For the initial snapshot (`snapshot`, `catchup`, and `always` all behave the same), the +/// connector ingests the snapshot in one or several transactions, depending on `timestamp_column`. +/// If `timestamp_column` is not set, the whole snapshot is ingested in a single Feldera +/// transaction. If `timestamp_column` is set, the connector ingests the snapshot in a series of +/// timestamp ranges of width equal to the `LATENESS` attribute of the column, each range in a +/// separate transaction. #[derive(Debug, Clone, Eq, PartialEq, Deserialize, Serialize, ToSchema, Default)] pub enum IcebergTransactionMode { #[default] @@ -69,6 +77,10 @@ pub enum IcebergTransactionMode { None, #[serde(rename = "snapshot")] Snapshot, + #[serde(rename = "catchup")] + Catchup, + #[serde(rename = "always")] + Always, } /// AWS Glue catalog config. @@ -280,6 +292,18 @@ pub struct IcebergReaderConfig { /// is used. pub datetime: Option, + /// Optional final snapshot id. + /// + /// Valid only in `follow` and `snapshot_and_follow` modes. + /// + /// When set, the connector stops after fully ingesting the snapshot with + /// this id, signaling end-of-input. Unlike a Delta table version, an Iceberg + /// snapshot id is not ordered, so the bound is an exact match: the id must + /// name a snapshot committed after the starting snapshot and already present + /// in the table's current history. The connector rejects any other value at + /// startup, including a not-yet-committed id, rather than follow forever. + pub end_snapshot_id: Option, + /// Location of the table metadata JSON file. /// /// This propery is used to access an Iceberg table without a catalog. It is mutually diff --git a/crates/iceberg/src/input.rs b/crates/iceberg/src/input.rs index 91bac7db2db..358fefed611 100644 --- a/crates/iceberg/src/input.rs +++ b/crates/iceberg/src/input.rs @@ -207,9 +207,14 @@ struct IcebergMetrics { follow_snapshots_total: AtomicU64, /// Total records ingested in follow mode (inserts plus deletes). follow_records_total: AtomicU64, + /// Number of Feldera follow transactions started by this connector. + follow_transaction_starts: AtomicU64, + /// Sequence number the in-flight `catchup` transaction is catching up to; + /// [`SEQUENCE_METRIC_UNSET`] when no catchup window is open. + catchup_target_sequence_number: AtomicU64, } -/// Sentinel stored in the sequence-number gauge before a value is available. +/// Sentinel stored in the sequence-number gauges before a value is available. const SEQUENCE_METRIC_UNSET: u64 = u64::MAX; impl IcebergMetrics { @@ -222,6 +227,29 @@ impl IcebergMetrics { last_ingested_sequence_number: AtomicU64::new(SEQUENCE_METRIC_UNSET), follow_snapshots_total: AtomicU64::new(0), follow_records_total: AtomicU64::new(0), + follow_transaction_starts: AtomicU64::new(0), + catchup_target_sequence_number: AtomicU64::new(SEQUENCE_METRIC_UNSET), + } + } + + fn set_catchup_target_sequence_number(&self, sequence_number: i64) { + debug_assert!( + sequence_number >= 0, + "Iceberg sequence number must be non-negative" + ); + self.catchup_target_sequence_number + .store(sequence_number as u64, Ordering::Relaxed); + } + + fn clear_catchup_target_sequence_number(&self) { + self.catchup_target_sequence_number + .store(SEQUENCE_METRIC_UNSET, Ordering::Relaxed); + } + + fn catchup_target_sequence_number_metric(&self) -> f64 { + match self.catchup_target_sequence_number.load(Ordering::Relaxed) { + SEQUENCE_METRIC_UNSET => -1.0, + sequence_number => sequence_number as f64, } } @@ -291,6 +319,18 @@ impl ConnectorMetrics for IcebergMetrics { ValueType::Counter, self.follow_records_total.load(Ordering::Relaxed) as f64, ), + ( + "input_connector_iceberg_follow_transaction_starts", + "Number of Feldera follow transactions started by this connector.", + ValueType::Counter, + self.follow_transaction_starts.load(Ordering::Relaxed) as f64, + ), + ( + "input_connector_iceberg_catchup_target_sequence_number", + "Sequence number the in-flight catchup transaction is catching up to (-1 if no catchup window is open).", + ValueType::Gauge, + self.catchup_target_sequence_number_metric(), + ), ] } } @@ -576,6 +616,13 @@ impl IcebergInputReader { ); } + if endpoint.config.end_snapshot_id.is_some() && !endpoint.config.follow() { + bail!( + "the 'end_snapshot_id' property is only valid in 'follow' or 'snapshot_and_follow' mode, not '{}' mode", + endpoint.config.mode + ); + } + // Register metrics here rather than at endpoint construction: the // controller inserts this endpoint's status entry after constructing the // endpoint but before calling `open` (which builds this reader), and @@ -797,6 +844,19 @@ struct IcebergInputEndpointInner { /// Most recent resume point that is safe to check point at. Used to answer a /// checkpoint that stopped on a [`QueueEntry::Rollback`] boundary. last_checkpointable_status: Mutex, + + /// In-flight `catchup` transaction, shared so a mid-window `Rollback` in the + /// retry path can abandon it and continue under a fresh label. + catchup_state: Mutex, +} + +#[derive(Default)] +struct CatchupFollowState { + /// Snapshot id the window commits at: the last snapshot in the poll batch. + target_snapshot_id: Option, + /// Window transaction (queue's `start_transaction` type). `None` until the + /// window's first rows, or after a `Rollback` committed the previous one. + transaction: Option>, } impl IcebergInputEndpointInner { @@ -830,6 +890,7 @@ impl IcebergInputEndpointInner { metrics, last_resume_status: Mutex::new(Some(IcebergResumeInfo::initial())), last_checkpointable_status: Mutex::new(IcebergResumeInfo::initial()), + catchup_state: Mutex::new(CatchupFollowState::default()), } } @@ -838,10 +899,14 @@ impl IcebergInputEndpointInner { /// Returns `None` when `transaction_mode` is `none`, meaning the chunk is not /// wrapped in a Feldera transaction. Otherwise returns `Some(Some(label))`, /// where the label identifies the transaction in logs and metrics. + /// `snapshot`, `catchup`, and `always` all transact the initial snapshot the + /// same way; they differ only in the follow phase. fn allocate_snapshot_transaction(&self) -> Option> { match self.config.transaction_mode { IcebergTransactionMode::None => None, - IcebergTransactionMode::Snapshot => { + IcebergTransactionMode::Snapshot + | IcebergTransactionMode::Catchup + | IcebergTransactionMode::Always => { let index = self.transaction_index.fetch_add(1, Ordering::AcqRel); Some(Some(format!("snapshot-{index}"))) } @@ -1116,16 +1181,94 @@ impl IcebergInputEndpointInner { ); } - /// Allocate a transaction for the next follow snapshot, or `None` when - /// `transaction_mode` is `none`. Each followed snapshot is one transaction. - fn allocate_follow_transaction(&self) -> Option> { - match self.config.transaction_mode { - IcebergTransactionMode::None => None, - IcebergTransactionMode::Snapshot => { - let index = self.transaction_index.fetch_add(1, Ordering::AcqRel); - Some(Some(format!("follow-{index}"))) - } + /// Allocate and count a follow transaction, or `None` when the mode does not + /// transact the follow phase. Typed as the queue's `start_transaction` param. + fn new_follow_transaction_label(&self) -> Option> { + if matches!( + self.config.transaction_mode, + IcebergTransactionMode::Always | IcebergTransactionMode::Catchup + ) { + self.metrics + .follow_transaction_starts + .fetch_add(1, Ordering::Relaxed); + let index = self.transaction_index.fetch_add(1, Ordering::AcqRel); + Some(Some(format!("follow-{index}"))) + } else { + None + } + } + + /// Transaction for a followed snapshot's rows: `always` gets a fresh one per + /// snapshot, `catchup` shares the window's, `none`/`snapshot` get none. + fn follow_data_transaction(&self) -> Option> { + if self.config.transaction_mode == IcebergTransactionMode::Catchup { + self.catchup_transaction() + } else { + self.new_follow_transaction_label() + } + } + + /// Open a catchup window that commits at snapshot `target_snapshot_id`, the + /// last snapshot in the poll batch. Keyed on id, not sequence number: V1 + /// tables number every snapshot 0, so `>= target` would commit immediately. + fn begin_catchup_window(&self, target_snapshot_id: i64, target_sequence_number: i64) { + let mut state = self.catchup_state.lock().unwrap(); + state.target_snapshot_id = Some(target_snapshot_id); + state.transaction = None; + self.metrics + .set_catchup_target_sequence_number(target_sequence_number); + } + + /// Snapshot id the current catchup window commits at, if a window is open. + fn catchup_target_snapshot_id(&self) -> Option { + self.catchup_state.lock().unwrap().target_snapshot_id + } + + /// Transaction of the current catchup window, allocating (and counting) a + /// new one when the window has none yet or its previous transaction was + /// committed by a `Rollback`. + fn catchup_transaction(&self) -> Option> { + let mut state = self.catchup_state.lock().unwrap(); + if state.transaction.is_none() { + state.transaction = self.new_follow_transaction_label(); + } + state.transaction.clone() + } + + /// Drop the current catchup transaction label after a `Rollback` committed + /// it, so the window continues under a fresh label. + fn abandon_catchup_transaction(&self) { + self.catchup_state.lock().unwrap().transaction = None; + } + + /// Close the catchup window and clear the target metric. + fn reset_catchup_window(&self) { + *self.catchup_state.lock().unwrap() = CatchupFollowState::default(); + self.metrics.clear_catchup_target_sequence_number(); + } + + /// If `snapshot_id` is the configured `end_snapshot_id`, commit, record + /// end-of-input, and signal the controller. Returns whether the connector + /// has reached its end bound and should stop following. + fn finish_at_end_snapshot(&self, snapshot_id: i64) -> bool { + if self.config.end_snapshot_id != Some(snapshot_id) { + return false; } + self.queue.push_entry( + InputQueueEntry::new_with_aux( + Utc::now(), + QueueEntry::ResumeInfo(Some(IcebergResumeInfo::eoi(Some(snapshot_id)))), + ) + .with_commit_transaction(true), + Vec::new(), + ); + self.metrics.set_phase(IcebergPhase::Completed); + self.consumer.eoi(); + info!( + "iceberg {}: reached snapshot {snapshot_id} configured as 'end_snapshot_id'; stopping the connector", + &self.endpoint_name + ); + true } /// Commit the current follow transaction (if any) and record a @@ -1295,7 +1438,7 @@ impl IcebergInputEndpointInner { name_mapping: &Option>, files: &[ChangedFile], polarity: bool, - transaction: Option>, + mut transaction: Option>, descr: &str, input_stream: &mut dyn ArrowStream, receiver: &mut Receiver, @@ -1342,6 +1485,14 @@ impl IcebergInputEndpointInner { Vec::new(), ); + // The Rollback committed the open catchup transaction, so + // continue the window under a fresh label; the retry's rows + // are then not attributed to the committed transaction. + if self.config.transaction_mode == IcebergTransactionMode::Catchup { + self.abandon_catchup_transaction(); + transaction = self.follow_data_transaction(); + } + retry_count += 1; if retry_count > max_retries { let message = @@ -1365,7 +1516,6 @@ impl IcebergInputEndpointInner { /// ingest each one's changes, until the worker task is canceled. async fn follow_loop( &self, - table: Arc, used_columns: &[String], start_snapshot_id: Option, input_stream: &mut dyn ArrowStream, @@ -1373,13 +1523,7 @@ impl IcebergInputEndpointInner { ) { self.metrics.set_phase(IcebergPhase::Follow); if let Err(e) = self - .follow_loop_inner( - table, - used_columns, - start_snapshot_id, - input_stream, - receiver, - ) + .follow_loop_inner(used_columns, start_snapshot_id, input_stream, receiver) .await { self.consumer.error(true, e, Some("iceberg-follow")); @@ -1388,18 +1532,11 @@ impl IcebergInputEndpointInner { async fn follow_loop_inner( &self, - table: Arc, used_columns: &[String], start_snapshot_id: Option, input_stream: &mut dyn ArrowStream, receiver: &mut Receiver, ) -> Result<(), AnyError> { - // Projection and name mapping are fixed by the table schema; compute - // them once. Schema evolution across followed snapshots is a v2 concern. - let schema = table.metadata().current_schema().clone(); - let field_ids = Self::project_field_ids(&schema, used_columns); - let name_mapping = table_name_mapping(&table)?; - let mut last_snapshot_id = start_snapshot_id; loop { @@ -1419,68 +1556,97 @@ impl IcebergInputEndpointInner { continue; } + // Recompute projection and name mapping from the refreshed metadata + // so schema changes across snapshots are read with the current field + // ids (Iceberg resolves columns by field id, so old files still read + // correctly). The Feldera relation is fixed at pipeline creation, so + // a non-additive change still requires a restart to take effect. + let schema = table.metadata().current_schema().clone(); + let field_ids = Self::project_field_ids(&schema, used_columns); + let name_mapping = table_name_mapping(&table)?; + + // `catchup` ingests the whole poll batch in one transaction that + // commits when the loop reaches the batch's last snapshot. + let catchup = self.config.transaction_mode == IcebergTransactionMode::Catchup; + if catchup { + if let Some(target) = new_snapshots.last() { + self.begin_catchup_window(target.snapshot_id(), target.sequence_number()); + } + } + for snapshot in new_snapshots { wait_running(receiver).await; let snapshot_id = snapshot.snapshot_id(); // Compaction and other metadata-only rewrites carry no logical - // change; advance past them without reading. - if is_noop_operation(&snapshot) { - self.push_follow_boundary(snapshot_id); - last_snapshot_id = Some(snapshot_id); - continue; - } - - let delta = self.snapshot_delta(&table, &snapshot).await?; + // change; skip reading but still treat the snapshot as a + // boundary below. + if !is_noop_operation(&snapshot) { + let delta = self.snapshot_delta(&table, &snapshot).await?; + + if !delta.is_empty() { + let transaction = self.follow_data_transaction(); + + // Inserts before deletes: a followed snapshot's added and + // removed files are disjoint, so ordering only affects + // intermediate state, not the final Z-set. + self.push_changed_files( + &table, + &schema, + &field_ids, + &name_mapping, + &delta.added, + true, + transaction.clone(), + &format!("snapshot {snapshot_id} inserts"), + input_stream, + receiver, + ) + .await?; + self.push_changed_files( + &table, + &schema, + &field_ids, + &name_mapping, + &delta.removed, + false, + transaction, + &format!("snapshot {snapshot_id} deletes"), + input_stream, + receiver, + ) + .await?; + } - if !delta.is_empty() { - let transaction = self.allocate_follow_transaction(); + self.metrics + .follow_snapshots_total + .fetch_add(1, Ordering::Relaxed); + self.metrics + .set_last_ingested_sequence_number(snapshot.sequence_number()); + debug!( + "iceberg {}: ingested follow snapshot {snapshot_id} (sequence number {})", + &self.endpoint_name, + snapshot.sequence_number() + ); + } - // Inserts before deletes: a followed snapshot's added and - // removed files are disjoint, so ordering only affects - // intermediate state, not the final Z-set. - self.push_changed_files( - &table, - &schema, - &field_ids, - &name_mapping, - &delta.added, - true, - transaction.clone(), - &format!("snapshot {snapshot_id} inserts"), - input_stream, - receiver, - ) - .await?; - self.push_changed_files( - &table, - &schema, - &field_ids, - &name_mapping, - &delta.removed, - false, - transaction, - &format!("snapshot {snapshot_id} deletes"), - input_stream, - receiver, - ) - .await?; + if self.finish_at_end_snapshot(snapshot_id) { + self.reset_catchup_window(); + return Ok(()); } - // Commit the snapshot's transaction and record the resume point. - self.push_follow_boundary(snapshot_id); + // `catchup` holds one transaction open until the loop reaches the + // window's target snapshot; other modes commit after every + // snapshot. + let commit_boundary = + !catchup || self.catchup_target_snapshot_id() == Some(snapshot_id); + if commit_boundary { + self.push_follow_boundary(snapshot_id); + if catchup { + self.reset_catchup_window(); + } + } last_snapshot_id = Some(snapshot_id); - - self.metrics - .follow_snapshots_total - .fetch_add(1, Ordering::Relaxed); - self.metrics - .set_last_ingested_sequence_number(snapshot.sequence_number()); - info!( - "iceberg {}: ingested follow snapshot {snapshot_id} (sequence number {})", - &self.endpoint_name, - snapshot.sequence_number() - ); } } } @@ -1524,6 +1690,11 @@ impl IcebergInputEndpointInner { }); } + if let Err(e) = self.validate_end_snapshot(&table, snapshot_id) { + let _ = init_status_sender.send(Err(e)).await; + return; + } + let used_columns = match self .prepare_snapshot_query(&table, snapshot_id, &schema) .await @@ -1599,7 +1770,6 @@ impl IcebergInputEndpointInner { // start: the snapshot just ingested (snapshot_and_follow) or the // resolved starting snapshot (follow-only). self.follow_loop( - table.clone(), &used_columns, snapshot_id, input_stream.as_mut(), @@ -2096,6 +2266,39 @@ impl IcebergInputEndpointInner { } } + /// Require `end_snapshot_id` to name a committed snapshot after the start. + /// Snapshot ids are unordered, so a bound not in the followed history would + /// make the connector follow forever; reject it up front instead. + fn validate_end_snapshot( + &self, + table: &IcebergTable, + start_snapshot_id: Option, + ) -> Result<(), ControllerError> { + let Some(end) = self.config.end_snapshot_id else { + return Ok(()); + }; + if Some(end) == start_snapshot_id { + return Err(ControllerError::invalid_transport_configuration( + &self.endpoint_name, + &format!( + "'end_snapshot_id' {end} is the starting snapshot; set it to a later snapshot so the connector ingests at least one change before stopping" + ), + )); + } + let after = snapshots_after(table.metadata(), start_snapshot_id).map_err(|e| { + ControllerError::invalid_transport_configuration(&self.endpoint_name, &e.to_string()) + })?; + if !after.iter().any(|s| s.snapshot_id() == end) { + return Err(ControllerError::invalid_transport_configuration( + &self.endpoint_name, + &format!( + "'end_snapshot_id' {end} is not a snapshot committed after the starting snapshot in the table's current history. Set it to an existing snapshot that follows the start; an id that has not been committed yet cannot be used as an end bound because Iceberg snapshot ids are not ordered." + ), + )); + } + Ok(()) + } + /// The Iceberg snapshot the connector reads, `None` if the table has no matching snapshot. fn ingested_snapshot( &self, diff --git a/docs.feldera.com/docs/connectors/sources/iceberg.md b/docs.feldera.com/docs/connectors/sources/iceberg.md index 24df973b228..b6db8c13a05 100644 --- a/docs.feldera.com/docs/connectors/sources/iceberg.md +++ b/docs.feldera.com/docs/connectors/sources/iceberg.md @@ -30,12 +30,13 @@ The Iceberg input connector supports [fault tolerance](/pipelines/fault-toleranc | Property | Type | Description | |-----------------------------|--------|---------------| -| `mode`* | enum |

Table read mode. Supported values:

  • `snapshot` - read a snapshot of the table and stop.
  • `follow` - after the starting snapshot, continuously ingest new and deleted rows as the table commits new snapshots.
  • `snapshot_and_follow` - read a snapshot of the table, then switch to `follow` mode.

`follow` and `snapshot_and_follow` require an Iceberg catalog (set `catalog_type`); they cannot be used with `metadata_location`, which points at a fixed snapshot. See [Follow mode](#follow-mode) below.

| -| `transaction_mode` | enum | Determines how the connector breaks up its input into transactions. Supported values are `none` (default) and `snapshot`. See [below](#transactions) for details. | +| `mode`* | enum |

Table read mode. Supported values:

  • `snapshot` - read a snapshot of the table and stop.
  • `follow` - skip the initial snapshot and only ingest subsequent changes to the table (new and deleted rows) by following its transaction log.
  • `snapshot_and_follow` - read a snapshot of the table, then switch to `follow` mode.

`follow` and `snapshot_and_follow` require an Iceberg catalog (set `catalog_type`); they cannot be used with `metadata_location`, which points at a fixed snapshot. See [Follow mode](#follow-mode) below.

| +| `transaction_mode` | enum | Determines how the connector breaks up its input into transactions. Supported values are `none` (default), `snapshot`, `catchup`, and `always`. See [below](#transactions) for details. | | `timestamp_column` | string | Table column that serves as an event timestamp. When this option is specified, table rows are ingested in the timestamp order, respecting the [`LATENESS`](/sql/streaming#lateness-expressions) property of the column: each ingested row has a timestamp no more than `LATENESS` time units earlier than the most recent timestamp of any previously ingested row. See details [below](#ingesting-time-series-data-from-iceberg). | | `snapshot_filter` | string |

Optional row filter. When specified, only rows that satisfy the filter condition are included in the snapshot. The condition must be a valid SQL Boolean expression that can be used in the `where` clause of the `select * from snapshot where ..` query.

This option can be used to specify the range of event times to include in the snapshot, e.g.: `ts BETWEEN TIMESTAMP '2005-01-01 00:00:00' AND TIMESTAMP '2010-12-31 23:59:59'`.

| `snapshot_id` | integer|

Optional table snapshot id. When this option is set, the connector reads the specified snapshot of the table.

Note: at most one of `version` and `datetime` options can be specified. When neither of the two options is specified, the latest snapshot of the table is used.

| `datetime` | string |

Optional timestamp for the snapshot in the ISO-8601/RFC-3339 format, e.g., "2024-12-09T16:09:53+00:00". When this option is set, the connector reads the version of the table as of the specified point in time (based on the server time recorded in the transaction log, not the event time encoded in the data).

Note: at most one of `version` and `datetime` options can be specified. When neither of the two options is specified, the latest committed version of the table is used.

| +| `end_snapshot_id` | integer|

Optional final snapshot id. Valid only in `follow` and `snapshot_and_follow` modes. When set, the connector stops after fully ingesting the snapshot with this id, then signals end-of-input.

Unlike a Delta table version, an Iceberg snapshot id is not ordered, so this bound is an exact match: the id must name a snapshot committed after the starting snapshot and already present in the table's current history. The connector rejects any other value at startup (including a not-yet-committed id) rather than follow forever.

| | `metadata_location` | string | Location of the table metadata JSON file. This property is used to access an Iceberg table directly, without a catalog. It is mutually exclusive with the `catalog_type` property.| | `table_name` | string | Specifies the Iceberg table name within the catalog in the `namespace.table` format. This option is applicable when an Iceberg catalog is configured using the `catalog_type` property.| | `catalog_type` | enum | Type of the Iceberg catalog used to access the table. Supported options include `rest`, `glue`, and `s3tables`. This property is mutually exclusive with `metadata_location`.| @@ -173,11 +174,8 @@ nullable or have default values. In `follow` and `snapshot_and_follow` modes the connector continuously ingests changes committed to the table after its starting snapshot. It polls the catalog -for new snapshots and, for each one, diffs the snapshot against its predecessor -to find the data files it added and removed, ingesting added rows as inserts and -removed rows as deletes. Only the manifests a snapshot added are read, so the -cost of each step is proportional to the size of the change, not to the size of -the table. +for new snapshots and, for each one, ingests added rows as inserts and +removed rows as deletes. The starting snapshot is chosen the same way as in `snapshot` mode: by `snapshot_id`, by `datetime`, or, when neither is set, the latest snapshot at the @@ -193,18 +191,23 @@ Requirements and limitations: * **Copy-on-write only.** Follow mode reads copy-on-write changes. If a followed snapshot adds a merge-on-read delete file (position or equality deletes), the connector stops with an error. Configure the writer to use copy-on-write. -* **Compaction is skipped.** Snapshots that only rewrite files without changing - table contents (`operation = replace`) are recognized and skipped. ## Transactions The Iceberg connector can be configured to automatically initiate [transactions](/pipelines/transactions) -when ingesting the table snapshot. The `transaction_mode` property configures this feature: +when ingesting the table. The `transaction_mode` property configures this feature: * `none` - the connector does not group inputs into transactions. This is the default. -* `snapshot` - ingest the initial snapshot of the table in one or several transactions. - -### Ingesting table snapshot using transactions +* `snapshot` - ingest the initial snapshot of the table in one or several transactions. Changes + ingested afterward, in the follow phase, are not grouped into transactions. +* `catchup` - ingest the initial snapshot like `snapshot`. In the follow phase, the connector + groups all table commits that are already available into a single transaction: while catching up + on a backlog it ingests many commits per transaction, and once caught up it ingests about one + commit per transaction. This is the most efficient mode for backfill and steady-state following. +* `always` - ingest the initial snapshot like `snapshot`. In the follow phase, each table commit is + ingested in its own transaction. + +### Ingesting the table snapshot using transactions When `transaction_mode` is set to `snapshot`, the connector ingests the snapshot of the table in one or several transactions. The exact behavior depends on the value of the `timestamp_column` From 09f5e9e224fe4ba829e670378e31a21b2506511c Mon Sep 17 00:00:00 2001 From: Swanand Mulay <73115739+swanandx@users.noreply.github.com> Date: Thu, 30 Jul 2026 15:08:29 +0530 Subject: [PATCH 4/5] generate openapi.json --- openapi.json | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/openapi.json b/openapi.json index 5b5902107ae..7f8b7416c65 100644 --- a/openapi.json +++ b/openapi.json @@ -10065,6 +10065,12 @@ "description": "Optional timestamp for the snapshot in the ISO-8601/RFC-3339 format, e.g.,\n\"2024-12-09T16:09:53+00:00\".\n\nWhen this option is set, the connector finds and opens the snapshot of the table as of the\nspecified point in time (based on the server time recorded in the transaction\nlog, not the event time encoded in the data). In `snapshot` and `snapshot_and_follow`\nmodes, it retrieves this snapshot. In `follow` and `snapshot_and_follow` modes, it\nfollows transaction log records **after** this snapshot.\n\nNote: at most one of `snapshot_id` and `datetime` options can be specified.\nWhen neither of the two options is specified, the latest committed version of the table\nis used.", "nullable": true }, + "end_snapshot_id": { + "type": "integer", + "format": "int64", + "description": "Optional final snapshot id.\n\nValid only in `follow` and `snapshot_and_follow` modes.\n\nWhen set, the connector stops after fully ingesting the snapshot with\nthis id, signaling end-of-input. Unlike a Delta table version, an Iceberg\nsnapshot id is not ordered, so the bound is an exact match: the id must\nname a snapshot committed after the starting snapshot and already present\nin the table's current history. The connector rejects any other value at\nstartup, including a not-yet-committed id, rather than follow forever.", + "nullable": true + }, "max_retries": { "type": "integer", "format": "int32", @@ -10121,10 +10127,12 @@ }, "IcebergTransactionMode": { "type": "string", - "description": "Iceberg table transaction mode.\n\nDetermines how the connector breaks up its input into Feldera transactions.\n\n* `none` - the connector does not break up its input into transactions.\n* `snapshot` - ingest the initial snapshot of the table in one or several transactions.\n\n# How the table snapshot is ingested using transactions\n\nWhen `transaction_mode` is set to `snapshot`, the connector ingests the snapshot in one\nor several transactions, depending on `timestamp_column`. If `timestamp_column` is not set,\nthe whole snapshot is ingested in a single Feldera transaction. If `timestamp_column` is set,\nthe connector ingests the snapshot in a series of timestamp ranges of width equal to the\n`LATENESS` attribute of the column, each range in a separate transaction.", + "description": "Iceberg table transaction mode.\n\nDetermines how the connector breaks up its input into Feldera transactions.\n\n* `none` - the connector does not group its input into transactions.\n* `snapshot` - ingest the initial snapshot in one or more transactions (see below). Changes\ningested afterward, in the follow phase, are not grouped into transactions.\n* `catchup` - ingest the initial snapshot like `snapshot`. In the follow phase, the connector\ngroups all table commits that are already available into a single transaction: while catching\nup on a backlog it ingests many commits per transaction, and once caught up it ingests about\none commit per transaction. Most efficient for backfill and steady-state following.\n* `always` - ingest the initial snapshot like `snapshot`. In the follow phase, each table commit\nis ingested in its own transaction.\n\n# How the table snapshot is ingested using transactions\n\nFor the initial snapshot (`snapshot`, `catchup`, and `always` all behave the same), the\nconnector ingests the snapshot in one or several transactions, depending on `timestamp_column`.\nIf `timestamp_column` is not set, the whole snapshot is ingested in a single Feldera\ntransaction. If `timestamp_column` is set, the connector ingests the snapshot in a series of\ntimestamp ranges of width equal to the `LATENESS` attribute of the column, each range in a\nseparate transaction.", "enum": [ "none", - "snapshot" + "snapshot", + "catchup", + "always" ] }, "InputEndpointConfig": { From 11e072a1ab68d440a6ed0cc89426421097b5dc89 Mon Sep 17 00:00:00 2001 From: Swanand Mulay <73115739+swanandx@users.noreply.github.com> Date: Fri, 31 Jul 2026 14:17:59 +0530 Subject: [PATCH 5/5] [connectors] Apply Iceberg follow deletes before inserts A copy-on-write rewrite emits the same primary key in both the removed and added files. Applying the deletes first lets the insert win on a keyed relation, matching the Delta connector. - add a follow test for the copy-on-write delete path - reword the catchup and end_snapshot_id docs; drop the Delta comparison Signed-off-by: Swanand Mulay <73115739+swanandx@users.noreply.github.com> --- crates/adapters/src/test/iceberg.rs | 30 +++++++++++++++++++ crates/feldera-types/src/transport/iceberg.rs | 16 +++++----- crates/iceberg/src/input.rs | 17 +++++------ crates/iceberg/src/test/follow_table.py | 11 +++++-- .../docs/connectors/sources/iceberg.md | 2 +- openapi.json | 4 +-- 6 files changed, 58 insertions(+), 22 deletions(-) diff --git a/crates/adapters/src/test/iceberg.rs b/crates/adapters/src/test/iceberg.rs index 5936584eac5..537fa86f8d6 100644 --- a/crates/adapters/src/test/iceberg.rs +++ b/crates/adapters/src/test/iceberg.rs @@ -1144,6 +1144,36 @@ fn iceberg_rest_follow_transaction_catchup_end_snapshot_id() { }); } +/// Covers the follow delete path (removed files), which the append-only follow +/// tests miss. A copy-on-write overwrite removes the old files and adds new +/// ones; drop one row and edit another, then expect the four survivors. +#[test] +#[cfg(feature = "iceberg-tests-follow")] +fn iceberg_rest_follow_copy_on_write_delete() { + let all = data(5); + let ns = env_or("FELDERA_ICEBERG_NAMESPACE", "follow_ns"); + let table = &format!("{ns}.cow_delete"); + + follow_table_op("create", table, &all); + + let mut updated = all[..4].to_vec(); + updated[2].s = "cow-updated".to_string(); + + with_follow_pipeline(table, "snapshot_and_follow", |pipeline, out_path| { + wait(|| ingested_records(pipeline) >= 5, 120_000).expect("timed out ingesting snapshot"); + follow_table_op("overwrite", table, &updated); + // Overwrite reads 5 deletes + 4 inserts, so ingested reaches 14. + wait(|| ingested_records(pipeline) >= 14, 120_000) + .expect("timed out following the overwrite"); + // 8 records: 5 snapshot inserts, 2 for the edited row, 1 for the dropped + // row. Wait for the last so the fold sees complete output. + wait(|| output_record_count(out_path) >= 8, 60_000).expect("timed out writing output"); + + let zset = output_zset(out_path); + assert_eq!(zset, expected_zset(&updated)); + }); +} + /// Number of complete `insert_delete` records written to the output file so /// far (one JSON value per line). #[cfg(feature = "iceberg-tests-follow")] diff --git a/crates/feldera-types/src/transport/iceberg.rs b/crates/feldera-types/src/transport/iceberg.rs index 8b1e2385821..ac51fda4890 100644 --- a/crates/feldera-types/src/transport/iceberg.rs +++ b/crates/feldera-types/src/transport/iceberg.rs @@ -56,9 +56,9 @@ pub enum IcebergCatalogType { /// * `snapshot` - ingest the initial snapshot in one or more transactions (see below). Changes /// ingested afterward, in the follow phase, are not grouped into transactions. /// * `catchup` - ingest the initial snapshot like `snapshot`. In the follow phase, the connector -/// groups all table commits that are already available into a single transaction: while catching -/// up on a backlog it ingests many commits per transaction, and once caught up it ingests about -/// one commit per transaction. Most efficient for backfill and steady-state following. +/// batches all currently available table commits into a single transaction. Once that transaction +/// completes, it checks for commits added since the transaction began, ingests them in the next +/// transaction, and repeats continuously. Most efficient for backfill and steady-state following. /// * `always` - ingest the initial snapshot like `snapshot`. In the follow phase, each table commit /// is ingested in its own transaction. /// @@ -297,11 +297,11 @@ pub struct IcebergReaderConfig { /// Valid only in `follow` and `snapshot_and_follow` modes. /// /// When set, the connector stops after fully ingesting the snapshot with - /// this id, signaling end-of-input. Unlike a Delta table version, an Iceberg - /// snapshot id is not ordered, so the bound is an exact match: the id must - /// name a snapshot committed after the starting snapshot and already present - /// in the table's current history. The connector rejects any other value at - /// startup, including a not-yet-committed id, rather than follow forever. + /// this id, signaling end-of-input. Iceberg snapshot ids are not ordered, so + /// the bound is an exact match: the id must name a snapshot committed after + /// the starting snapshot and already present in the table's current history. + /// The connector rejects any other value at startup, including a + /// not-yet-committed id, rather than follow forever. pub end_snapshot_id: Option, /// Location of the table metadata JSON file. diff --git a/crates/iceberg/src/input.rs b/crates/iceberg/src/input.rs index 358fefed611..cf351e23fc5 100644 --- a/crates/iceberg/src/input.rs +++ b/crates/iceberg/src/input.rs @@ -1587,18 +1587,17 @@ impl IcebergInputEndpointInner { if !delta.is_empty() { let transaction = self.follow_data_transaction(); - // Inserts before deletes: a followed snapshot's added and - // removed files are disjoint, so ordering only affects - // intermediate state, not the final Z-set. + // Deletes before inserts so a same-key rewrite keeps the new + // row on a keyed relation (matches the Delta connector). self.push_changed_files( &table, &schema, &field_ids, &name_mapping, - &delta.added, - true, + &delta.removed, + false, transaction.clone(), - &format!("snapshot {snapshot_id} inserts"), + &format!("snapshot {snapshot_id} deletes"), input_stream, receiver, ) @@ -1608,10 +1607,10 @@ impl IcebergInputEndpointInner { &schema, &field_ids, &name_mapping, - &delta.removed, - false, + &delta.added, + true, transaction, - &format!("snapshot {snapshot_id} deletes"), + &format!("snapshot {snapshot_id} inserts"), input_stream, receiver, ) diff --git a/crates/iceberg/src/test/follow_table.py b/crates/iceberg/src/test/follow_table.py index 0415ce6cc4e..db2879ad758 100644 --- a/crates/iceberg/src/test/follow_table.py +++ b/crates/iceberg/src/test/follow_table.py @@ -113,7 +113,9 @@ def arrow_chunk(json_file): def main(): parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument("--op", choices=["create", "append"], required=True) + parser.add_argument( + "--op", choices=["create", "append", "overwrite"], required=True + ) parser.add_argument("--table", required=True, help="table as 'namespace.name'") parser.add_argument("--json-file", required=True, help="ndjson chunk to append") args = parser.parse_args() @@ -134,7 +136,12 @@ def main(): else: table = cat.load_table(args.table) - table.append(arrow_chunk(args.json_file)) + chunk = arrow_chunk(args.json_file) + if args.op == "overwrite": + # Copy-on-write rewrite: removes the old data files, adds `chunk`. + table.overwrite(chunk) + else: + table.append(chunk) # Print the current snapshot id so the caller can log progress. print(table.metadata.current_snapshot_id) diff --git a/docs.feldera.com/docs/connectors/sources/iceberg.md b/docs.feldera.com/docs/connectors/sources/iceberg.md index b6db8c13a05..88d5a46814c 100644 --- a/docs.feldera.com/docs/connectors/sources/iceberg.md +++ b/docs.feldera.com/docs/connectors/sources/iceberg.md @@ -36,7 +36,7 @@ The Iceberg input connector supports [fault tolerance](/pipelines/fault-toleranc | `snapshot_filter` | string |

Optional row filter. When specified, only rows that satisfy the filter condition are included in the snapshot. The condition must be a valid SQL Boolean expression that can be used in the `where` clause of the `select * from snapshot where ..` query.

This option can be used to specify the range of event times to include in the snapshot, e.g.: `ts BETWEEN TIMESTAMP '2005-01-01 00:00:00' AND TIMESTAMP '2010-12-31 23:59:59'`.

| `snapshot_id` | integer|

Optional table snapshot id. When this option is set, the connector reads the specified snapshot of the table.

Note: at most one of `version` and `datetime` options can be specified. When neither of the two options is specified, the latest snapshot of the table is used.

| `datetime` | string |

Optional timestamp for the snapshot in the ISO-8601/RFC-3339 format, e.g., "2024-12-09T16:09:53+00:00". When this option is set, the connector reads the version of the table as of the specified point in time (based on the server time recorded in the transaction log, not the event time encoded in the data).

Note: at most one of `version` and `datetime` options can be specified. When neither of the two options is specified, the latest committed version of the table is used.

| -| `end_snapshot_id` | integer|

Optional final snapshot id. Valid only in `follow` and `snapshot_and_follow` modes. When set, the connector stops after fully ingesting the snapshot with this id, then signals end-of-input.

Unlike a Delta table version, an Iceberg snapshot id is not ordered, so this bound is an exact match: the id must name a snapshot committed after the starting snapshot and already present in the table's current history. The connector rejects any other value at startup (including a not-yet-committed id) rather than follow forever.

| +| `end_snapshot_id` | integer|

Optional final snapshot id. Valid only in `follow` and `snapshot_and_follow` modes. When set, the connector stops after fully ingesting the snapshot with this id, then signals end-of-input.

Iceberg snapshot ids are not ordered, so this bound is an exact match: the id must name a snapshot committed after the starting snapshot and already present in the table's current history. The connector rejects any other value at startup (including a not-yet-committed id) rather than follow forever.

| | `metadata_location` | string | Location of the table metadata JSON file. This property is used to access an Iceberg table directly, without a catalog. It is mutually exclusive with the `catalog_type` property.| | `table_name` | string | Specifies the Iceberg table name within the catalog in the `namespace.table` format. This option is applicable when an Iceberg catalog is configured using the `catalog_type` property.| | `catalog_type` | enum | Type of the Iceberg catalog used to access the table. Supported options include `rest`, `glue`, and `s3tables`. This property is mutually exclusive with `metadata_location`.| diff --git a/openapi.json b/openapi.json index 7f8b7416c65..d5559473ad4 100644 --- a/openapi.json +++ b/openapi.json @@ -10068,7 +10068,7 @@ "end_snapshot_id": { "type": "integer", "format": "int64", - "description": "Optional final snapshot id.\n\nValid only in `follow` and `snapshot_and_follow` modes.\n\nWhen set, the connector stops after fully ingesting the snapshot with\nthis id, signaling end-of-input. Unlike a Delta table version, an Iceberg\nsnapshot id is not ordered, so the bound is an exact match: the id must\nname a snapshot committed after the starting snapshot and already present\nin the table's current history. The connector rejects any other value at\nstartup, including a not-yet-committed id, rather than follow forever.", + "description": "Optional final snapshot id.\n\nValid only in `follow` and `snapshot_and_follow` modes.\n\nWhen set, the connector stops after fully ingesting the snapshot with\nthis id, signaling end-of-input. Iceberg snapshot ids are not ordered, so\nthe bound is an exact match: the id must name a snapshot committed after\nthe starting snapshot and already present in the table's current history.\nThe connector rejects any other value at startup, including a\nnot-yet-committed id, rather than follow forever.", "nullable": true }, "max_retries": { @@ -10127,7 +10127,7 @@ }, "IcebergTransactionMode": { "type": "string", - "description": "Iceberg table transaction mode.\n\nDetermines how the connector breaks up its input into Feldera transactions.\n\n* `none` - the connector does not group its input into transactions.\n* `snapshot` - ingest the initial snapshot in one or more transactions (see below). Changes\ningested afterward, in the follow phase, are not grouped into transactions.\n* `catchup` - ingest the initial snapshot like `snapshot`. In the follow phase, the connector\ngroups all table commits that are already available into a single transaction: while catching\nup on a backlog it ingests many commits per transaction, and once caught up it ingests about\none commit per transaction. Most efficient for backfill and steady-state following.\n* `always` - ingest the initial snapshot like `snapshot`. In the follow phase, each table commit\nis ingested in its own transaction.\n\n# How the table snapshot is ingested using transactions\n\nFor the initial snapshot (`snapshot`, `catchup`, and `always` all behave the same), the\nconnector ingests the snapshot in one or several transactions, depending on `timestamp_column`.\nIf `timestamp_column` is not set, the whole snapshot is ingested in a single Feldera\ntransaction. If `timestamp_column` is set, the connector ingests the snapshot in a series of\ntimestamp ranges of width equal to the `LATENESS` attribute of the column, each range in a\nseparate transaction.", + "description": "Iceberg table transaction mode.\n\nDetermines how the connector breaks up its input into Feldera transactions.\n\n* `none` - the connector does not group its input into transactions.\n* `snapshot` - ingest the initial snapshot in one or more transactions (see below). Changes\ningested afterward, in the follow phase, are not grouped into transactions.\n* `catchup` - ingest the initial snapshot like `snapshot`. In the follow phase, the connector\nbatches all currently available table commits into a single transaction. Once that transaction\ncompletes, it checks for commits added since the transaction began, ingests them in the next\ntransaction, and repeats continuously. Most efficient for backfill and steady-state following.\n* `always` - ingest the initial snapshot like `snapshot`. In the follow phase, each table commit\nis ingested in its own transaction.\n\n# How the table snapshot is ingested using transactions\n\nFor the initial snapshot (`snapshot`, `catchup`, and `always` all behave the same), the\nconnector ingests the snapshot in one or several transactions, depending on `timestamp_column`.\nIf `timestamp_column` is not set, the whole snapshot is ingested in a single Feldera\ntransaction. If `timestamp_column` is set, the connector ingests the snapshot in a series of\ntimestamp ranges of width equal to the `LATENESS` attribute of the column, each range in a\nseparate transaction.", "enum": [ "none", "snapshot",