bench(parquet): add async RowSelection policy baseline

[hhhizzz]

Which issue does this PR close?

• Part of [Parquet] Adaptively to pick between RowSelection and Mask filter representation #8846.
• Part of [EPIC] Faster performance for parquet predicate evaluation for non selective filters #7456.
• Split out from Optimize parquet row filter auto strategy with adaptive fallback #9956.

Rationale for this change

This PR lands a reusable benchmark baseline before changing Parquet reader
behavior. It measures the async push-decoder path with
RowSelectionPolicy::Auto, so the same benchmark IDs can be compared directly
between main and candidate branches.

The existing row_selection_cursor benchmark remains unchanged. It continues
to provide the low-level phase diagram for forced Mask and Selectors when a
caller already has a RowSelection. Forced-policy and manual
decode-then-filter diagnostics are intentionally kept out of this daily Auto
baseline and can be added later as separate oracle benchmarks.

What changes are included in this PR?

This PR changes benchmarks only. Production reader code and public APIs are
unchanged.

It adds arrow_reader_row_selection_policy, an Auto-only async benchmark backed
by a deterministic in-memory Parquet fixture:

• 65,536 rows per row group;
• one Int32 predicate column and eight Int32 output columns;
• uncompressed pages with dictionary encoding disabled;
• cached metadata and lazy fixture construction outside the timed loop;
• an exact correctness preflight before timing every benchmark case;
• explicit PageIndexPolicy::Skip metadata loading.

Page indexes are intentionally disabled so this target isolates RowSelection
execution and materialization from page-level I/O pruning. This is also why its
deterministic fixture remains separate from arrow_reader_row_filter.

The final 12 benchmark IDs are divided by the variable they isolate:

• auto/shape — eight controlled selectivity and run-shape cases;
• auto/order — the same two sparse and two fragmented row groups in exact
  reverse order;
• auto/scale — the fragmented shape at one and eight row groups.

The shape cases are:

case	selectivity	selectors/RG	average run	selected run	skipped run
sparse_1_56_run32	1.56%	64	1,024	32	2,016
moderate_12_5_run32	12.5%	512	128	32	224
fragmented_50_run1	50%	65,536	1	1	1
regular_50_run32	50%	2,048	32	32	32
clustered_50_run128	50%	512	128	128	128
bursty_50_same_summary	50%	2,048	32	32 mean	32 mean
dense_98_44_skip1_select63	98.44%	2,048	32	63	1
all_selected	100%	0	n/a	n/a	n/a

The regular and bursty cases have identical selectivity, selector density,
mean selected run, and mean skipped run; only run variance and ordering differ.
The order cases have the same row-group multiset and output row count, so row
group order is their only variable.

Before Criterion starts timing, the preflight collects payload_0 from every
selected output row and compares it with the exact expected global-row
sequence. This detects wrong-row selection, output reordering, and row-group
boundary errors rather than checking only the total row count. The timed runner
still only consumes batches and accumulates rows.

No workload-derived cases are included yet. Future ClickBench or TPC-DS cases
should be based on captured selector traces with recorded provenance rather
than synthetic approximations.

Follow-up roadmap

1. Use this Auto-only target for direct main versus candidate comparisons.
2. Add a separate oracle target for forced Mask, forced Selectors, and
   manual decode-then-filter diagnostics when needed.
3. Keep threshold, payload, overlap, predicate-chain, and page-alignment sweeps
   in a separate extended benchmark rather than the daily baseline.
4. Run broader DataFusion TPC-DS or ClickBench validation only after focused
   regressions have been understood.

Are these changes tested?

Yes. Local validation included:

cargo fmt --all -- --check
cargo clippy -p parquet --bench arrow_reader_row_selection_policy --features arrow,async -- -D warnings
cargo bench -p parquet --bench arrow_reader_row_selection_policy --features arrow,async --no-run
cargo bench -p parquet --bench arrow_reader_row_selection_policy --features arrow,async -- --list
cargo bench -p parquet --bench arrow_reader_row_selection_policy --features arrow,async -- --test
cargo test -p parquet --all-features
git diff --check

