feat: implementation for ArrowReader::build_deletes_row_selection

Author: sdd

crates/iceberg/src/arrow/reader.rs

@@ -33,7 +33,9 @@ use bytes::Bytes;
 use fnv::FnvHashSet;
 use futures::future::BoxFuture;
 use futures::{try_join, FutureExt, StreamExt, TryFutureExt, TryStreamExt};
-use parquet::arrow::arrow_reader::{ArrowPredicateFn, ArrowReaderOptions, RowFilter, RowSelection};
+use parquet::arrow::arrow_reader::{
+    ArrowPredicateFn, ArrowReaderOptions, RowFilter, RowSelection, RowSelector,
+};
 use parquet::arrow::async_reader::AsyncFileReader;
 use parquet::arrow::{ParquetRecordBatchStreamBuilder, ProjectionMask, PARQUET_FIELD_ID_META_KEY};
 use parquet::file::metadata::{ParquetMetaData, ParquetMetaDataReader};
@@ -340,15 +342,82 @@ impl ArrowReader {
     /// Using the Parquet page index, we build a `RowSelection` that rejects rows that are indicated
     /// as having been deleted by a positional delete, taking into account any row groups that have
     /// been skipped entirely by the filter predicate
-    #[allow(unused)]
     fn build_deletes_row_selection(
         parquet_metadata: &Arc<ParquetMetaData>,
         selected_row_groups: &Option<Vec<usize>>,
         positional_deletes: &[usize],
     ) -> Result<RowSelection> {
-        // TODO
+        let mut selected_row_groups_idx = 0;
+        let mut curr_pos_del_idx = 0;
+        let pos_del_len = positional_deletes.len();
+
+        let page_index = parquet_metadata.row_groups().iter().enumerate();
+
+        let mut results: Vec<RowSelector> = Vec::new();
+        let mut current_page_base_idx: usize = 0;
+        for (idx, row_group_metadata) in page_index {
+            let page_num_rows = row_group_metadata.num_rows() as usize;
+            let next_page_base_idx = current_page_base_idx + page_num_rows;
+
+            // skip any row groups that aren't in the row group selection
+            if let Some(selected_row_groups) = selected_row_groups {
+                // skip row groups that aren't present in selected_row_groups
+                if selected_row_groups_idx == selected_row_groups.len() {
+                    break;
+                }
+                if idx == selected_row_groups[selected_row_groups_idx] {
+                    selected_row_groups_idx += 1;
+                } else {
+                    current_page_base_idx += page_num_rows;
+                    continue;
+                }
+            }
+
+            let mut next_deleted_row_idx = positional_deletes[curr_pos_del_idx];
+
+            // if the index of the next deleted row is beyond this page, skip
+            // to the next page
+            if next_deleted_row_idx >= next_page_base_idx {
+                continue;
+            }
+
+            let mut current_idx = current_page_base_idx;
+            while next_deleted_row_idx < next_page_base_idx {
+                // select all rows that precede the next delete index
+                if current_idx < next_deleted_row_idx {
+                    let run_length = next_deleted_row_idx - current_idx;
+                    results.push(RowSelector::select(run_length));
+                    current_idx += run_length;
+                }
+
+                // skip all consecutive deleted rows
+                let mut run_length = 1;
+                while curr_pos_del_idx < pos_del_len - 1
+                    && positional_deletes[curr_pos_del_idx + 1] == next_deleted_row_idx + 1
+                {
+                    run_length += 1;
+                    curr_pos_del_idx += 1;
+                }
+                results.push(RowSelector::skip(run_length));
+                current_idx += run_length;
+
+                curr_pos_del_idx += 1;
+                if curr_pos_del_idx >= pos_del_len {
+                    break;
+                }
+                next_deleted_row_idx = positional_deletes[curr_pos_del_idx];
+            }
+
+            if let Some(selected_row_groups) = selected_row_groups {
+                if selected_row_groups_idx == selected_row_groups.len() {
+                    break;
+                }
+            }
+
+            current_page_base_idx += page_num_rows;
+        }
 
-        Ok(RowSelection::default())
+        Ok(results.into())
     }
 
     fn build_field_id_set_and_map(