Repository navigation
Conversation
## Which issue does this PR close? - Closes apache#24310. ## Rationale for this change Record batch memory accounting runs on hot execution paths. The current implementation materializes ArrayData for every array and allocates a hash set even for small batches, adding measurable overhead to queries that frequently update memory reservations. ## What changes are included in this PR? - Traverse Arrow arrays directly and recursively count their backing buffers without materializing ArrayData. - Track the first 16 buffer identities inline, then promote to a hash set for wider batches or counters spanning many batches. - Preserve shared-buffer deduplication and full buffer-capacity accounting semantics. - Add parity coverage against the previous ArrayData traversal for primitive, binary/view, list/view, fixed-size, struct, union, dictionary, map, and all legal run-end index layouts. - Add a focused Criterion benchmark across column counts, row counts, and primitive/list/struct layouts. Criterion point estimates from `cargo bench -p datafusion-common --bench record_batch_memory`, measured sequentially on upstream `main` and this PR on the same machine: ### Column count (8,192 rows, Int64) | Columns | main | this PR | Speedup | |---:|---:|---:|---:| | 1 | 81.999 ns | 10.645 ns | 7.70x | | 4 | 308.11 ns | 38.669 ns | 7.97x | | 16 | 1.2543 us | 173.55 ns | 7.23x | | 64 | 4.7457 us | 1.3914 us | 3.41x | ### Row count (4 Int64 columns) | Rows | main | this PR | Speedup | |---:|---:|---:|---:| | 1 | 299.20 ns | 39.939 ns | 7.49x | | 128 | 302.99 ns | 38.914 ns | 7.79x | | 8,192 | 307.64 ns | 39.740 ns | 7.74x | | 65,536 | 310.61 ns | 38.994 ns | 7.97x | ### Array layout (4 columns, 8,192 rows) | Layout | main | this PR | Speedup | |---|---:|---:|---:| | Primitive Int64 | 311.01 ns | 40.140 ns | 7.75x | | List of Int64 | 778.29 ns | 82.865 ns | 9.39x | | Struct of two Int64 fields | 1.0072 us | 113.70 ns | 8.86x | Each list row contains two Int64 values. Each struct column contains two Int64 child fields. ## Are these changes tested? Yes. - `cargo fmt --all -- --check` - `cargo clippy --all-targets --all-features -- -D warnings` - `cargo test -p datafusion-common utils::memory --lib` (10 passed) - `RUST_BACKTRACE=1 cargo test --profile ci --exclude datafusion-examples --exclude datafusion-benchmarks --exclude datafusion-cli --workspace --lib --tests --bins --features avro,json,backtrace,extended_tests,recursive_protection,parquet_encryption` - Focused Criterion comparison shown above ## Are there any user-facing changes? No API or behavior changes. This reduces CPU and allocation overhead in record batch memory accounting. AI assistance: OpenAI Codex assisted with implementation and test execution. I reviewed the change and its behavior end to end. Backport provenance: - Upstream PR: apache#24319 - Upstream commit: c7be84a - The upstream diff is unchanged, including its own tests and benchmark. - No fork adaptation is included in this commit. Local Arrow 59 validity-buffer regressions are deliberately deferred to the final additional-test section. - The body above is the original upstream description; its benchmark estimates and verification statements describe upstream, not this fork's measurements. Signed-off-by: discord9 <discord9@163.com>
## Which issue does this PR close? Addresses the shared accounting bug identified while reviewing [apache#25491](apache#25491 (comment)). Split out so it can be reviewed and backported independently of prepared-build reuse. ## Rationale for this change Hash joins reserve memory for their hash buckets, but omit the row-index vector used to link build rows with the same key. As a result, a join can allocate beyond its admitted budget and underreport its build memory. For example, 100 million build rows need an additional 400 MB for this vector that the memory pool currently does not see. ## What changes are included in this PR? The shared hash-map constructor now reserves the bucket estimate and row-index storage together before allocating either, and includes both in the build-memory metric. This covers ordinary CollectLeft and partitioned joins, using four bytes per row for the u32 representation and eight for u64. Perfect-hash array maps keep their existing accounting. ## What is the testing strategy for this PR? A regression gives the constructor one byte less than the complete budget and checks that it refuses the allocation without charging the pool or metric. It also verifies success at the exact limit and release on drop. The test fails against the original constructor and passes with this fix. All 533 hash-join tests, full workspace Clippy with all targets/features and warnings denied, formatting, and whitespace checks pass on this branch. The combined prepared-build branch also passed the extended workspace run with this constructor fix (11,995 Rust tests and 521 SQL test files; eight tests ignored). Local checks used Rust 1.98.1 and temporary source overrides for the requested public releases of Arrow 60.0.0, object_store 0.14.2, sqlparser 0.63.0, and compression-codecs 0.4.42 because the registry mirror lacked them. No dependency or lockfile changes are included. Ordinary-path benchmark comparisons for the combined branch, including the accounting fix, are reported in apache#25491; they are not an isolated performance claim for this fix. ## Are there any user-facing changes? Hash joins using these maps account for their row-index storage in memory limits and build-memory metrics. A budget that previously succeeded only because this allocation was omitted can now be rejected before allocation. Backport provenance: - Upstream PR: apache#25508 - Upstream commit: 43f567a - All upstream additions and deletions are retained, including its regression test. Only patch anchors and indentation are relocated: the fork still has inline map construction instead of upstream's new_join_hashmap helper. - No helper extraction or fork interface adaptation is included here. Those changes follow in the immediately next commit. - This original-only intermediate state cannot build test targets: the upstream test references that absent helper. Checks apply to the following adapted state, not retroactively to this commit. - The description above records upstream's verification, not new fork results. Signed-off-by: discord9 <discord9@163.com>
The fork constructs the join map inline, while upstream apache#25508 updates and regression-tests a new_join_hashmap helper. Extract that existing constructor and wire its caller, retaining the preceding upstream allocation/accounting logic unchanged. This supplies the missing interface without mixing it into the upstream backport. Upstream PR: apache#25508 Upstream commit: 43f567a This is a local fork adaptation, not an upstream-authored change. Verification of this adapted state: cargo fmt --all -- --check; full all-target/all-feature Clippy with -D warnings; datafusion-physical-plan nextest (1707 passed). The preceding original-only intermediate test target was not buildable and is not retrospectively claimed to have passed. Signed-off-by: discord9 <discord9@163.com>
…ashJoinExec (apache#25496) ## Which issue does this PR close? - Closes apache#25269. ## Rationale for this change A hash join reserves every build-side batch as it arrives, then copies all of them into a single batch without reserving the copy, while the input batches stay alive until the build finishes. Near the memory limit the process therefore holds about twice the build side while the memory pool sees it once, so instead of a `ResourcesExhausted` error the process can be killed by the OS. It also means nothing that reacts to pool pressure can rely on what a hash join reports. Measured on a 1.25 GB build side (TPC-H SF1 `lineitem`, 16 columns, as the build side): the pool was charged ~1.25 GB while peak RSS was ~2.46 GB. ## What changes are included in this PR? - `concat_build_batches`: reserves the copy before `concat_batches`, drops the input batches, then trims the reservation to what the concatenated batch retains (`get_record_batch_memory_size`), growing instead if the estimate was low. `build_mem_used` follows every grow and shrink. - Nothing extra is reserved when there is nothing to copy: concatenating a single batch is zero-copy, and for `Utf8View`/`BinaryView` columns only the views are copied while the data buffers stay shared with the inputs. - `try_create_array_map` becomes `array_map_key_range`: it only decides whether the perfect hash join applies and reserves the `ArrayMap`. Both the hash map and the `ArrayMap` paths now hand the batches to the helper by value, so the inputs are gone before the visited bitmap and the null-aware scope maps are allocated. - Join key arrays that do not share the buffers of the build batch (any on-expression that is not a plain column) are kept for the whole join and are now reserved too. This uses a new public `RecordBatchMemoryCounter::count_array`. This does not reduce memory usage: the build still peaks at about twice the build side, it is now counted. A memory-limited query that only fit because the copy was not counted can now fail with `ResourcesExhausted`. ## What is the testing strategy for this PR? - `join_build_concat_is_reserved` (CollectLeft and Partitioned, hash map and `ArrayMap`): fails with `ResourcesExhausted` at map + 1.5x inputs, passes at map + 3x inputs. It fails on `main` in all four variants. - `join_build_key_arrays_are_reserved` (hash map and `ArrayMap`): with a single build batch and an `a1 + 1` join key, the join fails at inputs + map + half the key array and passes with room for it; a plain column key passes at the lower limit. It fails when the reservation is removed. - Unit tests of the helper: output equals `concat_batches` in both orders, the copy is reserved and the reservation then matches the retained batch, a single batch is not reserved twice, view data buffers are not reserved twice. - Existing `single_partition_join_overallocation` / `partitioned_join_overallocation` are unchanged. ## Are there any user-facing changes? One public addition: `RecordBatchMemoryCounter::count_array`. Hash joins running close to their memory limit may now report `ResourcesExhausted` where they previously exceeded the limit unnoticed. Backport provenance: - Upstream PR: apache#25496 - Upstream commit: 21a3215 - Preserve upstream count_array, concat/accounting helpers and collector, plus all original tests. Resolve patch anchors and import wrapping for the fork. - The upstream rustdoc replacement's old line is absent from the fork; retain its added line at the corresponding existing helper without fabricating a deletion. Existing fork &bounds context is unchanged. - No fork candidate guard or missing Array import is mixed into this commit; those adaptations follow immediately. This original-only state has not been checked for buildability. Validate the adapted state, not this intermediate. - The original description's tests and benchmarks refer to upstream results. Signed-off-by: discord9 <discord9@163.com>
Import Array for the upstream concat estimator: this trait was already imported in the upstream parent but not in the older fork. Use the fork's existing supported-type candidate helper when checking the array-map key range, preserving upstream's eligibility requirement. Upstream PR: apache#25496 Upstream commit: 21a3215 This commit contains only fork adaptation, not upstream-authored changes. The adapted state compiled in full all-target/all-feature Clippy. Previously started common/physical-plan nextest also completed: 2289 passed. Signed-off-by: discord9 <discord9@163.com>
## Which issue does this PR close? Part of [Comet apache#6013](apache/datafusion-comet#6013). [Comet apache#6037](apache/datafusion-comet#6037) uses this API. The shared row-index accounting fix in [apache#25508](apache#25508) is included in the base branch. ## Rationale for this change Independent joins against the same build snapshot repeat preparation and retain separate hash tables. `CollectLeft` shares a build among partitions of one join, but cannot share it across independently created plans. For example, eight concurrent requests joining events to the same users snapshot can share one users lookup while probing their own event batches. Preparing that lookup once avoids seven redundant builds. Applications can replace the snapshot while existing consumers finish with the old one. This supports both Comet broadcast tasks and other DataFusion applications. Comet creates independent native joins for its executor tasks, so their existing builds can overlap across task slots. The intended benefit is sharing one retained build batch and hash table and avoiding repeated materialization and hashing work. Removing repeated builds does not imply a proportional latency improvement: cold consumers wait for preparation, then probe independently. Latency depends on scheduling, contention, build size, coordination, and whether a prepared build is already retained. This PR does not establish end-to-end Comet latency, process CPU-time, or RSS savings. ## What changes are included in this PR? `HashJoinExec::prepare_build` returns an immutable `PreparedHashJoinBuild` that compatible joins attach explicitly: ```rust let prepared = first_join .prepare_build(users_snapshot, durable_pool, config) .await?; let first = first_join.builder() .with_prepared_build(Arc::clone(&prepared)) .build()?; let second = second_join.builder() .with_prepared_build(prepared) .build()?; ``` Consumers share build rows, lookup storage, and the retained memory reservation. Probe progress, residual predicates, and dynamic filters remain independent. Preparation charges retained data and temporary working memory; errors and cancellation release unfinished work. Concatenation alignment is charged once per output buffer, so small input batches do not multiply that reservation. Supported builds use non-spilling `CollectLeft` inner joins, direct-column keys with matching types, and fixed-width or UTF-8 build columns. Attachment and execution check schema, keys, and null equality, replaces the unused build subtree with a schema placeholder, and preserves the handle through supported resets and projections. Prepared state is process-local and cannot be serialized. The caller owns snapshot identity, concurrent preparation, and eviction. Input buffers must survive producer cleanup, and the supplied pool must cover every consumer. Attach after physical optimization, supply fresh probe plans and dynamic-filter expressions, and retain a build handle while any associated filter outlives its plan. Compatibility checks cannot verify snapshot contents. ## What is the testing strategy for this PR? Fifteen focused tests cover equivalence with ordinary joins, both null-equality modes, empty and duplicate keys, batch boundaries, concurrent consumers with independent filters and predicates, plan rewrites, public key changes after attachment, admission failures, small-batch concatenation under a fixed memory limit, cancellation, and buffer/reservation lifetimes. The latest regressions exercise under-admission without shrinking the reservation, checked copy/scratch arithmetic, and IN-list buffer sharing across one/two keys and one/two input batches. Existing serialization coverage rejects prepared state; the API example is a doctest. The benchmark checks outputs across 24 cases. Prepared copy and scratch sizing retain checked arithmetic. A concat output that exceeds its admitted bound returns an error in every build profile before shrinking the reservation. Numeric IN-list membership shares the retained batch's key buffers; charging its full array size again would duplicate that payload charge. General array/schema metadata and consumer-local filter expressions are outside the buffer reservation. Validation of [8a964a8](apache@8a964a8) passed: all 15 focused prepared-build tests in the normal `ci` profile; `cargo fmt --all`; full workspace Clippy with all targets/features and `-D warnings`; the complete repository lint suite, including Rust and HTML documentation builds; and the extended workspace suite (12,159 Rust tests and all 524 SQL logic files, with 8 Rust tests ignored). All 15 prepared-build tests also passed in the optimized `release-nonlto` profile. The public-key-mutation regression was verified to fail before restoring execution-time validation: the ordinary join returned its two expected rows while the stale prepared lookup returned none. With the guard restored, the mutated descriptor is rejected with a planning error. Local validation used Rust 1.98.1 and temporary source overrides for the official Arrow 60.0.0 (`ef1fa157`), object_store 0.14.2 (`279572ea`), and sqlparser 0.63.0 (`85b1a6f2`) releases. Dependency overrides and generated lockfile changes are not included. The rebased head has not been benchmarked. ```bash cargo bench -p datafusion-physical-plan --features test_utils --profile release-nonlto --bench prepared_hash_join ``` ## Are there any user-facing changes? Opt-in Rust APIs and an `EXPLAIN` marker for prepared builds. Ordinary SQL planning and join selection are unchanged. Backport provenance: - Upstream PR: apache#25491 - Upstream commit: acf5c89 - Preserve the upstream prepared-build API, benchmark and original tests. The three new Rust files and serialization test are upstream-exact. - Resolve conflicts against the older fork's existing owner/codec structure. Upstream-only neighboring null-mark/ProbeCompletion/RightAnti documentation and null-aware parameter context have no counterpart; do not fabricate them. The serialization guard uses self.prepared_build before encoding children, retaining upstream's rejection and message. - The immediately following commit contains the older-fork probe-state and collector-call adaptations. Additional local tests are deferred to the end. - The upstream description's tests and benchmark measurements are upstream results, not a claim that this original-only intermediate state was tested. Signed-off-by: discord9 <discord9@163.com>
Use the older fork's execution-local probe counters and NULL-aware atomic flags instead of upstream-parent fields unavailable here; remove the absent collector argument. Prepared eligibility remains non-null-aware CollectLeft INNER only. Preserve the ordinary join's existing semantics. Document the actual fork ownership and probe state, keep initializer/drop ordering aligned, and relocate the unchanged apache#25508 test after test imports. No additional local regression tests are included. Upstream PR: apache#25491 Upstream commit: acf5c89 This is a local fork adaptation. Compilation will be checked on the final adapted series; no per-intermediate test-pass claim is made. Signed-off-by: discord9 <discord9@163.com>
## Why are the changes needed? ### Which issue does this PR close? Part of apache#10586. This fixes the unsigned string/binary byte-array case; it does not claim to resolve every Parquet ordering issue. It is also the correctness prerequisite for apache#24526. ### Rationale for this change A Parquet scan can skip a file, row group, or page when its statistics prove that no row can satisfy the filter. If those statistics use a different comparison order from the query, that proof is invalid: DataFusion can silently discard a row that should be returned. For example, suppose a Parquet row group contains `'aé'`, `'az'`, and `'b'`: ```sql SELECT s FROM t WHERE s = 'az'; -- Expected: one row containing 'az' ``` Parquet's deprecated byte-array `min`/`max` fields use signed comparison, whereas Arrow compares strings using unsigned UTF-8 bytes. The same values therefore have two different orders: | Comparison | Values in ascending order | Minimum / maximum | | --- | --- | --- | | Legacy signed-byte order | `aé < az < b` | `aé` / `b` | | Arrow's unsigned-byte order | `az < aé < b` | `az` / `b` | The first UTF-8 byte of `é` is `0xC3`: it sorts before `z`'s `0x7A` as a signed byte, but after it as an unsigned byte. If DataFusion interprets the legacy interval `['aé', 'b']` using Arrow's ordering, it sees `'az' < 'aé'` and can incorrectly skip the row group. Merely checking that `min <= max` does not help: the two reported endpoints are still in ascending order. The newer `min_value`/`max_value` fields and page-index bounds also need an ordering declaration that the reader understands. A missing or unknown `column_orders` entry is not enough to justify assuming Arrow's ordering. The [Parquet statistics definition](https://github.com/apache/parquet-format/blob/master/src/main/thrift/parquet.thrift) and [logical-type ordering rules](https://github.com/apache/parquet-format/blob/master/LogicalTypes.md) describe these distinctions. This bug is independently observable with an ordinary equality filter; it does not require a large `IN` list. ## What changes were proposed in this PR? ### What changes are included in this PR? The change makes a recognized comparison order a prerequisite for using unsigned byte-array bounds. At the point where Parquet metadata becomes Arrow statistics, DataFusion checks whether the column's physical/logical type and footer establish the expected unsigned order. It also rejects row-group bounds taken from the deprecated signed-order fields. When that evidence is missing, min/max is reported as unknown, so pruning keeps potentially matching data instead of guessing. The rule is applied at the granularity where the statistics are used. An unsafe row group prevents DataFusion from claiming a trustworthy bound for the whole file, but it does not make other row groups' valid bounds unusable. Page-index bounds are checked against the footer independently. The same safeguards cover static and runtime row-group pruning, as well as the inverse predicates used to decide whether every row already satisfies a filter. Null counts and unrelated columns' statistics remain available. The row-group pruning API gains a metadata-aware entry point so callers can supply the footer needed for this decision. The existing entry point remains source-compatible and behaves conservatively when that information is unavailable. Signed logical types such as decimals retain their existing behavior. ### Are there any user-facing changes? Queries no longer discard matching data because of these untrustworthy byte-array bounds. Older files, or files with an unrecognized ordering, may require more scanning. Modern files with trustworthy bounds retain min/max pruning. There is no file-format change or breaking public API change. ## How was this PR tested? ### Are these changes tested? The regression uses actual serialized Parquet files containing the example above, with modern, deprecated, missing-order, and unknown-order metadata. It checks that the matching `az` row survives file, row-group, runtime, and page pruning, while valid statistics can still eliminate unrelated data. On unchanged Apache `f1f0449a`, the adapted equality regression fails because the deprecated-order case loses `az`; the modern-statistics control passes. The dedicated Parquet-crate run passed 231 unit tests and four doctests. Its seven focused ordering tests also cover mixed safe/unsafe row groups, null counts, fixed-length binary and UUID, signed decimal, and logical types with undefined ordering. Formatting, all-targets/all-features Clippy with warnings denied, and `./dev/rust_lint.sh` passed. The extended workspace run passed 10,666 Rust tests, with eight ignored, and all 503 SQL-logic files. The existing metadata benchmark was run on Apache `f1f0449a` and this patch using the same valid modern-footer fixture. Across nine full-statistics cases there was no material regression; the largest case, with 256 columns and 128 row groups, measured 1.330 ms before and 1.332 ms after. Both runs used Rust 1.97.0, `release-nonlto`, 20 samples, and separate build directories on an Apple M5 Max. <details> <summary>Validation commands</summary> ```sh cargo test --locked --profile ci -p datafusion-datasource-parquet cargo bench --locked --profile release-nonlto -p datafusion-datasource-parquet \ --bench parquet_metadata_statistics -- \ metadata_full --sample-size 20 --warm-up-time 0.5 --measurement-time 1 --noplot RUST_BACKTRACE=1 cargo test --locked --profile ci \ --exclude datafusion-examples --exclude datafusion-benchmarks --exclude datafusion-cli \ --workspace --lib --tests --bins \ --features avro,json,backtrace,extended_tests,recursive_protection,parquet_encryption ``` </details> Backport provenance: - Upstream PR: apache#24525 - Upstream commit: 26b40dd - Upstream self-tests and all additions/removals retained; only unavoidable fork patch context resolved. - Any fork-specific adaptation is kept in a separate following commit. Signed-off-by: discord9 <discord9@163.com>
## Why are the changes needed? ### Which issue does this PR close? Related to apache#8668 and apache#8609; follows apache#24074 and the merged statistics-ordering fix apache#24525. ### Rationale for this change Queries often select a sparse set of string identifiers. Parquet min/max statistics can make these queries much cheaper by ruling out row groups or pages that cannot contain any requested identifier. For example, this query asks for 21 IDs, spaced ten apart: ```sql SET datafusion.execution.parquet.max_in_list_size = 1024; SELECT * FROM events WHERE customer_id IN ( 'id000', 'id010', 'id020', 'id030', 'id040', 'id050', 'id060', 'id070', 'id080', 'id090', 'id100', 'id110', 'id120', 'id130', 'id140', 'id150', 'id160', 'id170', 'id180', 'id190', 'id200' ); ``` A row group whose values fall between `id003` and `id007` cannot contain a match. The default pruning limit is 20, so this list is not eligible for the `IN` min/max rewrite unless the caller raises the limit. apache#24074 made that limit configurable. With a raised limit, DataFusion can already reject this row group, but it does so by constructing a growing expression tree resembling: ```text (min <= id000 AND id000 <= max) OR (min <= id010 AND id010 <= max) OR ... ``` For hundreds or thousands of identifiers, building and evaluating that tree can become expensive in its own right. Replacing the list with one enclosing range, `[id000, id200]`, would be cheaper, but would lose the gaps: that broad range overlaps `[id003, id007]` even though none of the requested IDs is present there. The aim is to keep the useful pruning precision of the existing per-value checks while making large lists cheaper to represent and evaluate. In the included local microbenchmark, evaluating 1,024 values against 4,096 intervals falls from 68.8 ms to 0.198 ms. This measures pruning work only, not end-to-end query speedup. ## What changes were proposed in this PR? ### What changes are included in this PR? Eligible large string lists are stored as a sorted, deduplicated set of values inside the pruning predicate, rather than expanded into one comparison branch per value. For each inclusive statistics interval, DataFusion finds the first requested value at or after the interval's minimum, then checks whether that value is also at or before its maximum. In the example, the first requested ID at or after `id003` is `id010`. Since `id010 > id007`, the row group can be skipped. An interval such as `[id019, id021]` must be kept because it could contain `id020`. This takes a binary search per interval after sorting the values once per constructed predicate, and the expression tree no longer grows with the number of IDs. The result remains a conservative pruning decision. An overlapping interval means only that a match is possible; the original `IN` expression still performs exact row filtering. The original literal information also remains available to other pruning mechanisms, including Bloom filters. Inverted or unusable bounds yield UNKNOWN. A single known bound can still exclude the entire domain; otherwise, incomplete bounds yield UNKNOWN and keep the container eligible for reading. **Limits and scope** For a nonempty input list of `N` values, before this optimization deduplicates it, and configured cap `C = datafusion.execution.parquet.max_in_list_size`: | Condition | Min/max pruning behavior | | --- | --- | | `N <= min(20, C)` | Existing per-value rewrite | | `20 < N <= C`, positive, non-null literal strings on a direct string column | Compact sorted-domain lookup | | `20 < N <= C`, other lists | Existing per-value rewrite, including non-string lists, `NOT IN`, and NULL-containing lists | | `N > C` | This `IN` min/max rewrite is skipped; it does not disable all pruning | The compact path supports `Utf8`, `LargeUtf8`, `Utf8View`, and dictionary-encoded string columns. Unsupported expressions keep their existing handling. The default cap remains 20, and `C = 0` disables this `IN` min/max rewrite. Other predicates and literal/containment pruning, including Bloom filters, remain available. Raising the cap can still produce expensive comparison trees for lists that do not use the compact path. The lower threshold of 20 preserves the existing expression shape and limits this PR's scope; it is not a measured performance crossover. Page-index pruning now receives the same configured cap as row-group pruning, so raising the limit can benefit both. An interval search requires bounds in the same comparison order as the query. The compact path reads the same statistics adapters as the existing per-value path. The already-merged apache#24525 masks untrusted Parquet byte-array bounds. For other statistics sources, `PruningStatistics` requires conservative bounds in Arrow's comparison order; arbitrary provider bounds are trusted rather than validated by this expression. ### Are there any user-facing changes? Users who raise `datafusion.execution.parquet.max_in_list_size` get cheaper min/max pruning for eligible large string lists, and page pruning now honors that setting. The configuration default, exact query results, and existing public APIs are unchanged. This is a focused optimization for literal string lists, not a general rewrite of every large `IN` expression. ## How was this PR tested? ### Are these changes tested? The pruning-crate suite passed 95 tests. The standalone Parquet regressions use lists of 20, 21, 256, and 1,024 values and check both exact query results and scan/pruning metrics. They cover gaps inside the list's enclosing range, row-group pruning, page-only pruning, and the default and zero-cap controls. Two correctness regressions exercise the less obvious interactions. A direct physical-source test uses `NOT IN (..., NULL)`, row-filter pushdown, and `LIMIT 1`, so logical optimizer folding cannot hide an incorrect decision to bypass filtering. A real-file test combines this PR with apache#24525 and verifies that compact page pruning cannot lose a matching row when the footer's ordering is missing or unknown. Against unchanged Apache `f1f0449a`, the positive row-group/page tests fail as expected; the `NOT IN (..., NULL)` control passes. The benchmark compares the actual raised-cap `IN` path on Apache `f1f0449a` and this patch, using separate build directories and checking that both return the same nontrivial pruning results. Local results on an Apple M5 Max (18 CPUs, 128 GiB), Rust 1.97.0, `release-nonlto`, 20 samples: | Values | Predicate construction, main → PR | Evaluate 4,096 intervals, main → PR | | --- | --- | --- | | 20 | 31.1 → 27.4 µs | 177.6 → 165.7 µs | | 21 | 30.5 → 3.52 µs | 183.8 → 98.1 µs | | 256 | 356.6 → 21.4 µs | 5.53 → 0.144 ms | | 1,024 | 1.466 → 0.081 ms | 68.8 → 0.198 ms | A balanced explicit OR tree is included as another comparison: at 1,024 values it takes 8.03 ms to evaluate the same intervals. These measurements isolate pruning overhead; no end-to-end workload improvement is claimed. The latest documentation/comment update passed configuration-doc generation, formatting, all-targets/all-features Clippy with warnings denied, and `./dev/rust_lint.sh` (including the workspace Rust documentation check). Focused validation also passed all 3 Parquet IN-list integration tests and all 9 Parquet statistics-ordering tests. The CI follow-up in [1a2b38c](apache@1a2b38c) updates the `SHOW ALL VERBOSE` snapshot to match the revised `max_in_list_size` help text. The stale row caused both the cargo test and benchmark verification jobs to fail. After the correction, the targeted `information_schema.slt` test and full extended suite passed: 10,796 Rust tests (8 ignored) and all 505 SQL logic files. Formatting, Clippy with all targets and features and warnings denied, and `./dev/rust_lint.sh` also passed. <details> <summary>Validation commands</summary> ```sh cargo test --locked --profile ci -p datafusion-pruning cargo test --locked --profile ci -p datafusion \ --test parquet_integration string_in_list_pruning cargo bench --locked --profile release-nonlto -p datafusion-pruning \ --bench string_in_list_pruning -- \ --sample-size 20 --warm-up-time 0.5 --measurement-time 1 --noplot RUST_BACKTRACE=1 cargo test --locked --profile ci \ --exclude datafusion-examples --exclude datafusion-benchmarks --exclude datafusion-cli \ --workspace --lib --tests --bins \ --features avro,json,backtrace,extended_tests,recursive_protection,parquet_encryption ``` </details> Backport provenance: - Upstream PR: apache#24526 - Upstream commit: 9ad1217 - Upstream self-tests and all additions/removals retained; only unavoidable fork patch context resolved. - Any fork-specific adaptation is kept in a separate following commit. Signed-off-by: discord9 <discord9@163.com>
## Which issue does this PR close? <!-- We generally require a GitHub issue to be filed for all bug fixes and enhancements and this helps us generate change logs for our releases. You can link an issue to this PR using the GitHub syntax. For example `Closes apache#123` indicates that this PR will close issue apache#123. --> - Part of apache#24711. ## Rationale for this change Large string NOT IN lists create long expression chains for Parquet pruning. These expressions are expensive to build and evaluate. <!-- Why are you proposing this change? If this is already explained clearly in the issue then this section is not needed. Explaining clearly why changes are proposed helps reviewers understand your changes and offer better suggestions for fixes. Please explain the problem you are trying to solve in terms of the user-visible behavior, rather than the implementation. For example, "The code in `foo.rs` doesn't handle nulls" is a symptom of the implementation. "COUNT(DISTINCT) returns wrong results when the column contains nulls" is the user-visible problem. --> ## What changes are included in this PR? - Use a compact sorted domain for large, non-null string NOT IN lists. - Preserve the existing behavior for lists containing NULL. - Add unit tests and Parquet row-group and page-pruning tests. - Add a mixed-container regression test. - Extend the existing benchmark and configuration documentation. - In a local benchmark with 1,024 values, construction was about 17 times faster and evaluation was about 27 times faster. <!-- There is no need to duplicate the description in the issue here, but it is sometimes worth providing a summary of the individual changes in this PR. --> ## Are these changes tested? Yes <!-- We typically require tests for all PRs in order to: 1. Prevent the code from being accidentally broken by subsequent changes 2. Serve as another way to document the expected behavior of the code If tests are not included in your PR, please explain why (for example, are they covered by existing tests)? --> ## Are there any user-facing changes? No public API changes <!-- If there are user-facing changes then we may require documentation to be updated before approving the PR. If there are any breaking changes to public APIs, please add the `api change` label. --> Backport provenance: - Upstream PR: apache#24781 - Upstream commit: eaf7fff - Upstream self-tests and all additions/removals retained; only unavoidable fork patch context resolved. - Any fork-specific adaptation is kept in a separate following commit. Signed-off-by: discord9 <discord9@163.com>
## Which issue does this PR close? - Closes apache#24711. ## Rationale for this change apache#24781 added compact pruning for non-NULL `NOT IN` lists. Issue apache#24711 still needs support for NULL-containing lists. Large string lists can create expensive per-value pruning expressions. The compact path reduces this cost when `max_in_list_size` permits the rewrite. NULL values require special handling. Filter pruning treats FALSE and UNKNOWN equally, but full-match detection must keep these results separate. ## What changes are included in this PR? This PR makes these changes: - Supports compact pruning for large string `IN` lists that contain NULL. - Treats `NOT IN` with NULL as unable to match rows. - Treats all-NULL `IN` lists as unable to match rows. - Records whether a pruning predicate is safe for full-match inversion. - Uses the same configured list-size cap when it builds the inverse predicate. - Adds unit tests, Parquet integration tests, documentation, and benchmarks. ## What is the testing strategy for this PR? The tests cover positive and negative lists, all-NULL lists, nested `OR` expressions, configured caps, and Parquet full-match detection. These commands passed: ```shell cargo test -p datafusion-pruning cargo test -p datafusion-datasource-parquet cargo test -p datafusion --test parquet_integration string_in_list_pruning cargo fmt --all -- --check cargo clippy --all-targets --all-features -- -D warnings ``` Focused benchmarks also cover compact `IN` and constant-false `NOT IN` expressions with NULL. ## Are there any user-facing changes? Large string `IN` and `NOT IN` lists that contain NULL can now use compact pruning when the cap permits it. Query results and configuration defaults do not change. This PR adds two non-breaking `PruningPredicate` accessors: `max_in_list_size` and `can_be_inverted_for_full_match`. Backport provenance: - Upstream PR: apache#24954 - Upstream commit: 33028d5 - Upstream self-tests and all additions/removals retained; only unavoidable fork patch context resolved. - Any fork-specific adaptation is kept in a separate following commit. Signed-off-by: discord9 <discord9@163.com>
## Which issue does this PR close? - Closes apache#24709 ## Rationale for this change Compact pruning for large `IN` lists only supports string columns. Other ordered types use one min/max comparison for each literal. Large lists create large expression trees and make pruning more expensive. These types can use the same compact sorted-domain approach. ## What changes are included in this PR? This PR adds compact `IN` and `NOT IN` pruning for: - Variable-length binary types - Signed and unsigned integers - Decimal types - Date and time types - Timestamp and duration types - Dictionary-encoded forms of these types String and binary values share the byte-domain implementation. Primitive types use typed contiguous arrays. The compact path preserves NULL behavior and dictionary value NULLs. Unsupported types keep the existing per-value fallback. Floating-point types remain unsupported because of NaN and signed zero. `FixedSizeBinary` and nested types also remain unsupported. The default `max_in_list_size` value remains 20. ### Benchmarks With 1,024 literals and 4,096 statistics containers: - String `IN` evaluation is 37.8x faster. - String `NOT IN` evaluation is 21.4x faster. - Int64 `IN` evaluation is 118x faster. - Int64 `NOT IN` evaluation is 111x faster. - Predicate construction is 15.9x to 32.1x faster. Commands: ```shell cargo bench -p datafusion-pruning --bench string_in_list_pruning cargo bench -p datafusion-pruning --bench primitive_in_list_pruning ``` ## What is the testing strategy for this PR? The tests cover: - All supported type families - Both `IN` and `NOT IN` - NULL-containing and all-NULL lists - Dictionary and run-end-encoded values - Missing, inverted, and truncated statistics - Parquet row-group and page pruning - Per-value fallback for unsupported types - Result equality between compact pruning and the original filter The following commands passed: ```shell cargo test -p datafusion-pruning cargo test -p datafusion --test parquet_integration string_in_list_pruning cargo test -p datafusion-sqllogictest --test sqllogictests -- information_schema cargo clippy -p datafusion-pruning -p datafusion --all-targets --all-features -- -D warnings cargo fmt --all -- --check ``` ## Are there any user-facing changes? Large `IN` and `NOT IN` lists can use compact pruning for more ordered types. This behavior applies when `datafusion.execution.parquet.max_in_list_size` is set above 20. The default value and SQL results do not change. This PR does not change public APIs. Backport provenance: - Upstream PR: apache#25012 - Upstream commit: f3f420b - Upstream self-tests and all additions/removals retained; only unavoidable fork patch context resolved. - Any fork-specific adaptation is kept in a separate following commit. Signed-off-by: discord9 <discord9@163.com>
…c filters (apache#25579) The fix prevents one aggregate’s filter from discarding rows that another aggregate still needs in dynamic filters. ### Example: Query: `MIN(a), MAX(a), MIN(b), MAX(b)` ``` First batch: (1, NULL), (8, NULL) | v a has bounds [1, 8] b has no bounds yet | +---------+----------+ | | Before After | | a < 1 OR a > 8 filter = true | | v v Next batch: (2, 5), (3, 7) | | Both rows rejected Both rows accepted | | v v Result: Now b has bounds [5, 7] 1 8 NULL NULL | v Filtering can start: a < 1 OR a > 8 OR b < 5 OR b > 7 | v Correct result: 1 8 5 7 ``` The latter is correct 1. According to [Postgres](https://www.postgresql.org/docs/current/functions-aggregate.html): > - MIN: “Computes the minimum of the non-null input values.” > - MAX: “Computes the maximum of the non-null input values.” 2. According to [DataFusion](https://github.com/apache/datafusion/blob/6ff484a96d7eca4b1bf4a263d7190ada5af2343c/datafusion/sqllogictest/test_files/aggregate.slt#L5490-L5500) Executing `1, NULL, 3, 4, 5` expects `MIN = 1, MAX = 5` --------- Co-authored-by: Gabriel <45515538+gabotechs@users.noreply.github.com> Backport provenance: - Upstream PR: apache#25579 - Upstream commit: 0af6818 - Conflict resolution only supplies the missing test module/import context; the upstream logic, native test body and SQL assertions are preserved. - Original upstream tests are retained; fork-only adaptation, if any, follows separately. Signed-off-by: discord9 <discord9@163.com>
…he#25500) ## Which issue does this PR close? Follow-up to apache#25227 and [its discussion](apache#25227 (comment)). ## Rationale for this change Some lossless casts unnecessarily discard column statistics or fail to preserve strict ordering information. For example, UInt32 → Int64 currently requires both bounds to be exact to propagate statistics, even though every UInt32 value fits in Int64. Recognizing this conversion as globally safe preserves available statistics even when one bound is absent or bounds are inexact. ## What changes are included in this PR? - Recognize unsigned-to-signed widening casts: UInt8 → Int16 / Int32 / Int64, UInt16 → Int32 / Int64, and UInt32 → Int64. - Recognize Utf8 → Utf8View and Binary → LargeBinary / BinaryView. - Extend the existing `check_bigger_cast` / `is_bigger_cast` predicates to recognize these lossless conversions. ## What is the testing strategy for this PR? - Test values, nulls, integer extrema, and strict ordering properties. - Test empty values, Unicode, long strings, and non-UTF8 binary values. - Test statistics propagation with exact, inexact, and missing bounds. - Verify unsafe reverse conversions remain excluded. Validation on the latest revision: 25 relevant unit tests passed; one existing test is ignored. Formatting, all-target/all-feature Clippy, and the standard `rust_lint.sh` suite passed. ## Are there any user-facing changes? More lossless casts retain statistics and strict ordering information, enabling existing optimizer rules. Backport provenance: - Upstream PR: apache#25500 - Upstream commit: b172f73 - Retain the upstream matcher and its complete self-tests; only the absent fork example-doc context is adjusted. - Original upstream tests are retained; fork-only adaptation, if any, follows separately. Signed-off-by: discord9 <discord9@163.com>
Use the older fork's two-argument project_column_statistics_through_expr helper in the upstream regression test. This commit contains only the test helper adaptation; upstream production logic and tests are in the previous commit. Upstream PR: apache#25500 Upstream commit: b172f73 Signed-off-by: discord9 <discord9@163.com>
…25602) ## Which issue does this PR close? <!-- We generally require a GitHub issue to be filed for all bug fixes and enhancements and this helps us generate change logs for our releases. You can link an issue to this PR using the GitHub syntax. For example `Closes apache#123` indicates that this PR will close issue apache#123. --> - Closes apache#25291. - Alternative to apache#25292. ## Rationale for this change <!-- Why are you proposing this change? If this is already explained clearly in the issue then this section is not needed. Explaining clearly why changes are proposed helps reviewers understand your changes and offer better suggestions for fixes. Please explain the problem you are trying to solve in terms of the user-visible behavior, rather than the implementation. For example, "The code in `foo.rs` doesn't handle nulls" is a symptom of the implementation. "COUNT(DISTINCT) returns wrong results when the column contains nulls" is the user-visible problem. --> Avoid scanning redundant files/row-groups/pages in the probe side of hash joins, based on the values dictated by the build side. ## What changes are included in this PR? <!-- There is no need to duplicate the description in the issue here, but it is sometimes worth providing a summary of the individual changes in this PR. --> - wiring up PushdownStrategy::Map/HashTableLookupExpr to carry a representation of the build side values from a hash join, through - `KeyRangeBitmap` implementation, which maps the build-side values array into a finite-sized bitmap bucket, and can answer probing questions for certain ranges - extend `build_predicate_expression` to build the associated pruning expression from `HashTableLookupExpr` using the new `KeyRangeBitmapPruningExpr`, which implements `PhysicalExpr` on top of `KeyRangeBitmap` - also extend `build_predicate_expression` so that it now pushes down pruning for `CaseExprs`, since that also unlocks the partitioned hash-join scenario this pr targets ## What is the testing strategy for this PR? <!-- We typically require tests for all PRs in order to: 1. Prevent the code from being accidentally broken by subsequent changes 2. Serve as another way to document the expected behavior of the code Briefly describe how this PR is tested, and point to the specific tests you added. For example: 'This new feature is covered by the `sqllogictest` cases added in `foo.slt`'. If this PR does not add tests, explain why. For example, if the change is already covered by existing tests, please mention it. You should also check the `codecov` bot reply on this PR to confirm the changed code is exercised. --> A number of unit tests added, and one SLT added. Also verified on the query shape that motivated the original issue ```sql > copy (select i as k, random() as v from generate_series(0, 1999999) t(i)) to '/tmp/fact.parquet' stored as parquet options ('format.max_row_group_size' '1000'); +---------+ | count | +---------+ | 2000000 | +---------+ 1 row(s) fetched. Elapsed 0.093 seconds. > create external table fact stored as parquet location '/tmp/fact.parquet'; 0 row(s) fetched. Elapsed 0.008 seconds. > create table dim as select i as k from generate_series(0, 1999999) t(i) where i % 10000 < 200; 0 row(s) fetched. Elapsed 0.014 seconds. > select count(*), sum(v) from fact join dim on fact.k = dim.k; select count(*), sum(v) from fact join dim on fact.k = dim.k; select count(*), sum(v) from fact join dim on fact.k = dim.k; select count(*), sum(v) from fact join dim on fact.k = dim.k; +----------+--------------------+ | count(*) | sum(fact.v) | +----------+--------------------+ | 40000 | 20121.308621485692 | +----------+--------------------+ 1 row(s) fetched. Elapsed 0.031 seconds. +----------+--------------------+ | count(*) | sum(fact.v) | +----------+--------------------+ | 40000 | 20121.308621485692 | +----------+--------------------+ 1 row(s) fetched. Elapsed 0.017 seconds. +----------+-------------------+ | count(*) | sum(fact.v) | +----------+-------------------+ | 40000 | 20121.30862148569 | +----------+-------------------+ 1 row(s) fetched. Elapsed 0.011 seconds. +----------+-------------------+ | count(*) | sum(fact.v) | +----------+-------------------+ | 40000 | 20121.30862148569 | +----------+-------------------+ 1 row(s) fetched. Elapsed 0.011 seconds. > explain analyze select count(*), sum(v) from fact join dim on fact.k = dim.k; ... | | DataSourceExec: file_groups={12 groups: [[tmp/fact.parquet:0..1927333], [tmp/fact.parquet:1927333..3854666], [tmp/fact.parquet:3854666..5781999], [tmp/fact.parquet:5781999..7709332], [tmp/fact.parquet:7709332..9636665], ...]}, projection=[k, v], output_ordering=[k@0 ASC NULLS LAST], file_type=parquet, predicate=DynamicFilter [ k@0 >= 0 AND k@0 <= 1990199 AND hash_lookup ], dynamic_rg_pruning=eligible, pruning_predicate=k_null_count@1 != row_count@2 AND k_max@0 >= 0 AND k_null_count@1 != row_count@2 AND k_min@3 <= 1990199 AND k_null_count@1 != row_count@2 AND KEY_RANGE_BITMAP(k_min@3, k_max@0, 20000/1048576 buckets), required_guarantees=[], metrics=[output_rows=200.0 K, elapsed_compute=9.65ms, output_bytes=25.0 MB, output_batches=200, files_ranges_pruned_statistics=12 total → 12 matched, row_groups_pruned_statistics=2.00 K total → 200 matched, row_groups_pruned_bloom_filter=200 total → 200 matched, page_index_pages_pruned=200 total → 200 matched, page_index_rows_pruned=200.0 K total → 200.0 K matched, limit_pruned_row_groups=0 total → 0 matched, batches_split=0, bytes_processed=22.1 MB, bytes_scanned=2.1 MB, file_open_errors=0, file_scan_errors=0, files_opened=12, files_processed=12, num_predicate_creation_errors=0, predicate_evaluation_errors=0, pushdown_rows_matched=0, pushdown_rows_pruned=0, row_groups_pruned_dynamic_filter=0, predicate_cache_inner_records=0, predicate_cache_records=0, bloom_filter_eval_time=64.23µs, metadata_load_time=345.27µs, page_index_eval_time=1.33ms, row_pushdown_eval_time=36ns, statistics_eval_time=896.06µs, time_elapsed_opening=6.99ms, time_elapsed_processing=24.90ms, time_elapsed_scanning_total=224.99ms, time_elapsed_scanning_until_data=25.01ms, output_rows_skew=1.19%, scan_efficiency_ratio=9.7% (2.24 M/23.13 M)] | ... ``` ## Are there any user-facing changes? <!-- If there are user-facing changes then we may require documentation to be updated before approving the PR. If there are any breaking changes to public APIs, please add the `api change` label. --> --------- Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com> Co-authored-by: Jay Zhan <jayzhan211@gmail.com> Backport provenance: - Upstream PR: apache#25602 - Upstream commit: 4bb9f8d - Upstream additions/removals and self-tests retained; fork codec/import context is resolved separately. Signed-off-by: discord9 <discord9@163.com>
Local fork correctness changes on top of the twelve upstream backports: - treat computed NULL/TRY_CAST/CASE results conservatively for full-match predicates, and keep the inverse-row-group guard; - retain dynamic-filter producer reservations through snapshots and partitioned remapping; - make visited-bitmap charges execution-local and carry the accepted key-range-bitmap reservation with the derived lookup. These are local changes, not upstream PR content. Local regression tests are kept in the following commit. Signed-off-by: discord9 <discord9@163.com>
Add the local Arrow 59 all-valid, Parquet inverse-order/full-match, aggregate producer, prepared-build lease and execution-local bitmap regression coverage. These tests are intentionally last, after the upstream backports and local correctness changes. Signed-off-by: discord9 <discord9@163.com>
Record the final aggregate/bitmap plan metrics for the existing local regression, without changing product behavior. Signed-off-by: discord9 <discord9@163.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Which issue does this PR close?
No fork issue is closed automatically. This is a complete, reviewable backport of twelve Apache DataFusion PRs:
Stacked on #37 (
fix/25531-cast-constraint-propagation), the current GreptimeDB DataFusion pin. This PR supersedes the grouped history in #39; the old branch is preserved for audit and was not force-pushed.Rationale for this change
Bring the complete prepared hash-join, compact IN-list pruning, aggregate NULL-bound recovery, lossless-cast and key-range-bitmap stack to this fork without dropping upstream APIs, tests or benchmarks. The final code must preserve NULL, ordering and full-result semantics while restoring dual-key pruning.
This PR does not claim that the originally reported historical traces/logs JOIN slowdown is reproduced or fixed. The separate real-SQL experiments found the opposite on their fixtures; see the performance note below.
What changes are included in this PR?
The history is intentionally one upstream PR per commit, followed immediately by a separate fork adaptation where needed:
Arrayimport and existing perfect-hash candidate helper.No upstream self-test was moved into the local-test commit. Upstream authors, dates, full SHAs and PR URLs are preserved in the commit messages.
Are these changes tested?
8a03b98d6bda6e1c22c5573b9e850648d1bb441f;git diffagainst it is empty.cargo check --locked --offline --all-targetsfor the seven affected crates passed.cargo fmt --all -- --checkpassed; full all-target/all-feature Clippy passed; 4,047 nextest tests passed with 2 skipped; Parquet integration passed 228 tests.files_time_range_prunedcounters per lifecycle, exactly 120 physical hourly SSTs per table and clean owned shutdown.Performance note (separate from this PR's correctness gate)
Known baseline test issue
The workspace extended SQL run still has pre-existing signature-error wording mismatches on unchanged baseline
cc5534492d5408831395609cdafb6520c7d514b7; this PR does not claim that extended command is green. The earlier dynamic-row-group snapshot mismatch is resolved by the final SLT-only follow-up commit.Are there any user-facing changes?
Queries can benefit from restored aggregate bounds and discrete hash-join/IN-list pruning while preserving complete NULL/cast results. The complete upstream backports include prepared-join and pruning APIs. No persisted or wire-format migration is introduced here.
This is a Draft for review, not merge approval.