You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by sd...@apache.org on 2025/01/15 06:59:05 UTC
(iceberg-rust) branch feature/arrow-reader-delete-file-retrieval created (now 50d552a0)
This is an automated email from the ASF dual-hosted git repository.
sdd pushed a change to branch feature/arrow-reader-delete-file-retrieval
in repository https://gitbox.apache.org/repos/asf/iceberg-rust.git
at 50d552a0 feat(wip): arrow reader delete file read
This branch includes the following new commits:
new 50d552a0 feat(wip): arrow reader delete file read
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iceberg-rust) 01/01: feat(wip): arrow reader delete file read
Posted by sd...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sdd pushed a commit to branch feature/arrow-reader-delete-file-retrieval
in repository https://gitbox.apache.org/repos/asf/iceberg-rust.git
commit 50d552a03c78615d7e8ba5190c2b566f6b487b03
Author: Scott Donnelly <sc...@donnel.ly>
AuthorDate: Wed Jan 15 06:58:40 2025 +0000
feat(wip): arrow reader delete file read
---
crates/iceberg/src/arrow/reader.rs | 84 ++++++++++++++++++++++++++++++++++++--
crates/iceberg/src/deletes.rs | 25 ++++++++++++
crates/iceberg/src/lib.rs | 1 +
crates/iceberg/src/scan.rs | 13 ++++++
4 files changed, 120 insertions(+), 3 deletions(-)
diff --git a/crates/iceberg/src/arrow/reader.rs b/crates/iceberg/src/arrow/reader.rs
index b4e15821..91850c80 100644
--- a/crates/iceberg/src/arrow/reader.rs
+++ b/crates/iceberg/src/arrow/reader.rs
@@ -32,7 +32,8 @@ use arrow_string::like::starts_with;
use bytes::Bytes;
use fnv::FnvHashSet;
use futures::future::BoxFuture;
-use futures::{try_join, FutureExt, StreamExt, TryFutureExt, TryStreamExt};
+use futures::{try_join, FutureExt, StreamExt, TryFutureExt, TryStreamExt, SinkExt};
+use itertools::Itertools;
use parquet::arrow::arrow_reader::{ArrowPredicateFn, ArrowReaderOptions, RowFilter, RowSelection};
use parquet::arrow::async_reader::AsyncFileReader;
use parquet::arrow::{ParquetRecordBatchStreamBuilder, ProjectionMask, PARQUET_FIELD_ID_META_KEY};
@@ -47,10 +48,12 @@ use crate::expr::visitors::page_index_evaluator::PageIndexEvaluator;
use crate::expr::visitors::row_group_metrics_evaluator::RowGroupMetricsEvaluator;
use crate::expr::{BoundPredicate, BoundReference};
use crate::io::{FileIO, FileMetadata, FileRead};
-use crate::scan::{ArrowRecordBatchStream, FileScanTask, FileScanTaskStream};
-use crate::spec::{Datum, PrimitiveType, Schema};
+use crate::scan::{ArrowRecordBatchStream, FileScanTask, FileScanTaskDeleteFile, FileScanTaskStream};
+use crate::spec::{DataContentType, Datum, PrimitiveType, Schema};
use crate::utils::available_parallelism;
use crate::{Error, ErrorKind};
+use roaring::RoaringBitmap;
+use crate::deletes::Deletes;
/// Builder to create ArrowReader
pub struct ArrowReaderBuilder {
@@ -257,6 +260,81 @@ impl ArrowReader {
Ok(Box::pin(record_batch_stream) as ArrowRecordBatchStream)
}
+ // retrieve all delete files concurrently from FileIO and parse them
+ // into `Deletes` objects
+ async fn get_deletes(
+ delete_file_entries: Vec<FileScanTaskDeleteFile>,
+ file_io: FileIO,
+ concurrency_limit_data_files: usize,
+ ) -> Result<Vec<Deletes>> {
+ if delete_file_entries.is_empty() {
+ return Ok(vec![]);
+ }
+
+ futures::stream::iter(delete_file_entries.into_iter().map(Ok))
+ .map_ok(|entry| {
+ let file_io = file_io.clone();
+ async move {
+ let FileScanTaskDeleteFile {
+ file_path,
+ file_type,
+ ..
+ } = entry;
+
+ let record_batch_stream = Self::create_parquet_record_batch_stream_builder(
+ &file_path,
+ file_io,
+ false,
+ )
+ .await?
+ .build()?
+ .map_err(|err| Error::new(ErrorKind::DataInvalid, err.to_string()).with_source(err))
+ .boxed();
+
+ match file_type {
+ DataContentType::PositionDeletes => Self::parse_positional_delete_file(record_batch_stream).await,
+ DataContentType::EqualityDeletes => Self::parse_equality_delete_file(record_batch_stream).await,
+ _ => Err(Error::new(
+ ErrorKind::Unexpected,
+ "Expected equality or positional delete",
+ )),
+ }
+ }
+ })
+ .try_buffer_unordered(concurrency_limit_data_files)
+ .try_collect()
+ .await
+ }
+
+ async fn parse_positional_delete_file(_record_batches: ArrowRecordBatchStream) -> Result<Deletes> {
+ todo!()
+ }
+
+ async fn parse_equality_delete_file(_record_batches: ArrowRecordBatchStream) -> Result<Deletes> {
+ todo!()
+ }
+
+ async fn create_parquet_record_batch_stream_builder(
+ data_file_path: &str,
+ file_io: FileIO,
+ should_load_page_index: bool,
+ ) -> Result<ParquetRecordBatchStreamBuilder<ArrowFileReader<impl FileRead + Sized>>> {
+ // Get the metadata for the Parquet file we need to read and build
+ // a reader for the data within
+ let parquet_file = file_io.new_input(data_file_path)?;
+ let (parquet_metadata, parquet_reader) =
+ try_join!(parquet_file.metadata(), parquet_file.reader())?;
+ let parquet_file_reader = ArrowFileReader::new(parquet_metadata, parquet_reader);
+
+ // Start creating the record batch stream, which wraps the parquet file reader
+ let record_batch_stream_builder = ParquetRecordBatchStreamBuilder::new_with_options(
+ parquet_file_reader,
+ ArrowReaderOptions::new().with_page_index(should_load_page_index),
+ )
+ .await?;
+ Ok(record_batch_stream_builder)
+ }
+
fn build_field_id_set_and_map(
parquet_schema: &SchemaDescriptor,
predicate: &BoundPredicate,
diff --git a/crates/iceberg/src/deletes.rs b/crates/iceberg/src/deletes.rs
new file mode 100644
index 00000000..268a2131
--- /dev/null
+++ b/crates/iceberg/src/deletes.rs
@@ -0,0 +1,25 @@
+use roaring::RoaringBitmap;
+use arrow_array::RecordBatch;
+
+// Represents a parsed Delete file that can be safely stored
+// in the Object Cache.
+pub(crate) enum Deletes {
+ // Positional delete files are parsed into a map of
+ // filename to a sorted list of row indices.
+
+ // TODO: Ignoring the stored rows that are present in
+ // positional deletes for now. I think they only used for statistics?
+ Vector(RoaringBitmap),
+
+ // Equality delete files are initially parsed solely as an
+ // unprocessed list of `RecordBatch`es from the equality
+ // delete files.
+ // I don't think we can do better than this by
+ // storing a Predicate (because the equality deletes use the
+ // field_id rather than the field name, so if we keep this as
+ // a Predicate then a field name change would break it).
+ // Similarly, I don't think we can store this as a BoundPredicate
+ // as the column order could be different across different data
+ // files and so the accessor in the bound predicate could be invalid).
+ Equality(Vec<RecordBatch>),
+}
diff --git a/crates/iceberg/src/lib.rs b/crates/iceberg/src/lib.rs
index eaecfea6..af4d9448 100644
--- a/crates/iceberg/src/lib.rs
+++ b/crates/iceberg/src/lib.rs
@@ -86,3 +86,4 @@ mod utils;
pub mod writer;
mod puffin;
+mod deletes;
diff --git a/crates/iceberg/src/scan.rs b/crates/iceberg/src/scan.rs
index 7a100b34..4e7a380c 100644
--- a/crates/iceberg/src/scan.rs
+++ b/crates/iceberg/src/scan.rs
@@ -933,6 +933,19 @@ pub struct FileScanTask {
pub predicate: Option<BoundPredicate>,
}
+/// A task to scan part of file.
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
+pub struct FileScanTaskDeleteFile {
+ /// The delete file path
+ pub file_path: String,
+
+ /// delete file type
+ pub file_type: DataContentType,
+
+ /// partition id
+ pub partition_spec_id: i32,
+}
+
impl FileScanTask {
/// Returns the data file path of this file scan task.
pub fn data_file_path(&self) -> &str {