This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-rs.git
The following commit(s) were added to refs/heads/main by this push:
new 01b1275aab test(parquet): differential tests of `ParquetPushDecoder`
against the sync reader (#11485)
01b1275aab is described below
commit 01b1275aab90c033da3f64f5edbfb4f061606cae
Author: Adrian Garcia Badaracco <[email protected]>
AuthorDate: Mon Oct 12 10:45:12 2026 +0000
test(parquet): differential tests of `ParquetPushDecoder` against the sync
reader (#11485)
# Which issue does this PR close?
- Part of https://github.com/apache/arrow-rs/issues/11234.
- Stack of PRs and review order:
https://github.com/apache/arrow-rs/issues/11234#issuecomment-6097375160
# Rationale for this change
`ParquetPushDecoder` has no test that compares it with the sync reader
across many scan configurations. This PR adds one. It has value on
`main`, and https://github.com/apache/arrow-rs/pull/11238 will use the
same harness to check its new mode. This PR is split out of that PR to
make it smaller.
# What changes are included in this PR?
Tests only: one new file and one new dev-dependency.
```text
Scan { batch_size, projection, row_groups, selection, limit, offset,
predicates, policy, .. }
├── push_decoder() ──▶ drive_file ──▶ rows ─┐
└── sync_batches() ─────────────────▶ rows ─┴─ assert_same_rows
```
- Two test files with list and struct columns: one with a new page each
25 rows or less, one where the page boundaries differ a lot between
columns.
- Two property tests with
[`proptest`](https://crates.io/crates/proptest)
(`test_fuzz_equivalence`, `test_fuzz_equivalence_heterogeneous_pages`),
1,000 scans each. The strategies combine batch size, projection, row
groups, row selection, selection policy, predicates, predicate cache
size, offset and limit. They include the edge values: batch sizes next
to the size of a page and of a row group, a limit of 0, and an offset
after the last row.
- If a scan fails, `proptest` shrinks it to a smaller scan that fails.
`proptest` is a new dev-dependency of `parquet`. `parquet-variant`
already uses it. Two choices that differ from the `proptest` defaults:
| Choice | Reason |
|---|---|
| The scans are the same in each run (fixed seed) | A failure in CI
comes from the change under test |
| No `proptest-regressions` files | The tests are in `src/`, and the
fixed seed makes a failure reproducible without them |
Each test fails if less than two thirds of its scans return rows, so
that it does not compare empty results only.
The tests compare rows, not batch boundaries: the sync reader lets a
batch span two row groups, and the push decoder does not.
# Are these changes tested?
The 4 new tests pass on `main` in about 3 seconds (debug build). To
check that they can fail, I removed the last row of each batch of
exactly 7 rows in `drive_file`. Both property tests failed, and
`proptest` reduced the failing scan to one with a batch size of 26 and
two predicates.
# Are there any user-facing changes?
No.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
---------
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
Cargo.lock | 1 +
parquet/Cargo.toml | 1 +
parquet/tests/arrow_reader/mod.rs | 1 +
.../tests/arrow_reader/push_decoder_equivalence.rs | 632 +++++++++++++++++++++
4 files changed, 635 insertions(+)
diff --git a/Cargo.lock b/Cargo.lock
index 4e89351a8d..ac62747684 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -2585,6 +2585,7 @@ dependencies = [
"parquet-geospatial",
"parquet-variant",
"parquet-variant-compute",
+ "proptest",
"rand 0.10.3",
"ring",
"seq-macro",
diff --git a/parquet/Cargo.toml b/parquet/Cargo.toml
index ec1c869c95..e34565c260 100644
--- a/parquet/Cargo.toml
+++ b/parquet/Cargo.toml
@@ -80,6 +80,7 @@ criterion = { workspace = true, default-features = false,
features = ["async_fut
snap = { version = "1.0", default-features = false }
tempfile = { version = "3.0", default-features = false }
insta = { workspace = true, default-features = true }
+proptest = { version = "1.0", default-features = false, features = ["std"] }
brotli = { version = "9.0", default-features = false, features = ["std"] }
flate2 = { version = "1.0", default-features = false, features =
["rust_backend"] }
lz4_flex = { version = "0.14", default-features = false, features = ["std",
"frame"] }
diff --git a/parquet/tests/arrow_reader/mod.rs
b/parquet/tests/arrow_reader/mod.rs
index c1b63dbe8f..d4a4460c4a 100644
--- a/parquet/tests/arrow_reader/mod.rs
+++ b/parquet/tests/arrow_reader/mod.rs
@@ -54,6 +54,7 @@ mod parquet_testing;
mod partial_offset_index;
#[cfg(feature = "async")]
mod predicate_cache;
+mod push_decoder_equivalence;
mod row_filter;
mod statistics;
diff --git a/parquet/tests/arrow_reader/push_decoder_equivalence.rs
b/parquet/tests/arrow_reader/push_decoder_equivalence.rs
new file mode 100644
index 0000000000..f2e51540f5
--- /dev/null
+++ b/parquet/tests/arrow_reader/push_decoder_equivalence.rs
@@ -0,0 +1,632 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Differential tests for [`ParquetPushDecoder`].
+//!
+//! Each test describes a scan as a [`Scan`], and [`assert_same_rows`] checks
+//! that the push decoder returns the same rows, in the same order, as
+//! [`ParquetRecordBatchReader`] for the same scan. The `test_fuzz_*` tests
+//! use [`proptest`] to combine the scan options at random and to shrink a
+//! scan that fails.
+//!
+//! [`ParquetRecordBatchReader`]:
parquet::arrow::arrow_reader::ParquetRecordBatchReader
+
+use arrow_array::builder::{Int64Builder, ListBuilder};
+use arrow_array::cast::AsArray;
+use arrow_array::types::Int64Type;
+use arrow_array::{
+ Array, ArrayRef, BooleanArray, Int64Array, RecordBatch, RecordBatchReader,
StringArray,
+ StructArray,
+};
+use arrow_schema::{DataType, Field, SchemaRef};
+use arrow_select::concat::concat_batches;
+use bytes::Bytes;
+use parquet::DecodeResult;
+use parquet::arrow::arrow_reader::{
+ ArrowPredicate, ArrowPredicateFn, ArrowReaderMetadata, ArrowReaderOptions,
+ ParquetRecordBatchReaderBuilder, RowFilter, RowSelection,
RowSelectionPolicy, RowSelector,
+};
+use parquet::arrow::push_decoder::{ParquetPushDecoder,
ParquetPushDecoderBuilder};
+use parquet::arrow::{ArrowWriter, ProjectionMask};
+use parquet::file::metadata::PageIndexPolicy;
+use parquet::file::properties::WriterProperties;
+use parquet::schema::types::SchemaDescriptor;
+use proptest::collection::vec;
+use proptest::option::weighted;
+use proptest::prelude::*;
+use proptest::sample::select;
+use proptest::test_runner::{Config, TestRng, TestRunner};
+use std::cell::Cell;
+use std::ops::Range;
+use std::sync::{Arc, LazyLock};
+
+const ROWS_PER_ROW_GROUP: usize = 600;
+const ROWS_PER_PAGE: usize = 25;
+const NUM_ROWS: usize = 1800;
+
+/// Three row groups of 600 rows, with a new data page each 25 rows or less,
+/// columns:
+///
+/// * `a`: 0, 1, 2, ... (plain)
+/// * `b`: `a % 10` (dictionary)
+/// * `c`: a string (plain)
+/// * `l`: a list of 0 to 3 `a` values, some lists null (plain)
+/// * `s`: a struct of `a * 2` and `a % 7` (plain)
+struct TestFile {
+ data: Bytes,
+ batch: RecordBatch,
+}
+
+static TEST_FILE: LazyLock<TestFile> = LazyLock::new(|| {
+ let a: Vec<i64> = (0..NUM_ROWS as i64).collect();
+ let mut lists = ListBuilder::new(Int64Builder::new());
+ for &v in &a {
+ if v % 11 == 5 {
+ lists.append_null();
+ } else {
+ for i in 0..(v % 4) {
+ lists.values().append_value(v * 10 + i);
+ }
+ lists.append(true);
+ }
+ }
+ let s = StructArray::from(vec![
+ (
+ Arc::new(Field::new("x", DataType::Int64, false)),
+ Arc::new(Int64Array::from_iter_values(a.iter().map(|v| v * 2))) as
ArrayRef,
+ ),
+ (
+ Arc::new(Field::new("y", DataType::Int64, false)),
+ Arc::new(Int64Array::from_iter_values(a.iter().map(|v| v % 7))) as
ArrayRef,
+ ),
+ ]);
+ let batch = RecordBatch::try_from_iter(vec![
+ ("a", Arc::new(Int64Array::from(a.clone())) as ArrayRef),
+ (
+ "b",
+ Arc::new(Int64Array::from_iter_values(a.iter().map(|v| v % 10)))
as ArrayRef,
+ ),
+ (
+ "c",
+ Arc::new(StringArray::from_iter_values(
+ a.iter().map(|v| format!("row-{v:08}-padding")),
+ )) as ArrayRef,
+ ),
+ ("l", Arc::new(lists.finish()) as ArrayRef),
+ ("s", Arc::new(s) as ArrayRef),
+ ])
+ .unwrap();
+
+ let props = WriterProperties::builder()
+ .set_max_row_group_row_count(Some(ROWS_PER_ROW_GROUP))
+ .set_data_page_row_count_limit(ROWS_PER_PAGE)
+ .set_write_batch_size(ROWS_PER_PAGE)
+ .set_dictionary_enabled(false)
+ .set_column_dictionary_enabled("b".into(), true)
+ .build();
+ let mut buffer = vec![];
+ let mut writer = ArrowWriter::try_new(&mut buffer, batch.schema(),
Some(props)).unwrap();
+ writer.write(&batch).unwrap();
+ writer.close().unwrap();
+ TestFile {
+ data: Bytes::from(buffer),
+ batch,
+ }
+});
+
+/// Row groups of [`HETEROGENEOUS_FILE`].
+const HETEROGENEOUS_ROW_GROUP_ROWS: [usize; 3] = [700, 700, 400];
+
+/// The rows of [`TEST_FILE`] in row groups of 700, 700 and 400 rows. A page
+/// ends after 40 rows or at about 200 bytes (checked every 5 rows), so each
+/// column has its own page boundaries.
+static HETEROGENEOUS_FILE: LazyLock<Bytes> = LazyLock::new(|| {
+ let props = WriterProperties::builder()
+ .set_max_row_group_row_count(Some(HETEROGENEOUS_ROW_GROUP_ROWS[0]))
+ .set_data_page_row_count_limit(40)
+ .set_data_page_size_limit(200)
+ .set_write_batch_size(5)
+ .set_dictionary_enabled(false)
+ .set_column_dictionary_enabled("b".into(), true)
+ .build();
+ let mut buffer = vec![];
+ let mut writer =
+ ArrowWriter::try_new(&mut buffer, TEST_FILE.batch.schema(),
Some(props)).unwrap();
+ writer.write(&TEST_FILE.batch).unwrap();
+ writer.close().unwrap();
+ Bytes::from(buffer)
+});
+
+fn load_metadata(data: &Bytes, policy: PageIndexPolicy) -> ArrowReaderMetadata
{
+ let options = ArrowReaderOptions::new().with_page_index_policy(policy);
+ ArrowReaderMetadata::load(data, options).unwrap()
+}
+
+static WITH_PAGE_INDEX: LazyLock<ArrowReaderMetadata> =
+ LazyLock::new(|| load_metadata(&TEST_FILE.data,
PageIndexPolicy::Required));
+static WITHOUT_PAGE_INDEX: LazyLock<ArrowReaderMetadata> =
+ LazyLock::new(|| load_metadata(&TEST_FILE.data, PageIndexPolicy::Skip));
+static HETEROGENEOUS_WITH_PAGE_INDEX: LazyLock<ArrowReaderMetadata> =
+ LazyLock::new(|| load_metadata(&HETEROGENEOUS_FILE,
PageIndexPolicy::Required));
+static HETEROGENEOUS_WITHOUT_PAGE_INDEX: LazyLock<ArrowReaderMetadata> =
+ LazyLock::new(|| load_metadata(&HETEROGENEOUS_FILE,
PageIndexPolicy::Skip));
+
+fn metadata(page_index: bool) -> ArrowReaderMetadata {
+ if page_index {
+ WITH_PAGE_INDEX.clone()
+ } else {
+ WITHOUT_PAGE_INDEX.clone()
+ }
+}
+
+fn schema() -> SchemaDescriptor {
+ metadata(true)
+ .metadata()
+ .file_metadata()
+ .schema_descr()
+ .clone()
+}
+
+fn columns(names: &[&str]) -> ProjectionMask {
+ ProjectionMask::columns(&schema(), names.iter().copied())
+}
+
+/// Decode `file`, pushing exactly the requested ranges. Returns the batches
+/// and every range that the decoder requested, in order.
+fn drive_file(
+ mut decoder: ParquetPushDecoder,
+ file: &Bytes,
+) -> (Vec<RecordBatch>, Vec<Range<u64>>) {
+ let mut batches = vec![];
+ let mut requested = vec![];
+ let mut buffered_after_last_batch = 0;
+ loop {
+ match decoder.try_decode().unwrap() {
+ DecodeResult::NeedsData(ranges) => {
+ assert!(!ranges.is_empty());
+ let data = ranges
+ .iter()
+ .map(|r| file.slice(r.start as usize..r.end as usize))
+ .collect();
+ requested.extend_from_slice(&ranges);
+ decoder.push_ranges(ranges, data).unwrap();
+ }
+ DecodeResult::Data(batch) => {
+ batches.push(batch);
+ buffered_after_last_batch = decoder.buffered_bytes();
+ }
+ DecodeResult::Finished => break,
+ }
+ }
+ // The last batch finishes the last row group, which releases its bytes.
+ assert_eq!(
+ buffered_after_last_batch, 0,
+ "bytes left after the last batch"
+ );
+ (batches, requested)
+}
+
+/// A predicate on one column: an Int64 column unless noted.
+#[derive(Clone, Copy, Debug)]
+enum Cmp {
+ Lt(i64),
+ Ge(i64),
+ ModNotZero(i64),
+ All,
+ None,
+ /// The list column `l` is not null.
+ ListNotNull,
+ /// Field `x` of the struct column `s` is less than the value.
+ StructXLt(i64),
+ /// Null where the value is a multiple of the argument, else true.
+ NullEvery(i64),
+}
+
+#[derive(Clone, Debug)]
+struct PredicateSpec {
+ column: &'static str,
+ cmp: Cmp,
+}
+
+impl PredicateSpec {
+ fn new(column: &'static str, cmp: Cmp) -> Self {
+ Self { column, cmp }
+ }
+
+ fn build(&self) -> Box<dyn ArrowPredicate> {
+ let cmp = self.cmp;
+ Box::new(ArrowPredicateFn::new(
+ columns(&[self.column]),
+ move |batch: RecordBatch| {
+ let column = batch.column(0);
+ Ok(match cmp {
+ Cmp::ListNotNull => {
+ let list = column.as_list::<i32>();
+ (0..list.len()).map(|i|
Some(!list.is_null(i))).collect()
+ }
+ Cmp::StructXLt(x) => column
+ .as_struct()
+ .column(0)
+ .as_primitive::<Int64Type>()
+ .iter()
+ .map(|v| Some(v.unwrap() < x))
+ .collect(),
+ Cmp::NullEvery(m) => column
+ .as_primitive::<Int64Type>()
+ .iter()
+ .map(|v| (v.unwrap() % m != 0).then_some(true))
+ .collect(),
+ _ => column
+ .as_primitive::<Int64Type>()
+ .iter()
+ .map(|v| {
+ let v = v.unwrap();
+ Some(match cmp {
+ Cmp::Lt(x) => v < x,
+ Cmp::Ge(x) => v >= x,
+ Cmp::ModNotZero(x) => v % x != 0,
+ Cmp::All => true,
+ Cmp::None => false,
+ _ => unreachable!(),
+ })
+ })
+ .collect::<BooleanArray>(),
+ })
+ },
+ ))
+ }
+}
+
+/// Apply the options of a [`Scan`] to an `ArrowReaderBuilder`. A macro,
+/// because the builders of the two readers have different types.
+macro_rules! apply_options {
+ ($scan:expr, $builder:expr) => {{
+ let scan: &Scan = $scan;
+ let mut builder = $builder;
+ if let Some(batch_size) = scan.batch_size {
+ builder = builder.with_batch_size(batch_size);
+ }
+ if let Some(projection) = &scan.projection {
+ builder = builder.with_projection(projection.clone());
+ }
+ if let Some(row_groups) = &scan.row_groups {
+ builder = builder.with_row_groups(row_groups.clone());
+ }
+ if let Some(selection) = &scan.selection {
+ builder = builder.with_row_selection(selection.clone());
+ }
+ if let Some(limit) = scan.limit {
+ builder = builder.with_limit(limit);
+ }
+ if let Some(offset) = scan.offset {
+ builder = builder.with_offset(offset);
+ }
+ if let Some(policy) = scan.policy {
+ builder = builder.with_row_selection_policy(policy);
+ }
+ if let Some(size) = scan.max_predicate_cache_size {
+ builder = builder.with_max_predicate_cache_size(size);
+ }
+ if !scan.predicates.is_empty() {
+ let predicates = scan.predicates.iter().map(|p|
p.build()).collect();
+ builder = builder.with_row_filter(RowFilter::new(predicates));
+ }
+ builder
+ }};
+}
+
+/// A scan configuration, so that the same scan can be built for more than
+/// one reader.
+#[derive(Clone, Debug, Default)]
+struct Scan {
+ /// Read [`HETEROGENEOUS_FILE`] instead of [`TEST_FILE`].
+ heterogeneous: bool,
+ page_index_off: bool,
+ batch_size: Option<usize>,
+ projection: Option<ProjectionMask>,
+ row_groups: Option<Vec<usize>>,
+ selection: Option<RowSelection>,
+ limit: Option<usize>,
+ offset: Option<usize>,
+ /// Built again for each reader, as predicates have state.
+ predicates: Vec<PredicateSpec>,
+ policy: Option<RowSelectionPolicy>,
+ max_predicate_cache_size: Option<usize>,
+}
+
+impl Scan {
+ fn metadata(&self) -> ArrowReaderMetadata {
+ match (self.heterogeneous, self.page_index_off) {
+ (false, _) => metadata(!self.page_index_off),
+ (true, false) => HETEROGENEOUS_WITH_PAGE_INDEX.clone(),
+ (true, true) => HETEROGENEOUS_WITHOUT_PAGE_INDEX.clone(),
+ }
+ }
+
+ fn data(&self) -> &'static Bytes {
+ if self.heterogeneous {
+ &HETEROGENEOUS_FILE
+ } else {
+ &TEST_FILE.data
+ }
+ }
+
+ fn push_decoder(&self) -> ParquetPushDecoder {
+ let builder =
ParquetPushDecoderBuilder::new_with_metadata(self.metadata());
+ apply_options!(self, builder).build().unwrap()
+ }
+
+ /// The batches of the scan from the sync reader.
+ fn sync_batches(&self) -> (SchemaRef, Vec<RecordBatch>) {
+ let builder = ParquetRecordBatchReaderBuilder::new_with_metadata(
+ self.data().clone(),
+ self.metadata(),
+ );
+ let reader = apply_options!(self, builder).build().unwrap();
+ let schema = reader.schema();
+ (schema, reader.map(|batch| batch.unwrap()).collect())
+ }
+}
+
+/// What the push decoder did for a scan.
+struct Decoded {
+ /// Rows returned.
+ rows: usize,
+ /// Every range requested, in order.
+ requested: Vec<Range<u64>>,
+}
+
+/// Assert that the push decoder gives the same rows as the sync reader for
+/// `scan`, in the same order.
+///
+/// The batch boundaries can differ: the push decoder ends a batch at the end
+/// of each row group.
+#[track_caller]
+fn assert_same_rows(scan: &Scan) -> Decoded {
+ let (schema, expected) = scan.sync_batches();
+ let (actual, requested) = drive_file(scan.push_decoder(), scan.data());
+ if let Some(batch_size) = scan.batch_size {
+ assert!(
+ actual.iter().all(|b| b.num_rows() <= batch_size),
+ "a batch has more than {batch_size} rows for {scan:?}"
+ );
+ }
+ let expected = concat_batches(&schema, &expected).unwrap();
+ let actual = concat_batches(&schema, &actual).unwrap();
+ // Do not print the rows: proptest runs a failing scan many times to
+ // shrink it.
+ assert!(
+ actual == expected,
+ "rows differ for {scan:?}: {} rows, expected {}",
+ actual.num_rows(),
+ expected.num_rows()
+ );
+ Decoded {
+ rows: actual.num_rows(),
+ requested,
+ }
+}
+
+#[test]
+#[cfg_attr(miri, ignore)] // Takes too long
+fn test_empty_selection_requests_nothing() {
+ let decoded = assert_same_rows(&Scan {
+ batch_size: Some(64),
+ selection: Some(RowSelection::from(vec![RowSelector::skip(NUM_ROWS)])),
+ ..Default::default()
+ });
+ assert_eq!(decoded.rows, 0);
+ assert!(decoded.requested.is_empty());
+}
+
+/// A selection of `total` rows: runs of skipped and selected rows in turn.
+/// It shrinks to a selection with fewer and shorter runs.
+fn selection_strategy(total: usize) -> impl Strategy<Value = RowSelection> {
+ let runs = select(vec![3usize, 40, 400]).prop_flat_map(|max_run|
vec(1..=max_run, 1..200));
+ (any::<bool>(), runs).prop_map(move |(mut skip, runs)| {
+ let mut selectors = vec![];
+ let mut rows_left = total;
+ // The last run takes the rows that are left.
+ let last = std::iter::once(total);
+ for run in runs.into_iter().chain(last) {
+ let run = run.min(rows_left);
+ if run == 0 {
+ break;
+ }
+ selectors.push(match skip {
+ true => RowSelector::skip(run),
+ false => RowSelector::select(run),
+ });
+ rows_left -= run;
+ skip = !skip;
+ }
+ RowSelection::from(selectors)
+ })
+}
+
+/// No batch size (the default), a batch size next to the size of a page or of
+/// a row group, or any batch size.
+fn batch_size_strategy() -> impl Strategy<Value = Option<usize>> {
+ let boundaries = [1, ROWS_PER_PAGE, ROWS_PER_ROW_GROUP, 700, NUM_ROWS]
+ .into_iter()
+ .flat_map(|size| [size.saturating_sub(1).max(1), size, size + 1])
+ .collect::<Vec<_>>();
+ prop_oneof![
+ 1 => Just(None),
+ 4 => select(boundaries).prop_map(Some),
+ 4 => (1..2 * NUM_ROWS).prop_map(Some),
+ ]
+}
+
+fn predicate_strategy() -> impl Strategy<Value = PredicateSpec> {
+ let value = 0..NUM_ROWS as i64;
+ prop_oneof![
+ value
+ .clone()
+ .prop_map(|v| PredicateSpec::new("a", Cmp::Lt(v))),
+ value
+ .clone()
+ .prop_map(|v| PredicateSpec::new("a", Cmp::Ge(v))),
+ (2..7i64).prop_map(|m| PredicateSpec::new("b", Cmp::ModNotZero(m))),
+ (2..40i64).prop_map(|m| PredicateSpec::new("a", Cmp::ModNotZero(m))),
+ Just(PredicateSpec::new("b", Cmp::All)),
+ Just(PredicateSpec::new("l", Cmp::ListNotNull)),
+ value.prop_map(|v| PredicateSpec::new("s", Cmp::StructXLt(2 * v))),
+ (2..9i64).prop_map(|m| PredicateSpec::new("a", Cmp::NullEvery(m))),
+ Just(PredicateSpec::new("b", Cmp::None)),
+ ]
+}
+
+/// A scan with batch size, projection, row groups, row selection, selection
+/// policy, predicates, predicate cache size, offset and limit at random.
+fn scan_strategy(heterogeneous: bool) -> impl Strategy<Value = Scan> {
+ let row_group_rows = match heterogeneous {
+ true => HETEROGENEOUS_ROW_GROUP_ROWS,
+ false => [ROWS_PER_ROW_GROUP; 3],
+ };
+ // Some of the row groups, in any order.
+ let row_groups = (Just(vec![0, 1, 2]).prop_shuffle(), 1..=3usize)
+ .prop_map(|(row_groups, len)| row_groups[..len].to_vec());
+ weighted(0.3, row_groups)
+ .prop_flat_map(move |row_groups| {
+ // A selection must cover exactly the row groups of the scan.
+ let total: usize = match &row_groups {
+ Some(row_groups) => row_groups.iter().map(|&i|
row_group_rows[i]).sum(),
+ None => NUM_ROWS,
+ };
+ (
+ Just(row_groups),
+ proptest::bool::weighted(0.1),
+ batch_size_strategy(),
+ select(vec![
+ None,
+ Some(vec!["a"]),
+ Some(vec!["c"]),
+ Some(vec!["a", "c"]),
+ Some(vec!["b", "c"]),
+ Some(vec!["l"]),
+ Some(vec!["s", "b"]),
+ Some(vec!["l", "a", "s"]),
+ ]),
+ weighted(0.5, selection_strategy(total)),
+ // Mostly a limit and an offset that leave rows. Sometimes a
+ // limit of 0, or an offset after the last row.
+ weighted(0.3, prop_oneof![9 => 1..=total, 1 => Just(0)]),
+ weighted(0.3, prop_oneof![9 => 0..total / 2, 1 => total..total
+ 300]),
+ vec(predicate_strategy(), 0..4),
+ select(vec![
+ None,
+ Some(RowSelectionPolicy::Selectors),
+ Some(RowSelectionPolicy::Mask),
+ ]),
+ select(vec![None, Some(0), Some(1)]),
+ )
+ })
+ .prop_map(
+ move |(
+ row_groups,
+ page_index_off,
+ batch_size,
+ projection,
+ selection,
+ limit,
+ offset,
+ predicates,
+ policy,
+ max_predicate_cache_size,
+ )| Scan {
+ heterogeneous,
+ page_index_off,
+ batch_size,
+ projection: projection.map(|names| columns(&names)),
+ row_groups,
+ selection,
+ limit,
+ offset,
+ predicates,
+ policy,
+ max_predicate_cache_size,
+ },
+ )
+}
+
+/// Scans of each fuzz test, so that it runs in a few seconds in a debug
+/// build.
+const FUZZ_CASES: u32 = 1000;
+
+/// Check [`FUZZ_CASES`] random scans. If a scan fails, proptest shrinks it to
+/// a smaller scan that fails. Most of the scans must return rows, so that the
+/// test does not compare empty results only.
+///
+/// The scans are the same in each run, so that a failure in CI comes from
+/// the change under test.
+fn fuzz(heterogeneous: bool) {
+ let config = Config {
+ cases: FUZZ_CASES,
+ failure_persistence: None,
+ ..Config::default()
+ };
+ let rng = TestRng::deterministic_rng(config.rng_algorithm);
+ let mut runner = TestRunner::new_with_rng(config, rng);
+ let scans_with_rows = Cell::new(0);
+ runner
+ .run(&scan_strategy(heterogeneous), |scan| {
+ if assert_same_rows(&scan).rows > 0 {
+ scans_with_rows.set(scans_with_rows.get() + 1);
+ }
+ Ok(())
+ })
+ .unwrap();
+ assert!(
+ scans_with_rows.get() * 3 >= FUZZ_CASES * 2,
+ "only {} of {FUZZ_CASES} scans returned rows",
+ scans_with_rows.get()
+ );
+}
+
+#[test]
+#[cfg_attr(miri, ignore)] // Takes too long
+fn test_fuzz_equivalence() {
+ fuzz(false);
+}
+
+#[test]
+#[cfg_attr(miri, ignore)] // Takes too long
+fn test_fuzz_equivalence_heterogeneous_pages() {
+ fuzz(true);
+}
+
+#[test]
+#[cfg_attr(miri, ignore)] // Takes too long
+fn test_page_boundaries_differ_per_column() {
+ let metadata = HETEROGENEOUS_WITH_PAGE_INDEX.metadata();
+ let row_groups: Vec<i64> = metadata
+ .row_groups()
+ .iter()
+ .map(|rg| rg.num_rows())
+ .collect();
+ assert_eq!(row_groups, [700, 700, 400]);
+ // Pages per column in row group 0: a, b, c, l, s.x, s.y
+ let page_index = metadata.page_index_for_row_group(0);
+ let pages: Vec<usize> = (0..6)
+ .map(|column| page_index.page_locations(column).unwrap().len())
+ .collect();
+ assert!(
+ pages[0] != pages[2] && pages[1] != pages[2] && pages[0] != pages[3],
+ "{pages:?}"
+ );
+}
