hcrosse opened a new issue, #3613: URL: https://github.com/apache/iceberg-rust/issues/3613
### Is your feature request related to a problem or challenge? The reader checks a predicate on every row even when statistics already prove that every row matches. For a time range covering whole files, skipping those checks cut read time by 17% in a prototype. At main [`896f9d6c`](https://github.com/apache/iceberg-rust/commit/896f9d6c50ff52fa9e43aec170d3f14bfb40b512), `ArrowReader` [installs a Parquet `RowFilter`](https://github.com/apache/iceberg-rust/blob/896f9d6c50ff52fa9e43aec170d3f14bfb40b512/crates/iceberg/src/arrow/reader/pipeline.rs#L540-L577) whenever a task has a predicate. Row-group pruning already uses Parquet statistics to reject groups that cannot match, but groups that must match still pass through the per-row filter. I measured reads of synthetic data from local disk with one reader thread in a Linux amd64 VM: 8 AMD EPYC vCPUs, 16 GB RAM, rustc 1.97.0 and dependency versions matching upstream's `Cargo.lock`. The prototype decides per file and skips evaluation only when every selected row group is proven to match. Comparing it with the default reader in the same build gave these median times to read and count rows over five fresh processes: | Predicate | Default | Prototype | Change | | --- | ---: | ---: | ---: | | `all`: time range covering every row | 2.94 s | 2.45 s | -17% | | `window`: time range covering parts of two files and nine whole files | 0.98 s | 0.86 s | -13% | | `most`: `bytes < 666667`, about 68% match | 3.87 s | 3.77 s | within noise | | `sel`: `user_name = 'user-0042'`, about 0.5% match | 1.88 s | 1.84 s | within noise | The `all` difference is about 16 ns per row: `(2.94 - 2.45) s / 30,000,000`. Row counts were identical in each pair. From the generated timestamp ranges, the prototype would keep the filter in the two boundary files and skip it in the nine inner files of `window`. I did not instrument those decisions. On this workload, a proof from whole-file metrics would give the same result. No median increased in these four comparisons, but I treat differences under about 10% as noise. On stock main, I measured `all` at 2.95 s with the reader's filter against 2.67 s with the predicate moved to the consumer, which still filters every row. That corroborating gap is at the edge of noise. Earlier runs put the overhead at roughly 13% and 21%, though the 21% run used different dependency versions. For `most`, the reader's filter took 4.01 s against 3.10 s with consumer filtering. Statistics cannot prove that predicate, so this issue does not address it. Keeping the filter matters for `sel`: it took 1.97 s against 2.57 s with consumer filtering. #### Reproduction This program writes one file of 10 million sorted longs and reads it through `ArrowReader` twice: once with a predicate that every row satisfies, and once without a predicate. The row-group statistics prove that predicate, because every group's min and max lie inside the range. Run it as `src/main.rs` of a Rust 2024 crate with `cargo run --release`. I used Rust 1.97.0 on linux/amd64 (8 vCPUs). The pinned `arrow-array` and `parquet` versions match [upstream's `Cargo.lock`](https://github.com/apache/iceberg-rust/blob/896f9d6c50ff52fa9e43aec170d3f14bfb40b512/Cargo.lock) at the tested revision. ```toml [dependencies] iceberg = { git = "https://github.com/apache/iceberg-rust";, rev = "896f9d6c50ff52fa9e43aec170d3f14bfb40b512" } arrow-array = "=59.3.2" futures = "0.3" parquet = { version = "=59.3.0", default-features = false, features = ["arrow"] } tokio = { version = "1", features = ["macros", "rt"] } ``` ```rust use std::{fs::File, sync::Arc, time::Instant}; use arrow_array::{Int64Array, RecordBatch}; use futures::{StreamExt, TryStreamExt, stream}; use iceberg::arrow::{ArrowReaderBuilder, schema_to_arrow_schema}; use iceberg::expr::{Bind, Reference}; use iceberg::spec::{DataFileFormat, Datum, NestedField, PrimitiveType, Schema as IcebergSchema}; use iceberg::{Runtime, io::FileIO, scan::FileScanTask}; use parquet::arrow::ArrowWriter; const ROWS: i64 = 10_000_000; const REPETITIONS: usize = 7; /// Reads the task `REPETITIONS` times and returns its row count and median time in seconds. async fn read(task: FileScanTask) -> Result<(usize, f64), Box<dyn std::error::Error>> { let mut rows = 0; let mut times = Vec::new(); for _ in 0..REPETITIONS { let reader = ArrowReaderBuilder::new(FileIO::new_with_fs(), Runtime::current()).build(); let start = Instant::now(); let tasks = stream::iter([Ok(task.clone())]).boxed(); let batches: Vec<RecordBatch> = reader.read(tasks)?.stream().try_collect().await?; times.push(start.elapsed().as_secs_f64()); rows = batches.iter().map(RecordBatch::num_rows).sum(); } times.sort_by(f64::total_cmp); Ok((rows, times[REPETITIONS / 2])) } #[tokio::main(flavor = "current_thread")] async fn main() -> Result<(), Box<dyn std::error::Error>> { let path = std::env::temp_dir().join("sorted.parquet"); let fields = [NestedField::required(1, "ts", PrimitiveType::Long).into()]; let iceberg_schema = Arc::new(IcebergSchema::builder().with_fields(fields).build()?); // Adds the `PARQUET:field_id` metadata that ArrowReader uses to match columns. let arrow_schema = Arc::new(schema_to_arrow_schema(&iceberg_schema)?); let values = Int64Array::from_iter_values(0..ROWS); let batch = RecordBatch::try_new(arrow_schema.clone(), vec![Arc::new(values)])?; let mut writer = ArrowWriter::try_new(File::create(&path)?, arrow_schema, None)?; writer.write(&batch)?; writer.close()?; let covers_all_rows = Reference::new("ts") .greater_than_or_equal_to(Datum::long(0)) .and(Reference::new("ts").less_than_or_equal_to(Datum::long(ROWS - 1))) .bind(iceberg_schema.clone(), true)?; let size = std::fs::metadata(&path)?.len(); let task = |predicate| { FileScanTask::builder() .with_data_file_path(path.display().to_string()) .with_data_file_format(DataFileFormat::Parquet) .with_file_size_in_bytes(size) .with_start(0) .with_length(size) .with_schema(iceberg_schema.clone()) .with_project_field_ids(vec![1]) .with_predicate(predicate) .with_case_sensitive(true) .build() }; let (rows_with, time_with) = read(task(Some(covers_all_rows))?).await?; let (rows_without, time_without) = read(task(None)?).await?; println!("rows: {rows_with} with predicate, {rows_without} without"); println!("time: {time_with:.3} s with predicate, {time_without:.3} s without"); Ok(()) } ``` Output: ```text rows: 10000000 with predicate, 10000000 without time: 0.148 s with predicate, 0.030 s without ``` Both reads return every row. The row filter made this read about five times slower, roughly 12 ns per row. The file has a single column, so the filter is most of the work. On the wider table measured earlier, the cost was 16 ns per row, or 17% of read time. The read without a predicate also skips row-group pruning, but that is a metadata check over about 10 row groups. #### Measured workload 1. Write an unpartitioned synthetic Iceberg table with 30 files, one day and 1,000,000 rows per file. Use a required timestamp sorted within each file, five string columns, an integer and a long, ZSTD(1), and row groups of 131,072 rows. The table occupied about 585 MiB. 2. Plan files with `TableScan::plan_files`. Read all columns through `ArrowReaderBuilder` with batch size 8,192, data-file concurrency 1 and a current-thread Tokio runtime. Count rows and time reading separately from planning. 3. Test `ts >= start AND ts < end` for the full 30 days and for days 10.5 through 20.5. Also test the numeric and string predicates listed with the timings. Repeat each configuration in five fresh processes. 4. For the stock comparison, rebuild the same planned tasks without their predicate and apply the predicate in the consumer. This also removes reader-side pruning, so it is a baseline, not the proposed behavior. I measured one VM size, amd64 only. Object storage, partitioned tables, other writers, nullable columns and delete files at benchmark scale were not measured. ### Describe the solution you'd like Skip the per-row evaluation when statistics prove every row matches. Proof can come from file metrics at planning time or row-group statistics at read time. Keep results identical and enable this by default, without a new option. Leave the filter in place whenever the statistics are insufficient. Proof must agree with row filtering, including nulls and NaN. Iceberg's strict-metrics semantics count null as matching `!=` and `NOT IN`, while the row filter drops it. An earlier prototype therefore returned 12 rows for `name != 'a'` where the default reader returned 7, retaining five null rows. A missing null count must not be read as zero. Java already [uses strict metrics to prove that a whole file can be deleted](https://github.com/apache/iceberg/blob/866bca5d6291d19b3480bb55529c9eebec82d0d2/core/src/main/java/org/apache/iceberg/ManifestFilterManager.java#L527-L604). Missing fields, promoted types, truncated string bounds and deprecated min/max sort orders from old writers all need safe handling. https://github.com/apache/iceberg-rust/issues/3536 reports file bounds that exclude actual values when the writer ignores row groups with truncated statistics, so those bounds cannot be used as proof. Equality deletes, positional-delete selections and runtime predicates must remain enforced even when the scan predicate is proven. Related work: - https://github.com/apache/datafusion/pull/23696 implements this optimization in DataFusion, merged on 2026-08-25, closing https://github.com/apache/datafusion/issues/23067, with benchmarks in https://github.com/apache/datafusion/pull/24328. - https://github.com/apache/datafusion/issues/25253 and the closed proposal https://github.com/apache/datafusion/pull/25290 document that older parquet-rs writers omitted null counts of zero. Only those writers' missing counts can be read as zero. - https://github.com/apache/iceberg-rust/issues/3298 and https://github.com/apache/iceberg-rust/pull/3302 address missing columns through constant folding. Proof must use the folded predicate. - https://github.com/apache/iceberg-rust/issues/3355 covers comparisons on missing null columns, whose corrected semantics must be preserved. - https://github.com/apache/iceberg-rust/issues/3343 and https://github.com/apache/iceberg-rust/pull/3351 add runtime predicates, which must stay enforced independently of a static proof. - https://github.com/apache/iceberg-rust/pull/3540 corrects filtering on promoted columns. Proof must use the promoted values. - https://github.com/apache/iceberg-rust/pull/2961 replaces equality-delete predicate trees with hash probes, which must remain active when scan-predicate evaluation is skipped. ### Willingness to contribute I would be willing to contribute to this feature with guidance from the Iceberg Rust community -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