All 12 benchmark cases passed their exact correctness preflight locally.

A full default Criterion measurement was also run on AMD64 Linux
(sz-data-b-1) at eaec65a202 against base cd17899fe4:

• 12 benchmark IDs completed;
• total elapsed: 142.080 seconds;
• 5 sample-time warnings reported that the default 5-second target could not
  complete 100 samples for slower cases; no case failed;
• selected Criterion medians:
  • fragmented_50_run1: 2.482 ms;
  • regular_50_run32: 1.773 ms;
  • bursty_50_same_summary: 1.801 ms;
  • dense_98_44_skip1_select63: 1.950 ms.

This run validates the target's runtime and observable shape signals; it is not
a current-versus-main performance comparison, and no performance improvement is
claimed in this PR.

Are there any user-facing changes?

No. This only changes benchmark code.

AI-assisted development

The benchmark design and implementation were developed with AI assistance and
reviewed against the requested benchmark scope, repository standards,
formatting, Clippy, the full Parquet test suite, release benchmark builds, and
the full Criterion run above.

[hhhizzz]

Hi @alamb , could you take a look at this PR when you have a chance? It is a relatively small benchmark-only PR for Parquet row-filter/materialization policy coverage, with no production reader behavior changes. The current checks are green.

[alamb]

Hi @alamb , could you take a look at this PR when you have a chance? It is a relatively small benchmark-only PR for Parquet row-filter/materialization policy coverage, with no production reader behavior changes. The current checks are green.

Yes, sorry @hhhizzz -- I will do so

The notion of what is a 'relatively small' PR has grown massively since even 9 months ago 😆 😭

[alamb]

Thank you @hhhizzz --- this is quite a comprehensive set of benchmarks -- however, I wonder if we really need all this coverage?

For example, there are 27 variants of FilterType -- I see there are some comments about some of the filter shapes matching particular TPCH / TPCDS predicate shapes, but I am trying ti figure out why a benchmark that compares all the different is going to be helpful in the long run to avoid regressions -- I worry taht this benchmark will generate so much data that we will find it hard to run / reason about

It seems like this benchmark may be most useful a development/tuning benchmark as it helps establish baseline timings for row-filter materialization policy choices: That is useful when designing future heuristics but it is hard to grok the results

Are there any important filter cases we should add to arrow_row_filter?

[alamb]

this is similar to arrow_reader_row_filter -

arrow-rs/parquet/benches/arrow_reader_row_filter.rs

