diff --git a/datafusion/notes/conditional-lazy-fully-matched-detection-plan.md b/datafusion/notes/conditional-lazy-fully-matched-detection-plan.md new file mode 100644 index 0000000000000..3bfe8a457a035 --- /dev/null +++ b/datafusion/notes/conditional-lazy-fully-matched-detection-plan.md @@ -0,0 +1,322 @@ + + +# Plan: Conditional and Lazy Fully-Matched Row-Group Detection + +## Objective + +Avoid constructing and evaluating an inverted pruning predicate when no later +scan stage can use the resulting `fully_matched` row-group flags. + +The optimization must preserve: + +- row-group pruning decisions; +- `LIMIT` pruning; +- page-index pruning and page-index load avoidance; +- suppression of row-level pushed-down filters for fully matched groups; +- existing conservative null handling; +- pruning and error metrics. + +## Current behavior + +`RowGroupAccessPlanFilter::prune_by_statistics_inner` performs two conceptually +different operations: + +1. It evaluates the normal pruning predicate and skips row groups that cannot + match. +2. For every surviving candidate, it calls + `identify_fully_matched_row_groups`. + +The second operation: + +- wraps the original expression in `NOT`; +- adds `IS NULL` disjuncts for nullable referenced columns; +- simplifies the expression; +- builds another `PruningPredicate`; +- extracts row-group statistics again; +- evaluates the inverted predicate; +- marks groups pruned by the inverted predicate as fully matched. + +This is performed unconditionally whenever statistics pruning succeeds and at +least one row group survives. + +The resulting flags are consumed only by: + +- `RowGroupAccessPlanFilter::prune_by_limit`; +- `RowGroupsPrunedParquetOpen::should_load_page_index`; +- `PagePruningAccessPlanFilter`, which skips fully matched groups; +- the push decoder, which suppresses its `RowFilter` for fully matched groups. + +Relevant implementation locations: + +- `datafusion/datasource-parquet/src/row_group_filter.rs` + - `prune_by_statistics_inner` + - `identify_fully_matched_row_groups` +- `datafusion/datasource-parquet/src/opener/mod.rs` + - `FiltersPreparedParquetOpen::prune_row_groups` + - `RowGroupsPrunedParquetOpen::should_load_page_index` + - `RowGroupsPrunedParquetOpen::build_stream` +- `datafusion/datasource-parquet/src/page_filter.rs` +- `datafusion/datasource-parquet/src/push_decoder.rs` + +## Why this should be lazy + +Fully-matched detection has no effect when all of the following are true: + +- there is no applicable order-insensitive `LIMIT`; +- page pruning cannot run for a surviving row group; +- row-level filter pushdown is disabled or no row filter can be built. + +In that case, only the normal may-match result affects the scan. The inverted +predicate is pure overhead. + +The common default configuration makes this case plausible: + +- row-group statistics pruning is enabled; +- row-filter pushdown is disabled by default; +- page-index support is enabled, but many files do not contain column and + offset indexes; +- many queries do not have a usable `LIMIT`. + +## Proposed design + +Separate ordinary pruning from fully-matched classification. + +### Phase 1: Perform normal row-group pruning + +Change the statistics-pruning path so it: + +1. evaluates the normal predicate; +2. updates skipped/matched statistics metrics; +3. records or returns surviving candidate indexes; +4. does not immediately evaluate the inverted predicate. + +Keep the current vectorized evaluation over all eligible row groups. + +Avoid adding a public boolean argument such as `detect_fully_matched` to the +existing API if possible. Prefer one of: + +- a separate crate-private method that classifies surviving groups; or +- an internal options/policy type with named semantics. + +The distinction between "may match" and "fully matches" should remain explicit +in the API. + +### Phase 2: Decide whether classification has a consumer + +After normal pruning, derive a `needs_fully_matched_detection` decision from +the prepared scan and surviving access plan. + +Detection is needed when at least one of these conditions holds: + +1. `prepared.limit.is_some() && !prepared.preserve_order`; +2. a row-level pushed-down `RowFilter` will be installed; +3. page-index pruning is possible for at least one surviving row group and one + page-predicate column. + +For page pruning, use footer offsets to establish applicability without loading +the page index: + +- map page-predicate columns to Parquet leaf columns; +- inspect surviving row groups; +- require both `column_index_offset` and `offset_index_offset`. + +Refactor the existing footer-offset logic in `should_load_page_index` into a +helper so the applicability check has one implementation. + +For the first implementation, `prepared.pushdown_filters && predicate.is_some()` +may conservatively enable detection. A follow-up can move row-filter candidate +prebuilding earlier and enable detection only when a non-empty +`RowFilterContext` exists. + +### Phase 3: Classify only when needed + +When the decision is true, invoke the existing inverted-predicate logic for +the surviving candidates. + +Preserve these correctness properties: + +- include `IS NULL` for nullable predicate columns; +- use `missing_null_counts_as_zero: false`; +- treat simplification, predicate construction, or evaluation errors as + "unable to prove fully matched"; +- never change an ordinary row-group prune result because classification + failed. + +When the decision is false, leave every surviving group's `fully_matched` flag +false. + +### Phase 4: Reuse statistics when classification is active + +Treat this as a follow-up unless it keeps the first change simple. + +Both normal and inverted predicates may request the same min, max, null-count, +and row-count arrays. Introduce a per-file evaluation cache keyed by: + +- logical column identity; +- statistics type; +- missing-null-count policy. + +The missing-null-count policy must be part of the key because ordinary pruning +uses `true`, while fully-matched proof uses `false`. + +Do not cache untrusted byte-array values without retaining the existing +per-row-group masking behavior. + +## Implementation sequence + +1. Extract a helper that reports whether page-index pruning is possible from + footer offsets for the current predicate columns and access plan. +2. Separate fully-matched classification from normal statistics pruning. +3. Add the consumer decision in the opener. +4. Preserve the current behavior when any consumer exists. +5. Add observability suitable for benchmarks. Prefer a benchmark-local counter + or an existing timer split over a permanent user-facing metric unless the + metric has independent operational value. +6. Add the optional statistics cache in a separate commit or PR after the lazy + behavior is measured. + +## Correctness tests + +Add focused tests for the decision matrix: + +- no limit, no page index, pushdown disabled: + - ordinary pruning is unchanged; + - no group is classified as fully matched. +- usable `LIMIT`: + - fully matched groups are still found; + - `prune_by_limit` keeps enough fully matched groups; + - order-sensitive limits do not trigger classification solely for `LIMIT`. +- page index present for a predicate column: + - classification still runs; + - fully matched groups avoid page-index work. +- page index absent: + - classification is skipped when there is no other consumer; + - page-index load remains skipped. +- page index present only on already-pruned groups: + - classification is skipped when there is no other consumer. +- row-filter pushdown enabled: + - fully matched groups still suppress the `RowFilter`; + - partially matched groups retain the filter. +- nullable columns and missing null counts: + - no group is incorrectly marked fully matched. +- predicate inversion/build/evaluation failure: + - groups remain scannable and unclassified. +- initial `ParquetAccessPlan` containing `Selection` and `Skip` entries: + - only surviving groups participate. + +Extend existing tests in: + +- `row_group_filter.rs`; +- opener page-index tests; +- push-decoder fully-matched tests; +- `datafusion/core/tests/parquet/row_group_pruning.rs`. + +## Performance benchmark + +Add a Criterion benchmark to `datafusion-datasource-parquet` that constructs +metadata with configurable: + +- row-group count: 1, 16, 128, 1024; +- predicate width: 1, 4, 8 columns; +- fraction surviving normal pruning; +- nullable versus non-nullable columns; +- presence or absence of consumers. + +Measure these cases: + +1. normal pruning only, classification skipped; +2. current-equivalent eager classification; +3. classification required by `LIMIT`; +4. classification required by page pruning; +5. classification required by row-filter pushdown. + +Primary measurements: + +- wall-clock time per file; +- `statistics_eval_time`; +- allocations if the benchmark environment supports allocation tracking. + +Expected result for the no-consumer case: + +- eliminate one predicate build and one statistics evaluation; +- reduce statistics-pruning CPU by approximately 30–50% when many groups + survive; +- no regression when every group is removed by the normal pass. + +End-to-end validation should include many small Parquet files because metadata +and pruning CPU are most visible there. + +## Risks and mitigations + +### Missing a consumer + +Risk: classification is skipped even though a later stage relies on the flags, +reducing performance or changing `LIMIT` planning. + +Mitigation: centralize the decision, enumerate all consumers in its +documentation, and add one test per consumer. + +### Circular page-index decision + +Risk: page-index loading currently uses fully-matched flags to decide whether +loading is useful, while lazy classification wants to know whether a page +index exists before producing those flags. + +Mitigation: split "footer says a relevant index exists" from "all surviving +groups are fully matched." The first check determines whether classification +has a page-pruning consumer; the second remains an optimization after +classification. + +### Row-filter applicability + +Risk: `pushdown_filters` is true but no pushdown candidate can be constructed, +causing unnecessary classification. + +Mitigation: begin conservatively, then optionally move candidate prebuilding +earlier so the decision uses an actual `RowFilterContext`. + +### Metric changes + +Risk: `row_groups_pruned_statistics.fully_matched` decreases when +classification is intentionally skipped. + +Mitigation: document that `fully_matched` counts proven groups used by enabled +optimizations, not all theoretically provable groups. Verify that total +pruned/matched accounting remains internally consistent. + +## Acceptance criteria + +- Query results and ordinary row-group pruning decisions are unchanged. +- All three consumers retain existing behavior when enabled and applicable. +- No inverted predicate is built in the no-consumer path. +- Existing fully-matched, page-index, limit, and pushdown tests pass. +- A benchmark demonstrates a repeatable pruning-CPU improvement in the + no-consumer case without a material regression in consumer cases. +- `cargo fmt --all`, workspace clippy, and the required extended tests pass. + +## PR strategy + +Prefer two reviewable changes: + +1. Lazy/conditional fully-matched detection with tests and benchmarks. +2. Optional reuse of extracted statistics between normal and inverted + evaluation. + +The first PR should not change public configuration or query semantics. diff --git a/datafusion/notes/sequential-bloom-filter-io-plan.md b/datafusion/notes/sequential-bloom-filter-io-plan.md new file mode 100644 index 0000000000000..0836292526288 --- /dev/null +++ b/datafusion/notes/sequential-bloom-filter-io-plan.md @@ -0,0 +1,410 @@ + + +# Plan: Remove Sequential Parquet Bloom-Filter I/O + +## Objective + +Reduce remote object-store latency during Parquet row-group Bloom-filter +pruning by replacing the current serial row-group/column read loop with bounded +or batched I/O. + +The optimization must preserve: + +- conservative pruning semantics; +- support for missing and partially available Bloom filters; +- support for files that omit `bloom_filter_length`; +- custom `ParquetFileReaderFactory` behavior; +- bytes-scanned and pruning metrics; +- bounded memory and request concurrency. + +## Target workload + +This optimization is intended for a specific workload: + +- Parquet files contain Split Block Bloom Filters; +- the query has equality or `IN` predicates that produce literal guarantees; +- min/max statistics leave many row groups as candidates; +- the Bloom filters can eliminate many of those candidates; +- storage has meaningful per-request latency. + +A representative query is a point lookup on an unsorted, high-cardinality key: + +```sql +SELECT ... +FROM remote_parquet +WHERE user_id = 'target'; +``` + +This is not a universal Parquet optimization. It should be evaluated separately +from scans where Bloom filters are absent, min/max statistics already prune +most groups, or data decoding dominates execution time. + +## Current behavior + +`FiltersPreparedParquetOpen::load_bloom_filters`: + +1. discovers columns referenced by literal guarantees; +2. iterates over every surviving row group; +3. iterates over each relevant Parquet column; +4. calls and awaits + `ParquetRecordBatchStreamBuilder::get_row_group_column_bloom_filter`; +5. inserts each successfully loaded `Sbbf` into a per-row-group map. + +Each pair is completed before the next read begins. + +In parquet-rs, one request normally reads the encoded Bloom header and bitset. +When `bloom_filter_length` is absent, reading a filter requires two serial +requests: + +1. read an estimated header range; +2. parse `num_bytes` and read the bitset. + +For `R` surviving row groups and `C` literal columns, DataFusion therefore +performs up to `R * C` serialized filter operations and potentially +`2 * R * C` serialized range requests. + +Relevant implementation locations: + +- `datafusion/datasource-parquet/src/opener/mod.rs` + - `FiltersPreparedParquetOpen::load_bloom_filters` +- `datafusion/datasource-parquet/src/reader.rs` + - `AsyncFileReader::get_bytes` + - `AsyncFileReader::get_byte_ranges` +- parquet-rs: + - `ParquetRecordBatchStreamBuilder::get_row_group_column_bloom_filter` + - Bloom-header parsing and `Sbbf` construction + +## Performance model + +Ignoring transfer time, the current latency is approximately: + +```text +serial_time = filter_requests * request_latency +``` + +With bounded parallelism `P`: + +```text +parallel_time ~= ceil(filter_requests / P) * request_latency +``` + +Example: + +- 128 surviving row groups; +- 2 predicate columns; +- one request per filter; +- 10 ms request latency. + +The serialized latency floor is approximately 2.56 seconds. Parallelism of 16 +has an ideal latency floor of approximately 160 ms, before service, parsing, +and transfer overhead. + +The benchmark must report actual end-to-end results rather than presenting +this idealized ratio as the expected production speedup. + +## Design constraints + +### A single `AsyncFileReader` is mutably borrowed + +The existing parquet-rs method takes `&mut self`, so DataFusion cannot place +multiple calls on one builder into `FuturesUnordered`. + +### Reader factories may be custom + +Creating one reader per Bloom filter would technically enable concurrency, but +it could: + +- repeatedly register metrics; +- create many connections or file handles; +- bypass assumptions in custom reader factories; +- introduce unbounded resource use. + +That approach should not be the default design. + +### Filter lengths may be missing + +Footer metadata may contain an offset without a length. A batch implementation +must retain the current two-stage header/bitset behavior. + +### Error isolation matters + +Current behavior logs an individual filter error, increments +`predicate_evaluation_errors`, and continues conservatively. A single failed +batched request must not cause incorrect pruning or fail the entire query. + +## Recommended architecture + +Prefer a parquet-rs batch API operating on one `AsyncFileReader`, followed by a +small DataFusion integration. + +### parquet-rs API + +Add an asynchronous method conceptually equivalent to: + +```rust +get_row_group_column_bloom_filters(requests) +``` + +where each request identifies a row group and column. The exact API should +follow parquet-rs conventions and preserve request/result ordering. + +The implementation should: + +1. inspect footer metadata for all requests; +2. return `None` entries without I/O when no Bloom offset exists; +3. partition present filters into known-length and unknown-length sets; +4. fetch known-length encoded filters with `get_byte_ranges`; +5. fetch unknown-length header estimates with `get_byte_ranges`; +6. parse headers and calculate bitset ranges; +7. fetch unknown-length bitsets with a second `get_byte_ranges`; +8. validate algorithm, compression, hash, and lengths as the existing scalar + method does; +9. construct `Sbbf` values; +10. return per-request outcomes so DataFusion can continue conservatively. + +Implement the existing scalar method in terms of the batch primitive, or share +one parsing helper, to prevent behavior drift. + +### Bounded batches + +Do not submit every filter in a very large file as one unbounded operation. +Process requests in configurable or internally fixed chunks. + +Initial benchmark values: + +- batch sizes: 8, 16, 32, 64; +- row groups: 16, 128, 1024; +- predicate columns: 1, 2, 8. + +The initial implementation can use a conservative internal batch size without +adding a DataFusion configuration option. Add a user-facing option only if +measurements show stores need materially different tuning. + +### DataFusion integration + +In `load_bloom_filters`: + +1. build request descriptors only for surviving row groups and literal + columns; +2. omit requests whose footer metadata has no Bloom offset; +3. invoke the batch API in bounded chunks; +4. place successful results into the existing + `Vec`; +5. preserve the current per-row-group column names and physical type metadata; +6. count errors and leave failed filters absent; +7. run the existing conservative pruning path. + +Retain the dedicated replacement reader used for subsequent decoding unless a +separate benchmark proves that reader reuse is safe and valuable. + +## DataFusion-only fallback + +If an upstream parquet-rs batch API is not initially feasible, prototype a +bounded worker-pool design: + +- create at most `P` readers through `ParquetFileReaderFactory`; +- assign multiple filter requests sequentially to each reader; +- execute the `P` worker queues concurrently; +- keep `P` small and fixed for the experiment. + +Do not create one reader per filter. + +Use this fallback primarily to establish the performance ceiling and collect +benchmark evidence. The upstream batch API remains preferable because it: + +- uses `AsyncFileReader::get_byte_ranges`; +- requires fewer reader instances; +- centralizes Bloom encoding details in parquet-rs; +- can benefit other parquet-rs consumers. + +## Preliminary cleanup + +Land or include these low-risk changes before interpreting benchmarks: + +1. Skip Bloom setup when + `remaining_row_group_count() == 0`, rather than checking whether the access + plan vector has zero slots. +2. Avoid allocating a dense per-file Bloom vector before determining that no + Bloom work is required, where practical. +3. Remove the nested `bloom_filter_eval_time` timer so reported evaluation + time is not double counted. +4. Keep Bloom I/O time and Bloom predicate-evaluation time distinguishable. + Add a dedicated load timer only if it is useful beyond the benchmark. + +These changes do not solve serialized I/O, but they make measurements easier +to interpret. + +## Correctness tests + +Test the batch primitive and DataFusion integration for: + +- no Bloom offset; +- one known-length filter; +- several known-length filters; +- unknown `bloom_filter_length`; +- mixed known and unknown lengths; +- multiple row groups and multiple columns; +- unsupported or malformed header metadata; +- truncated header; +- truncated bitset; +- invalid offset or length conversion; +- one failed filter among successful filters; +- duplicate requests, if the API allows them; +- empty request list; +- request count larger than the batch size; +- row groups already skipped by statistics or file-range pruning; +- unsupported DataFusion scalar types; +- Decimal128 physical encodings; +- equality, OR, and `IN` predicates; +- no false negatives compared with the existing scalar path. + +For malformed or failed reads, verify that DataFusion retains the affected row +group unless another valid statistic independently proves it can be skipped. + +## Object-store benchmark + +Add an instrumented `ObjectStore` wrapper that records: + +- `get_range` calls; +- `get_ranges` calls; +- number of ranges; +- bytes requested; +- maximum concurrent requests; +- injected latency; +- wall-clock duration. + +Benchmark at least: + +- latency: 0, 1, 5, 10, 25 ms; +- row groups: 16, 128, 1024; +- literal columns: 1, 2, 8; +- Bloom length present and absent; +- Bloom hit rate: 0%, 10%, 100% of groups retained; +- batch sizes: 1, 8, 16, 32, 64. + +Report: + +- Bloom-load wall time; +- time to first output batch; +- total query wall time; +- range-request count; +- bytes scanned; +- peak concurrent requests; +- peak memory attributable to loaded filters. + +Include three storage modes: + +1. in-memory store with injected latency, for deterministic scaling; +2. local filesystem, to detect overhead regressions; +3. an optional S3-compatible benchmark, for external validation. + +The benchmark should compare: + +- current scalar serial reads; +- DataFusion worker-pool prototype, if built; +- parquet-rs batched ranges. + +## Expected outcomes + +For latency-dominated Bloom loading with enough requests, bounded/batched I/O +should materially reduce the Bloom phase. A realistic target is: + +- at least 4x faster Bloom loading at 10 ms injected latency with 128 row + groups, 2 columns, and a batch/concurrency width of at least 8; +- no material regression at zero injected latency; +- unchanged bytes read for known-length filters, excluding intentional range + coalescing; +- bounded memory and request concurrency. + +End-to-end gains will be smaller and workload-dependent. They should be +reported only for the target Bloom-effective workload and not generalized to +all selective Parquet queries. + +## Risks and mitigations + +### Object-store throttling + +Risk: excessive concurrency increases tail latency or triggers throttling. + +Mitigation: bounded batches, conservative defaults, and concurrency metrics. + +### Read amplification + +Risk: coalescing distant ranges transfers substantially more bytes. + +Mitigation: batch exact ranges through `get_byte_ranges` first. Add contiguous +range coalescing only with a maximum-gap policy and separate byte metrics. + +### Batch failure scope + +Risk: one `get_ranges` failure loses all filters in a batch. + +Mitigation: conservatively retain affected groups. Consider retrying failed +batches as smaller batches or scalar requests only if benchmarks justify the +extra complexity. + +### Memory pressure + +Risk: parallel reads materialize many encoded filters simultaneously. + +Mitigation: bounded batch size and prompt insertion/drop of temporary buffers. +Measure peak memory for large Bloom filters. + +### Metrics inflation + +Risk: multiple readers or retries register duplicate counters or count bytes +twice. + +Mitigation: prefer one-reader batch I/O, document retry accounting, and assert +request/byte metrics in tests. + +### Upstream dependency + +Risk: the clean design requires a parquet-rs API addition and release. + +Mitigation: develop the parquet-rs API first or maintain a short-lived +DataFusion prototype solely for benchmark evidence. Avoid duplicating private +Thrift Bloom-header parsing in DataFusion. + +## Acceptance criteria + +- Results are identical to the scalar implementation across supported types + and malformed/missing-filter cases. +- No row group is pruned because a Bloom read failed. +- Request concurrency and temporary memory are bounded. +- The target latency benchmark shows at least a 4x Bloom-load improvement. +- Zero-latency/local benchmarks show no material regression. +- Bytes and pruning metrics remain correct and the Bloom evaluation timer is + not double counted. +- `cargo fmt --all`, workspace clippy, and required extended tests pass in + DataFusion. +- parquet-rs unit, integration, formatting, and clippy checks pass for the + upstream API change. + +## PR strategy + +Prefer a staged contribution: + +1. DataFusion benchmark plus low-risk guard/timer cleanup. +2. parquet-rs batched Bloom-read API with dedicated tests and benchmarks. +3. DataFusion adoption of the batch API and end-to-end benchmark results. + +This split gives reviewers evidence before changing I/O behavior and keeps the +cross-repository API work independently reviewable.