From 54e5f7f54797c59b76e1657e3a384f63bd23430b Mon Sep 17 00:00:00 2001 From: Swanand Mulay <73115739+swanandx@users.noreply.github.com> Date: Fri, 3 Jul 2026 02:55:03 +0530 Subject: [PATCH] [connectors] read only the DV delta on rewrites in delta follow mode A same-path Add+Remove in the transaction log is a deletion-vector rewrite: the Parquet bytes are unchanged and only the DV differs. Follow mode read the whole file on each side and let the circuit cancel the unchanged rows, so a K-row change ingested roughly 2N-K records. Pair such actions by path and decode both deletion vectors, then read only the rows the DV newly masks (deleted) or un-masks (restored). A K-row change now costs K records. The Python platform DV suite asserts total_input_records to catch the amplification Signed-off-by: Swanand Mulay <73115739+swanandx@users.noreply.github.com> --- .../integrated/delta_table/deletion_vector.rs | 250 +++++++++++++---- .../src/integrated/delta_table/input.rs | 263 ++++++++++++++---- .../test_delta_input_deletion_vectors.py | 136 ++++++--- 3 files changed, 508 insertions(+), 141 deletions(-) diff --git a/crates/adapters/src/integrated/delta_table/deletion_vector.rs b/crates/adapters/src/integrated/delta_table/deletion_vector.rs index d4659fe672a..8242e6c3a53 100644 --- a/crates/adapters/src/integrated/delta_table/deletion_vector.rs +++ b/crates/adapters/src/integrated/delta_table/deletion_vector.rs @@ -103,55 +103,78 @@ fn abbreviate(value: &str) -> String { } } -/// Build a [`TableProvider`] over the Parquet file at `path` that skips the -/// rows flagged by `bitmap`. +/// Build a [`TableProvider`] over the Parquet file at `path` that reads the rows +/// `mode` picks out of `bitmap` (see [`ReadMode`]). /// -/// The bitmap indexes rows by physical position, so the file is read in -/// order through a single-partition [`StreamingTable`] (a `ListingTable` -/// could split and reorder it). The Parquet decoder skips deleted rows via a -/// [`RowSelection`]; memory stays bounded to one batch. +/// The bitmap indexes rows by physical position, so the file is read in order +/// through a single-partition [`StreamingTable`] (a `ListingTable` could split +/// and reorder it). Row selection happens inside the Parquet decoder, so memory +/// stays bounded to one batch. /// -/// `logical_schema` is the table's Arrow schema, restricted by the caller to -/// the columns it wants read. Batches are projected to it by name (missing -/// columns become NULL), and it doubles as the Parquet projection: columns it -/// does not name are never decoded. The caller must restrict it itself, because +/// `logical_schema` is the table's Arrow schema, restricted by the caller to the +/// columns it wants read. Batches are projected to it by name (missing columns +/// become NULL), and it doubles as the Parquet projection: columns it does not +/// name are never decoded. The caller must restrict it itself, because /// [`StreamingTable`] does not push projections down. -pub(crate) async fn masked_parquet_table( +pub(crate) async fn filtered_parquet_table( store: Arc, path: Path, bitmap: RoaringTreemap, logical_schema: SchemaRef, + mode: ReadMode, ) -> AnyResult> { let partition = MaskedParquetPartition { store, path, bitmap: Arc::new(bitmap), schema: Arc::clone(&logical_schema), + mode, }; let provider = StreamingTable::try_new(logical_schema, vec![Arc::new(partition)]) - .map_err(|e| anyhow!("failed to build DV-masked streaming table: {e}"))?; + .map_err(|e| anyhow!("failed to build DV-filtered streaming table: {e}"))?; Ok(Arc::new(provider)) } +/// Which rows [`filtered_parquet_table`] reads, relative to its `bitmap`. +#[derive(Clone, Copy)] +pub(crate) enum ReadMode { + /// Read only the rows in the bitmap: a deletion-vector delta (the rows a + /// same-path rewrite masked or un-masked). + InBitmap, + /// Read the rows not in the bitmap: apply a deletion vector, so the file's + /// live rows come through. + NotInBitmap, +} + +impl ReadMode { + /// Does the read keep the bitmap rows (rather than the rest)? + fn reads_bitmap_rows(self) -> bool { + matches!(self, ReadMode::InBitmap) + } +} + /// Single-partition [`PartitionStream`] that lazily opens a Parquet file and -/// drops rows flagged by `bitmap` as batches flow through. +/// keeps or drops the rows flagged by `bitmap` (per `mode`) as batches flow +/// through. struct MaskedParquetPartition { store: Arc, path: Path, - /// Deleted row positions for this one file. A `RoaringTreemap` stays - /// compact even for dense deletes, and it is dropped once the partition - /// finishes streaming, so the footprint is bounded and short-lived. + /// Row positions this partition acts on. A `RoaringTreemap` stays compact + /// even for dense sets, and it is dropped once the partition finishes + /// streaming, so the footprint is bounded and short-lived. bitmap: Arc, /// The Delta logical schema; may differ from the file's own schema under /// schema evolution. schema: SchemaRef, + /// Which rows to read, relative to `bitmap`. + mode: ReadMode, } impl fmt::Debug for MaskedParquetPartition { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.debug_struct("MaskedParquetPartition") .field("path", &self.path) - .field("deleted_rows", &self.bitmap.len()) + .field("bitmap_rows", &self.bitmap.len()) .finish() } } @@ -226,30 +249,41 @@ fn logical_projection_mask( ProjectionMask::roots(builder.parquet_schema(), roots) } -/// Build the [`RowSelection`] selecting every row in `0..total_rows` that is -/// not in `bitmap`. Positions past `total_rows` are ignored. -fn bitmap_to_row_selection(bitmap: &RoaringTreemap, total_rows: u64) -> RowSelection { - // One selector per *run* of selected/skipped rows, so the vector is small - // when deletes are clustered and grows to O(deletes) only when they alternate - // with live rows. It is built per file, consumed right away to build the - // Parquet stream, and dropped, so the footprint is bounded and short-lived. +/// Append a [`RowSelector`] for `count` rows to `selectors`, selecting or +/// skipping them per `select`, merging into the previous selector when it is +/// the same kind. A zero-length selector is a no-op. `select`/skip is parquet's +/// [`RowSelector`] vocabulary, one level below [`ReadMode`]. +fn push_selector(selectors: &mut Vec, select: bool, count: u64) { + if count == 0 { + return; + } + let count = count as usize; + match selectors.last_mut() { + Some(last) if last.skip != select => last.row_count += count, + _ if select => selectors.push(RowSelector::select(count)), + _ => selectors.push(RowSelector::skip(count)), + } +} + +/// Build the [`RowSelection`] over `0..total_rows` that `mode` implies for +/// `bitmap`. Positions past `total_rows` are ignored. +/// +/// The result holds one selector per run of like rows, so it is O(runs): small +/// when the bitmap is clustered, O(bitmap) only when bitmap rows alternate with +/// the rest. It is built per file, consumed to start the Parquet stream, then +/// dropped. +fn bitmap_to_selection(bitmap: &RoaringTreemap, total_rows: u64, mode: ReadMode) -> RowSelection { + // A bitmap row is selected exactly when `mode` reads bitmap rows; the gaps + // between them are selected in the other case. + let select_bitmap_rows = mode.reads_bitmap_rows(); let mut selectors: Vec = Vec::new(); - // First row not yet covered by a selector. let mut cursor: u64 = 0; - for deleted in bitmap.iter().take_while(|&pos| pos < total_rows) { - if deleted > cursor { - selectors.push(RowSelector::select((deleted - cursor) as usize)); - } - // Extend the previous skip when deletes are consecutive. - match selectors.last_mut() { - Some(last) if last.skip => last.row_count += 1, - _ => selectors.push(RowSelector::skip(1)), - } - cursor = deleted + 1; - } - if cursor < total_rows { - selectors.push(RowSelector::select((total_rows - cursor) as usize)); + for pos in bitmap.iter().take_while(|&pos| pos < total_rows) { + push_selector(&mut selectors, !select_bitmap_rows, pos - cursor); // gap before this row + push_selector(&mut selectors, select_bitmap_rows, 1); // the bitmap row itself + cursor = pos + 1; } + push_selector(&mut selectors, !select_bitmap_rows, total_rows - cursor); // trailing gap RowSelection::from(selectors) } @@ -266,6 +300,7 @@ impl PartitionStream for MaskedParquetPartition { let path = self.path.clone(); let bitmap = Arc::clone(&self.bitmap); let logical_schema = Arc::clone(&self.schema); + let mode = self.mode; let stream = try_stream! { let reader = ParquetObjectReader::new(store, path.clone()); @@ -279,8 +314,9 @@ impl PartitionStream for MaskedParquetPartition { let total_rows = builder.metadata().file_metadata().num_rows() as u64; // Decode only the columns the logical schema names. let mask = logical_projection_mask(&builder, &logical_schema); - // Skip deleted rows inside the decoder. - let selection = bitmap_to_row_selection(&bitmap, total_rows); + // Pick rows inside the decoder: skip the flagged rows (apply a DV) or + // keep only them (read a DV delta). + let selection = bitmap_to_selection(&bitmap, total_rows, mode); let mut parquet_stream = builder .with_projection(mask) .with_row_selection(selection) @@ -336,7 +372,7 @@ mod tests { fn check(deleted: &[u64], total_rows: u64) { let bitmap = RoaringTreemap::from_iter(deleted.iter().copied()); - let selection = bitmap_to_row_selection(&bitmap, total_rows); + let selection = bitmap_to_selection(&bitmap, total_rows, ReadMode::NotInBitmap); assert_eq!( selected_rows(&selection), expected_rows(deleted, total_rows) @@ -360,7 +396,7 @@ mod tests { fn contiguous_run_merges_into_one_skip() { check(&[2, 3, 4], 10); let bitmap = RoaringTreemap::from_iter([2u64, 3, 4]); - let selection = bitmap_to_row_selection(&bitmap, 10); + let selection = bitmap_to_selection(&bitmap, 10, ReadMode::NotInBitmap); // select(2), skip(3), select(5) assert_eq!(selection.iter().count(), 3); } @@ -388,6 +424,52 @@ mod tests { } } + /// Rows in `0..total_rows` that *are* in `kept`. + fn expected_kept_rows(kept: &[u64], total_rows: u64) -> Vec { + (0..total_rows).filter(|row| kept.contains(row)).collect() + } + + fn check_keep(kept: &[u64], total_rows: u64) { + let bitmap = RoaringTreemap::from_iter(kept.iter().copied()); + let selection = bitmap_to_selection(&bitmap, total_rows, ReadMode::InBitmap); + assert_eq!( + selected_rows(&selection), + expected_kept_rows(kept, total_rows) + ); + } + + #[test] + fn keep_selection_edge_cases() { + check_keep(&[], 0); + check_keep(&[], 10); // keep nothing + check_keep(&[0], 5); // leading + check_keep(&[4], 5); // trailing + check_keep(&[1, 2, 3], 10); // contiguous run + check_keep(&[0, 1, 2], 3); // keep everything + check_keep(&[1, 7, 100], 5); // positions past EOF ignored + } + + proptest! { + /// The keep-selection picks exactly the bitmap, and is the complement of + /// the skip-selection over the same bitmap. + #[test] + fn keep_selection_is_the_bitmap( + kept in proptest::collection::btree_set(0u64..200, 0..50), + total_rows in 0u64..200, + ) { + let kept: Vec = kept.into_iter().collect(); + check_keep(&kept, total_rows); + + let bitmap = RoaringTreemap::from_iter(kept.iter().copied()); + let keep = selected_rows(&bitmap_to_selection(&bitmap, total_rows, ReadMode::InBitmap)); + let skip = selected_rows(&bitmap_to_selection(&bitmap, total_rows, ReadMode::NotInBitmap)); + let mut union: Vec = keep.iter().chain(skip.iter()).copied().collect(); + union.sort_unstable(); + prop_assert!(keep.iter().all(|r| !skip.contains(r))); + prop_assert_eq!(union, (0..total_rows).collect::>()); + } + } + /// Every delta-rs storage type maps to its `delta_kernel` counterpart and /// the remaining descriptor fields are copied verbatim. The decode itself /// is `delta_kernel`'s, but this conversion is ours, so it is tested here: @@ -443,11 +525,11 @@ mod tests { .unwrap() } - /// End-to-end check of [`masked_parquet_table`] against a real Parquet - /// file: DV-flagged rows are skipped inside the decoder, and the logical - /// schema drives the read. A file column it omits is pruned, a column it - /// widens is cast (`Int32` to `Int64`), and a column the file lacks comes - /// back NULL (schema evolution). + /// End-to-end check of [`filtered_parquet_table`] with [`ReadMode::NotInBitmap`] + /// against a real Parquet file: DV-flagged rows are skipped inside the + /// decoder, and the logical schema drives the read. A file column it omits is + /// pruned, a column it widens is cast (`Int32` to `Int64`), and a column the + /// file lacks comes back NULL (schema evolution). #[tokio::test] async fn masked_reader_applies_dv_and_logical_schema() { const TOTAL_ROWS: usize = 200; @@ -478,11 +560,12 @@ mod tests { let deleted = RoaringTreemap::from_iter((0..TOTAL_ROWS as u64).filter(|i| i % 2 == 0)); let store = unloaded_table(dir.path()).log_store().object_store(None); - let provider = masked_parquet_table( + let provider = filtered_parquet_table( store, Path::from("data.parquet"), deleted, Arc::clone(&logical), + ReadMode::NotInBitmap, ) .await .unwrap(); @@ -516,4 +599,75 @@ mod tests { let expected: Vec = (0..TOTAL_ROWS as i64).filter(|i| i % 2 != 0).collect(); assert_eq!(got, expected, "masked rows mismatch"); } + + /// End-to-end check of [`filtered_parquet_table`] with [`ReadMode::InBitmap`]: + /// the inverse of the masked reader. Given a bitmap of row positions (a + /// deletion-vector delta), it reads *only* those rows and applies the logical + /// schema, so a K-row change costs K rows. Uses the same file/schema shape as + /// the masked-reader test. + #[tokio::test] + async fn selected_reader_reads_only_the_bitmap_rows() { + const TOTAL_ROWS: usize = 200; + let dir = TempDir::new().unwrap(); + + let file_schema = Arc::new(ArrowSchema::new(vec![ + ArrowField::new("id", ArrowDataType::Int32, false), + ArrowField::new("payload", ArrowDataType::Utf8, false), + ])); + let ids = Int32Array::from_iter_values(0..TOTAL_ROWS as i32); + let payloads = StringArray::from_iter_values((0..TOTAL_ROWS).map(|i| format!("row_{i}"))); + let batch = RecordBatch::try_new( + Arc::clone(&file_schema), + vec![Arc::new(ids), Arc::new(payloads)], + ) + .unwrap(); + let file = std::fs::File::create(dir.path().join("data.parquet")).unwrap(); + let mut writer = parquet::arrow::ArrowWriter::try_new(file, file_schema, None).unwrap(); + writer.write(&batch).unwrap(); + writer.close().unwrap(); + + let logical = Arc::new(ArrowSchema::new(vec![ + ArrowField::new("id", ArrowDataType::Int64, true), + ArrowField::new("added_later", ArrowDataType::Utf8, true), + ])); + // A sparse, clustered set spanning both ends, so runs and gaps are exercised. + let wanted: Vec = [0u64, 1, 2, 50, 51, 199].into_iter().collect(); + let selected = RoaringTreemap::from_iter(wanted.iter().copied()); + + let store = unloaded_table(dir.path()).log_store().object_store(None); + let provider = filtered_parquet_table( + store, + Path::from("data.parquet"), + selected, + Arc::clone(&logical), + ReadMode::InBitmap, + ) + .await + .unwrap(); + let batches = SessionContext::new() + .read_table(provider) + .unwrap() + .collect() + .await + .unwrap(); + + let mut got: Vec = Vec::new(); + for batch in &batches { + assert_eq!(batch.schema().as_ref(), logical.as_ref()); + let id = batch + .column(0) + .as_any() + .downcast_ref::() + .unwrap(); + got.extend(id.iter().map(|v| v.unwrap())); + assert_eq!( + batch.column(1).null_count(), + batch.num_rows(), + "the column absent from the file must be all NULL" + ); + } + got.sort(); + let expected: Vec = wanted.iter().map(|&r| r as i64).collect(); + assert_eq!(got, expected, "must read exactly the bitmap rows"); + } } diff --git a/crates/adapters/src/integrated/delta_table/input.rs b/crates/adapters/src/integrated/delta_table/input.rs index d804ada5745..75fe65db27b 100644 --- a/crates/adapters/src/integrated/delta_table/input.rs +++ b/crates/adapters/src/integrated/delta_table/input.rs @@ -1,6 +1,8 @@ use crate::catalog::{ArrowStream, InputCollectionHandle}; use crate::format::InputBuffer; -use crate::integrated::delta_table::deletion_vector::{masked_parquet_table, read_deletion_vector}; +use crate::integrated::delta_table::deletion_vector::{ + ReadMode, filtered_parquet_table, read_deletion_vector, +}; use crate::integrated::delta_table::{delta_input_serde_config, register_storage_handlers}; use crate::transport::{InputEndpoint, InputQueue, InputReaderCommand, IntegratedInputEndpoint}; use crate::util::JobQueue; @@ -49,6 +51,7 @@ use feldera_types::program_schema::{Field, Relation}; use feldera_types::transport::delta_table::{DeltaTableReaderConfig, DeltaTableTransactionMode}; use futures_util::StreamExt; use rand::Rng; +use roaring::RoaringTreemap; use serde::{Deserialize, Serialize}; use serde_json::Value as JsonValue; use std::cmp::min; @@ -2748,14 +2751,67 @@ impl DeltaTableInputEndpointInner { // go using `ListingTable`, which understands partitioning and can probably // parallelize the load. - // Process deletes before inserts. Semantically, delete actions in - // are applied before insert actions; however the delta standard doesn't + // A same-path Add+Remove is a deletion-vector rewrite (DELETE/MERGE): + // Delta files are immutable and path-addressed, so the bytes are + // unchanged and only the DV differs. Its logical change is just the + // rows the DV newly masks (deleted) or un-masks (restored), so we + // read only those rows rather than re-reading the whole file. + // + // Index the data-change `Add` actions by path (a metadata-only + // rewrite has data_change == false and never pairs). The Delta + // protocol allows at most one `Add` per path in a commit. + let adds_by_path: BTreeMap<&str, &AddAction> = actions + .iter() + .filter_map(|action| match action { + Action::Add(add) if add.data_change => Some((add.path.as_str(), add)), + _ => None, + }) + .collect(); + + // Row-position delta of each same-path rewrite, keyed by file path. + // The pair is (newly deleted = rows the new DV masks but the old did + // not, restored = rows the old DV masked but the new does not). + let mut dv_deltas: BTreeMap<&str, (RoaringTreemap, RoaringTreemap)> = BTreeMap::new(); + for action in actions { + let Action::Remove(remove) = action else { + continue; + }; + if !remove.data_change { + continue; + } + if let Some(add) = adds_by_path.get(remove.path.as_str()) { + let description = format!("file '{}'", remove.path); + let old_dv = self + .decode_dv(table, remove.deletion_vector.as_ref(), &description) + .await?; + let new_dv = self + .decode_dv(table, add.deletion_vector.as_ref(), &description) + .await?; + // newly deleted = in the new DV but not the old; restored = the reverse. + dv_deltas.insert(remove.path.as_str(), (&new_dv - &old_dv, &old_dv - &new_dv)); + } + } + + // Process deletes before inserts. Semantically, delete actions are + // applied before insert actions; however the delta standard doesn't // guarantee that actions occur in any particular order in the transaction log // entry. + // + // Each pass hands `process_follow_action` the row-index delta for its + // side of a same-path pair (the `Remove` retracts the newly-deleted + // rows; the `Add` inserts the restored rows) or `None` for an unpaired + // action, which is then read whole. `dv_deltas` is keyed only by + // data-change pairs, so a metadata-only action never matches. for action in actions { - if matches!(action, Action::Remove(_)) { - self.process_action( + if let Action::Remove(remove) = action { + let newly_deleted = dv_deltas + .get(remove.path.as_str()) + .map(|(newly_deleted, _)| newly_deleted); + self.process_follow_action( action, + &remove.path, + newly_deleted, + false, table, &used_columns, input_stream, @@ -2767,9 +2823,15 @@ impl DeltaTableInputEndpointInner { } for action in actions { - if matches!(action, Action::Add(_)) { - self.process_action( + if let Action::Add(add) = action { + let restored = dv_deltas + .get(add.path.as_str()) + .map(|(_, restored)| restored); + self.process_follow_action( action, + &add.path, + restored, + true, table, &used_columns, input_stream, @@ -3095,60 +3157,77 @@ impl DeltaTableInputEndpointInner { }) } - /// Build a [`TableProvider`] over the data file at `path` (relative and - /// URL-encoded, as the Delta log stores it) with the rows flagged by - /// deletion vector `dv` masked out. + /// Build a streaming [`TableProvider`] over the data file at `path` + /// (relative and URL-encoded, as the Delta log stores it) that reads the + /// rows `mode` picks out of `bitmap`: the file's live rows when applying a + /// deletion vector ([`ReadMode::NotInBitmap`]), or a deletion-vector delta + /// when following a same-path rewrite ([`ReadMode::InBitmap`]). The caller + /// supplies the decoded bitmap: a deletion vector for the mask, or the + /// difference of two deletion vectors for the delta. /// /// `keep_column` picks the columns to read: the provider declares the /// snapshot schema restricted to them, which doubles as the Parquet - /// projection (see [`masked_parquet_table`]). It must match the columns the - /// unmasked side keeps, so the two providers mix in one query. It is a - /// predicate rather than a fixed list because the callers differ: follow - /// keeps its `used_columns`, CDC keeps `keeps_cdc_column`. - async fn masked_file_provider( + /// projection (see [`filtered_parquet_table`]). It must match the columns + /// any file it unions with keeps, so the providers line up in one query. It + /// is a predicate rather than a fixed list because the callers differ: + /// follow keeps its `used_columns`, CDC keeps `keeps_cdc_column`. + async fn file_provider( &self, table: &DeltaTable, path: &str, - dv: &DeletionVectorDescriptor, + bitmap: RoaringTreemap, + mode: ReadMode, keep_column: impl Fn(&str) -> bool, - description: &str, ) -> AnyResult> { - // Decode the DV into its bitmap of deleted row positions, retrying on - // transient object-store failures (the decode is idempotent). - let bitmap = self - .retry( - &format!( - "decoding deletion vector for {description} at table version {:?}", - table.version(), - ), - None, - || read_deletion_vector(dv, table), - ) - .await?; - // The provider declares the kept columns under their physical names so - // it lines up with the unmasked side; the caller renames them back. + // files read into one query line up by column; the caller renames them + // back to logical. let read_schema = self.physical_read_schema(keep_column)?; - // `Add.path` is URL-encoded per the Delta spec; decode it into a - // real object-store key. + // Decode the URL-encoded Delta log path (per the spec) into an + // object-store key. let file_path = Path::from_url_path(path) .map_err(|e| anyhow!("invalid file path '{path}' in Delta log action: {e}"))?; - masked_parquet_table( + filtered_parquet_table( table.log_store().object_store(None), file_path, bitmap, read_schema, + mode, ) .await } + /// Decode a deletion vector into its bitmap of row positions, or the empty + /// set when there is no active DV. Retries transient object-store failures. + async fn decode_dv( + &self, + table: &DeltaTable, + dv: Option<&DeletionVectorDescriptor>, + description: &str, + ) -> AnyResult { + match dv.filter(|d| is_active_dv(d)) { + None => Ok(RoaringTreemap::new()), + Some(dv) => { + self.retry( + &format!( + "decoding deletion vector for {description} at table version {:?}", + table.version(), + ), + None, + || read_deletion_vector(dv, table), + ) + .await + } + } + } + /// Build the [`DataFrame`] for one side (adds or removes) of a CDC /// transaction, or `None` when the side has no files. /// /// Each file is `(path, deletion_vector)`. Files with an active DV stream - /// through a [`masked_parquet_table`] that drops their deleted rows; the + /// through a [`filtered_parquet_table`] that drops their deleted rows; the /// rest are read together through one [`ListingTable`]. The pieces combine /// with `UNION ALL`, each restricted to the same CDC read set (the columns /// [`Self::project_cdc_columns`] keeps), so they line up by position for the @@ -3201,14 +3280,11 @@ impl DeltaTableInputEndpointInner { // Keep the same CDC read set as `project_cdc_columns` applies on the // plain side (via `keeps_cdc_column`), so the two providers union // cleanly. + let bitmap = self.decode_dv(table, Some(dv), description).await?; let provider = self - .masked_file_provider( - table, - path, - dv, - |name| self.keeps_cdc_column(name), - description, - ) + .file_provider(table, path, bitmap, ReadMode::NotInBitmap, |name| { + self.keeps_cdc_column(name) + }) .await?; let df = self.datafusion.read_table(provider).map_err(|e| { anyhow!("internal error processing {description}; {REPORT_ERROR}; error reading masked file '{path}': {e}") @@ -3292,19 +3368,16 @@ impl DeltaTableInputEndpointInner { ) -> AnyResult<()> { let description = format!("file '{path}'"); - // An active deletion vector routes the file through a masked - // streaming provider, restricted to `used_columns` so unread columns - // are never decoded; otherwise the regular `ListingTable` path + // An active deletion vector routes the file through a streaming provider + // that masks the deleted rows, restricted to `used_columns` so unread + // columns are never decoded; otherwise the regular `ListingTable` path // applies. let provider: Arc = if let Some(dv) = deletion_vector.filter(|d| is_active_dv(d)) { - self.masked_file_provider( - table, - path, - dv, - |name| used_columns.contains(&name), - &description, - ) + let bitmap = self.decode_dv(table, Some(dv), &description).await?; + self.file_provider(table, path, bitmap, ReadMode::NotInBitmap, |name| { + used_columns.contains(&name) + }) .await? } else { // Address files via the table's real `root_url()` (e.g. `file:///...` @@ -3317,6 +3390,90 @@ impl DeltaTableInputEndpointInner { ) }; + self.emit_provider( + provider, + polarity, + used_columns, + &description, + table, + input_stream, + receiver, + start_transaction, + ) + .await + } + + /// Ingest one follow-mode `Add`/`Remove` action with `polarity` (`true` + /// inserts, `false` deletes), reading only the rows that changed when the + /// action is one side of a same-path deletion-vector rewrite. + #[allow(clippy::too_many_arguments)] + async fn process_follow_action( + &self, + action: &Action, + path: &str, + dv_delta: Option<&RoaringTreemap>, + polarity: bool, + table: &DeltaTable, + used_columns: &[&str], + input_stream: &mut dyn ArrowStream, + receiver: &mut Receiver, + start_transaction: Option>, + ) -> AnyResult<()> { + match dv_delta { + // One side of a same-path rewrite: read only its delta rows. An + // empty delta means the rewrite left this side unchanged, so there + // is nothing to emit. + Some(positions) if !positions.is_empty() => { + let description = format!("file '{path}'"); + let provider = self + .file_provider(table, path, positions.clone(), ReadMode::InBitmap, |name| { + used_columns.contains(&name) + }) + .await?; + self.emit_provider( + provider, + polarity, + used_columns, + &description, + table, + input_stream, + receiver, + start_transaction, + ) + .await + } + Some(_) => Ok(()), + // An unpaired action: a whole file enters or leaves the table. Read + // it whole, applying the action's own deletion vector. + None => { + self.process_action( + action, + table, + used_columns, + input_stream, + receiver, + start_transaction, + ) + .await + } + } + } + + /// Read `provider` restricted to `used_columns`, apply the connector's + /// optional `filter` expression (`self.config.filter`), and push the rows to + /// the input stream with `polarity`. + #[allow(clippy::too_many_arguments)] + async fn emit_provider( + &self, + provider: Arc, + polarity: bool, + used_columns: &[&str], + description: &str, + table: &DeltaTable, + input_stream: &mut dyn ArrowStream, + receiver: &mut Receiver, + start_transaction: Option>, + ) -> AnyResult<()> { let df = self.datafusion.read_table(provider).map_err(|e| { anyhow!("internal error processing {description}; {REPORT_ERROR}; error reading Parquet file: {e}") })?; @@ -3343,7 +3500,7 @@ impl DeltaTableInputEndpointInner { df, polarity, None, - &description, + description, input_stream, receiver, start_transaction, diff --git a/python/tests/platform/test_delta_input_deletion_vectors.py b/python/tests/platform/test_delta_input_deletion_vectors.py index 6138c15e11a..2cc2f0281e2 100644 --- a/python/tests/platform/test_delta_input_deletion_vectors.py +++ b/python/tests/platform/test_delta_input_deletion_vectors.py @@ -19,10 +19,15 @@ import json from pathlib import Path +from typing import NamedTuple from feldera import PipelineBuilder from feldera.runtime_config import RuntimeConfig -from feldera.testutils import FELDERA_TEST_NUM_HOSTS, FELDERA_TEST_NUM_WORKERS +from feldera.testutils import ( + FELDERA_TEST_NUM_HOSTS, + FELDERA_TEST_NUM_WORKERS, + number_of_input_records, +) from tests import TEST_CLIENT from tests.utils import DeltaTestLocation, ensure_delta_spark_fixture @@ -32,6 +37,22 @@ CONNECTOR = "dv_in" TOTAL_ROWS = 200 EXPECTED_ROWS_AFTER_DV = 100 +# Rows the even-id soft-delete masks (== the odd-id survivors, both halves equal). +DV_DELETED_ROWS = TOTAL_ROWS - EXPECTED_ROWS_AFTER_DV +# Records a follow-mode DV rewrite must feed into the circuit. A same-path +# `add`/`remove` pair changes only the rows whose deletion-vector membership +# flipped, so the connector must ingest just those, not re-read the whole file. +# The follow phase reads one DV commit; snapshot reads the pinned base version. +# +# plain fixture (v0 base, v1 DV-delete): snapshot TOTAL_ROWS + follow deletes +# exactly the DV_DELETED_ROWS. +# cdc fixture (v1 base, v2 restore) : snapshot EXPECTED_ROWS_AFTER_DV + +# follow re-inserts the DV_DELETED_ROWS. +# +# Before the same-path pairing fix, follow re-read whole files (~2N-K records), +# so these totals were ~500 and ~400 -- the amplification these bounds catch. +FOLLOW_DV_DELETE_INPUT_RECORDS = TOTAL_ROWS + DV_DELETED_ROWS +FOLLOW_RESTORE_INPUT_RECORDS = EXPECTED_ROWS_AFTER_DV + DV_DELETED_ROWS # Bump to invalidate cached MinIO copies when the fixture definition changes. FIXTURE_VERSION = "dv_snapshot_v1" # CDC-shaped fixture (see fixtures/deletion_vectors.py --cdc): v0 inserts @@ -99,14 +120,28 @@ def _build_sql( ) -def _run_to_completion(pipeline_name: str, sql: str) -> tuple[int, int]: +class DvIngest(NamedTuple): + """Outcome of ingesting a DV fixture. + + ``total`` and ``even_id_rows`` are TABLE row counts; ``input_records`` is + ``total_input_records`` (records fed into the circuit). ``input_records`` + exposes read amplification the row counts cannot: the circuit nets a whole- + file re-read back to the same result, so only the ingested-record count + distinguishes a minimal DV-delta read from re-reading the whole file. + """ + + total: int + even_id_rows: int + input_records: int + + +def _run_to_completion(pipeline_name: str, sql: str) -> DvIngest: """Run the pipeline until end-of-input. - Returns ``(total_rows, even_id_rows)`` of TABLE. ``even_id_rows`` exists - because every fixture soft-deletes exactly the even ids: a correct ingest - leaves zero of them, and counting them catches an *inverted* deletion - vector (ingesting the deleted half), which COUNT(*) alone cannot, since - both halves have the same size. + ``even_id_rows`` exists because every fixture soft-deletes exactly the even + ids: a correct ingest leaves zero of them, and counting them catches an + *inverted* deletion vector (ingesting the deleted half), which COUNT(*) + alone cannot, since both halves have the same size. """ pipeline = PipelineBuilder( TEST_CLIENT, @@ -128,8 +163,10 @@ def _run_to_completion(pipeline_name: str, sql: str) -> tuple[int, int]: f" FROM {TABLE}" ) ) + # Read before stopping: the metric is only live while the pipeline runs. + input_records = number_of_input_records(pipeline) pipeline.stop(force=True) - return int(rows[0]["total"]), int(rows[0]["even_id_rows"]) + return DvIngest(int(rows[0]["total"]), int(rows[0]["even_id_rows"]), input_records) def _ingest_dv_fixture( @@ -142,12 +179,11 @@ def _ingest_dv_fixture( overwrite: bool = False, columns: str = "id INT NOT NULL, name VARCHAR, value DOUBLE", extra_config: dict | None = None, -) -> tuple[int, int]: - """Ensure the fixture exists, ingest it in `mode`, return the row counts. +) -> DvIngest: + """Ensure the fixture exists, ingest it in `mode`, return the outcome. Owns the location's lifecycle (create/cleanup); the tests reduce to a - call plus their assertions. Returns ``_run_to_completion``'s - ``(total_rows, even_id_rows)`` pair. + call plus their assertions. Returns ``_run_to_completion``'s [`DvIngest`]. """ builder_flags = (["--cdc"] if cdc else []) + (["--overwrite"] if overwrite else []) loc = DeltaTestLocation.create( @@ -172,12 +208,12 @@ def _ingest_dv_fixture( def test_delta_input_snapshot_with_deletion_vectors(pipeline_name): """Snapshot read of a DV-enabled table must skip soft-deleted rows.""" - total, even_id_rows = _ingest_dv_fixture(pipeline_name, mode="snapshot") - assert total == EXPECTED_ROWS_AFTER_DV, ( + result = _ingest_dv_fixture(pipeline_name, mode="snapshot") + assert result.total == EXPECTED_ROWS_AFTER_DV, ( "snapshot ingest of a DV-enabled table must drop the soft-deleted " - f"rows ({TOTAL_ROWS - EXPECTED_ROWS_AFTER_DV} rows have id % 2 = 0)" + f"rows ({DV_DELETED_ROWS} rows have id % 2 = 0)" ) - assert even_id_rows == 0, ( + assert result.even_id_rows == 0, ( "the surviving rows must be the odd ids, not the deleted even ids" ) @@ -185,39 +221,49 @@ def test_delta_input_snapshot_with_deletion_vectors(pipeline_name): def test_delta_input_follow_with_deletion_vectors(pipeline_name): """The follow path must apply the deletion vectors carried by log actions. - Replays the fixture's DV commit (v1) from version 0: the `remove` - retracts all rows of the rewritten file and the `add` re-inserts only - the DV survivors. + Replays the fixture's DV commit (v1) from version 0. The commit is a + same-path `remove`/`add` pair (the file is immutable; only its deletion + vector changed), so the connector reads just the newly-masked rows and + retracts them, leaving the DV survivors. It must not re-read the whole file. """ - total, even_id_rows = _ingest_dv_fixture( + result = _ingest_dv_fixture( pipeline_name, mode="snapshot_and_follow", extra_config={"version": 0, "end_version": 1}, ) - assert total == EXPECTED_ROWS_AFTER_DV, ( + assert result.total == EXPECTED_ROWS_AFTER_DV, ( "follow ingest of a DV commit must retract the soft-deleted rows; " f"{TOTAL_ROWS} rows means the add action's deletion vector was ignored" ) - assert even_id_rows == 0, ( + assert result.even_id_rows == 0, ( "the surviving rows must be the odd ids, not the deleted even ids" ) + # A same-path add/remove DV rewrite must ingest only the flipped rows. + # Re-reading the whole file (the pre-fix behavior) would net the same rows + # but roughly double this count, so the record total is what guards it. + assert result.input_records == FOLLOW_DV_DELETE_INPUT_RECORDS, ( + f"follow ingested {result.input_records} records for a " + f"{DV_DELETED_ROWS}-row DV delete; expected " + f"{FOLLOW_DV_DELETE_INPUT_RECORDS} (snapshot {TOTAL_ROWS} + " + f"{DV_DELETED_ROWS} deleted). A far larger count means the connector " + "re-read the whole rewritten file instead of just the DV delta" + ) def test_delta_input_follow_restore_with_deletion_vectors(pipeline_name): """Follow mode must apply a deletion vector carried by a `remove` action. Reuses the CDC fixture and replays its restore commit (v2) in follow mode, - starting from the post-delete snapshot (v1). Unlike CDC mode, follow mode - does not skip the same-path `remove`/`add` pair; it processes each action - on its own. The restore's `remove` carries the delete's deletion vector, so - masking it retracts only the surviving odd ids; the `add` (DV-free again) - re-inserts the whole file, bringing the soft-deleted even ids back. - - It drives the masked reader from a `remove` action whose deletion vector is - non-empty (the v0->v1 follow test's `remove` has no DV, as the file was - unmasked at v0). + starting from the post-delete snapshot (v1). The restore is a same-path + `remove` (carrying the delete's deletion vector) and `add` (DV cleared): the + connector pairs them and re-inserts exactly the rows the DV un-masked, the + soft-deleted even ids. It must not re-read the whole restored file. + + This exercises the restore direction of the pair, where the flipped rows come + from the `remove` side's non-empty deletion vector (the v0->v1 follow test's + `remove` has no DV, as the file was unmasked at v0). """ - total, even_id_rows = _ingest_dv_fixture( + result = _ingest_dv_fixture( pipeline_name, mode="snapshot_and_follow", fixture_version=CDC_FIXTURE_VERSION, @@ -226,15 +272,25 @@ def test_delta_input_follow_restore_with_deletion_vectors(pipeline_name): columns="id BIGINT NOT NULL", extra_config={"version": 1, "end_version": 2}, ) - assert total == TOTAL_ROWS, ( + assert result.total == TOTAL_ROWS, ( "the restore's `add` must re-insert the whole file after its `remove` " "(masked by the delete's deletion vector) retracts only the survivors; " - f"got {total} rows, expected {TOTAL_ROWS}" + f"got {result.total} rows, expected {TOTAL_ROWS}" ) - assert even_id_rows == TOTAL_ROWS // 2, ( + assert result.even_id_rows == TOTAL_ROWS // 2, ( "the restored even ids must be present; their absence means the " "`remove` action's deletion vector was applied to the wrong rows" ) + # The restore flips the same rows back: its `remove` (masked to the odd + # survivors) and `add` (DV cleared) differ only in the even ids, so follow + # must re-insert exactly those, not re-read the whole restored file. + assert result.input_records == FOLLOW_RESTORE_INPUT_RECORDS, ( + f"follow ingested {result.input_records} records for a " + f"{DV_DELETED_ROWS}-row restore; expected " + f"{FOLLOW_RESTORE_INPUT_RECORDS} (snapshot {EXPECTED_ROWS_AFTER_DV} + " + f"{DV_DELETED_ROWS} restored). A far larger count means the connector " + "re-read the whole file on each side of the same-path rewrite" + ) def test_delta_input_cdc_overwrite_masks_removed_dv_rows(pipeline_name): @@ -250,7 +306,7 @@ def test_delta_input_cdc_overwrite_masks_removed_dv_rows(pipeline_name): the whole file (including its DV-dead even rows) and wrongly cancel the new even events, leaving zero. """ - total, even_id_rows = _ingest_dv_fixture( + result = _ingest_dv_fixture( pipeline_name, mode="cdc", fixture_version=OVERWRITE_FIXTURE_VERSION, @@ -264,12 +320,12 @@ def test_delta_input_cdc_overwrite_masks_removed_dv_rows(pipeline_name): "cdc_order_by": "__feldera_ts asc, lsn asc", }, ) - assert total == EXPECTED_ROWS_AFTER_DV, ( + assert result.total == EXPECTED_ROWS_AFTER_DV, ( "the re-inserted even events must survive the `EXCEPT ALL`; a count of " "0 means the `remove`'s deletion vector was ignored and its DV-dead " "rows over-subtracted the new events" ) - assert even_id_rows == EXPECTED_ROWS_AFTER_DV, ( + assert result.even_id_rows == EXPECTED_ROWS_AFTER_DV, ( "every re-inserted event has an even id" ) @@ -283,7 +339,7 @@ def test_delta_input_cdc_with_deletion_vectors(pipeline_name): are both such commits; replaying from version 0 must yield only the single v3 insert. """ - total, _ = _ingest_dv_fixture( + result = _ingest_dv_fixture( pipeline_name, mode="cdc", fixture_version=CDC_FIXTURE_VERSION, @@ -297,7 +353,7 @@ def test_delta_input_cdc_with_deletion_vectors(pipeline_name): "cdc_order_by": "__feldera_ts asc, lsn asc", }, ) - assert total == 1, ( + assert result.total == 1, ( "CDC replay of DV rewrite commits must net to zero events: " "only the single v3 insert event may arrive; a higher count " "means soft-deleted or restored events were re-ingested"