Lines 322 to 452 in cf450b5

	pub(crate) enum FilterType {
	/// point lookup: selects a single row in 500K.
	/// ```text
	/// ┌───────────────┐ ┌───────────────┐
	/// │ │ │ │
	/// │ │ │ ... │
	/// │ │ │ │
	/// │ │ │ │
	/// │ ... │ │▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒│
	/// │ │ │ │
	/// │ │ │ ... │
	/// │ │ │ │
	/// │ │ │ │
	/// └───────────────┘ └───────────────┘
	/// ```
	PointLookup,
	/// selective (1%) unclustered filter: approx 5K selected rows in 500K.
	/// ```text
	/// ┌───────────────┐ ┌───────────────┐
	/// │ ... │ │ │
	/// │ │ │ │
	/// │▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒│ │ │
	/// │ │ │ ... │
	/// │ │ │ │
	/// │ │ │ │
	/// │ ... │ │ │
	/// │ │ │▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒│
	/// │ │ │ │
	/// └───────────────┘ └───────────────┘
	/// ```
	SelectiveUnclustered,
	/// moderately selective (10%) clustered filter: 50 selected runs of 1K
	/// rows each in 500K.
	/// ```text
	/// ┌───────────────┐ ┌───────────────┐
	/// │ │ │ │
	/// │ │ │ │
	/// │ ... │ │ ... │
	/// │ │ │ │
	/// │ │ │ │
	/// │ │ │ │
	/// │ │ │ │
	/// │▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒│ │▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒│
	/// │▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒│ │▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒│
	/// └───────────────┘ └───────────────┘
	/// ```
	ModeratelySelectiveClustered,
	/// moderately selective (~9%) unclustered filter: approx 45K selected
	/// rows in 500K.
	/// ```text
	/// ┌───────────────┐ ┌───────────────┐
	/// │ ... │ │ │
	/// │ │ │ │
	/// │▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒│ │▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒│
	/// │ │ │ │
	/// │ │ │ │
	/// │ │ │ ... │
	/// │ ... │ │ │
	/// │ │ │ │
	/// │ │ │▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒│
	/// └───────────────┘ └───────────────┘
	/// ```
	ModeratelySelectiveUnclustered,
	/// unselective (99%) unclustered filter: approx 495K selected rows in
	/// 500K.
	/// ```text
	/// ┌───────────────┐ ┌───────────────┐
	/// │▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒│ │▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒│
	/// │▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒│ │▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒│
	/// │▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒│ │▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒│
	/// │▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒│ │▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒│
	/// │▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒│ │▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒│
	/// │▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒│ │ │
	/// │▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒│ │▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒│
	/// │▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒│ │▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒│
	/// │▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒│ │▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒│
	/// └───────────────┘ └───────────────┘
	/// ```
	UnselectiveUnclustered,
	/// unselective (90%) clustered filter: 50 selected runs of 9K rows each
	/// in 500K.
	/// ```text
	/// ┌───────────────┐ ┌───────────────┐
	/// │▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒│ │▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒│
	/// │▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒│ │▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒│
	/// │▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒│ │▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒│
	/// │▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒│ │▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒│
	/// │▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒│ │▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒│
	/// │▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒│ │▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒│
	/// │▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒│ │▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒│
	/// │▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒│ │ ... │
	/// │ │ │ │
	/// └───────────────┘ └───────────────┘
	/// ```
	UnselectiveClustered,
	/// composite sparse filter: `SelectiveUnclustered` AND
	/// `ModeratelySelectiveClustered`, approx 0.1% selected rows in 500K.
	/// ```text
	/// ┌───────────────┐ ┌───────────────┐
	/// │ │ │ │
	/// │ │ │ ... │
	/// │ │ │ │
	/// │ │ │ │
	/// │ ... │ │ │
	/// │ │ │ │
	/// │ │ │▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒│
	/// │ │ │ │
	/// │▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒│ │ │
	/// └───────────────┘ └───────────────┘
	/// ```
	Composite,
	/// `utf8View <> ''` modeling [ClickBench] [Q21-Q27] with fragmented
	/// short string runs and sentinel values every 1K rows.
	/// ```text
	/// ┌───────────────┐ ┌───────────────┐
	/// │ │ │ │
	/// │ ... │ │ ... │
	/// │▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒│ │ │
	/// │▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒│ │ │
	/// │ │ │▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒│
	/// │ │ │▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒│
	/// │ ... │ │ ... │
	/// │ │ │ │
	/// │▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒│ │ │
	/// └───────────────┘ └───────────────┘
	/// ```
	///
	/// [ClickBench]: https://github.com/ClickHouse/ClickBench
	/// [Q21-Q27]: https://github.com/apache/datafusion/blob/b7177234e65cbbb2dcc04c252f6acd80bb026362/benchmarks/queries/clickbench/queries.sql#L22-L28
	Utf8ViewNonEmpty,
	}

Can we consolidate into a shared module like parquet/benches/row_filter_fixture.rs ?

[hhhizzz]

Done. I extracted the shared synthetic reader fixture into parquet/benches/arrow_reader_common/mod.rs and wired both arrow_reader_row_filter.rs and arrow_reader_materialization_policy.rs through it.

I used a bench-local module directory instead of a flat row_filter_fixture.rs so the shared helpers stay scoped to these reader benches. I can rename it if you prefer the flatter file name.

[alamb]

the only thing different in each of these branches is the selection policy, right? We could probably collapse them significantly

[hhhizzz]

Done. The async strategy handling is now collapsed through AsyncStrategy::row_selection_policy(), so the pushdown path is shared and only the full post-filter path remains separate.

[hhhizzz]

Thank you @hhhizzz --- this is quite a comprehensive set of benchmarks -- however, I wonder if we really need all this coverage?

For example, there are 27 variants of FilterType -- I see there are some comments about some of the filter shapes matching particular TPCH / TPCDS predicate shapes, but I am trying ti figure out why a benchmark that compares all the different is going to be helpful in the long run to avoid regressions -- I worry taht this benchmark will generate so much data that we will find it hard to run / reason about

It seems like this benchmark may be most useful a development/tuning benchmark as it helps establish baseline timings for row-filter materialization policy choices: That is useful when designing future heuristics but it is hard to grok the results

Are there any important filter cases we should add to arrow_row_filter?

I took another pass at narrowing the benchmark scope.

The materialization-policy target is now explicitly focused on policy-boundary/tuning cases rather than being a broad row-filter regression matrix:

• reduced the materialization-policy benchmark to one group: arrow_reader_materialization_policy_async_focus
• pruned duplicate/weak cases, including the small scalar-prefix case
• moved chained predicate-order coverage out of materialization and into arrow_reader_row_filter_async_predicate_order_focus
• kept the materialization cases as 21 named reader-level shapes rather than a full Cartesian product
• extracted the shared synthetic reader fixture so the row-filter and materialization benches use the same setup

For arrow_reader_row_filter, the important missing case I added is chained RowFilter predicate ordering: fixed-width predicate before var-width predicate, and the reverse order. The existing Composite case evaluated both predicates inside one ArrowPredicateFn, so it did not exercise pruning between sequential predicates.

If this still feels too broad, I can trim the materialization-policy target further, but the current split is intended to keep generic row-filter behavior in arrow_reader_row_filter and leave the separate materialization target for future Auto policy tuning.

[hhhizzz]

@alamb Thanks for the feedback. I pushed an update that narrows this PR toward a smaller, maintainable benchmark baseline rather than a broad policy-tuning sweep.

Main changes in this revision:

• Kept the shared arrow_reader_common fixture so the synthetic parquet data setup is not duplicated across reader benchmarks.
• Reduced arrow_reader_row_filter so it remains a reader regression baseline:
  • removed the sync strategy matrix
  • kept only a small async strategy matrix with representative fixed-width and Utf8View filters
  • reduced the nested-output focus case to full_post_filter vs Auto
• Reduced arrow_reader_materialization_policy to 10 representative cases, each still comparing:
  • full post-filtering
  • Auto
  • forced Mask
  • forced Selectors

The intent is that arrow_reader_row_filter covers general reader/filter/projection regressions, while arrow_reader_materialization_policy keeps just enough focused coverage to detect whether Auto is choosing a sensible fallback path for cases like high selectivity, projected predicate columns, count-only output, and variable-width deferred output.

I also measured the trimmed default Criterion runtime.(Tested a 24 core AMD64 linux machine)

target	benchmark ids	elapsed
arrow_reader_row_filter	74	12:42.39
arrow_reader_materialization_policy	40	6:38.09
combined	114	19:20.48

For comparison, before the reduction these two targets took about 35:32.88 combined. So this keeps the fallback/policy signal while bringing the default runtime down substantially.

Validation:

cargo bench -p parquet --features arrow,async --no-run --bench arrow_reader_row_filter --bench arrow_reader_materialization_policy

One note: I left row_selection_cursor as a separate target because it exercises the lower-level selector-vs-mask shape boundary. If you would prefer this PR to focus only on the reader/materialization benchmarks, I can split that target into a follow-up.

[hhhizzz]

@alamb I’ve force-pushed a substantially narrower, benchmark-only revision based on your feedback. The PR now contains 12 deterministic async RowSelectionPolicy::Auto benchmarks: 8 controlled shapes, 2 exact reverse-order cases, and 2 scale anchors. It verifies the exact selected row sequence before timing, explicitly disables PageIndex loading, and is now named arrow_reader_row_selection_policy to clarify its scope. All 12 cases completed a full Criterion run on AMD64 Linux in 142 seconds. Could you please take another look when you have time?