Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
305 changes: 295 additions & 10 deletions native/Cargo.lock

Large diffs are not rendered by default.

8 changes: 7 additions & 1 deletion native/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@

[workspace]
default-members = ["core", "spark-expr", "common", "proto", "jni-bridge", "shuffle", "operators", "comet-udf-sdk", "comet-test-udfs"]
members = ["core", "spark-expr", "common", "proto", "jni-bridge", "shuffle", "operators", "comet-udf-sdk", "comet-test-udfs"]
members = ["core", "spark-expr", "common", "proto", "jni-bridge", "shuffle", "operators", "comet-udf-sdk", "comet-test-udfs", "vendor/datafusion-physical-plan", "vendor/datafusion-pruning"]
# Crates under ../contrib are intentionally NOT workspace members. Core pulls them in as
# optional path dependencies when their corresponding contrib feature is enabled.
exclude = ["../contrib"]
Expand Down Expand Up @@ -86,6 +86,12 @@ iceberg = { git = "https://github.com/apache/iceberg-rust", rev = "af1da4c5bc861
iceberg-storage-opendal = { git = "https://github.com/apache/iceberg-rust", rev = "af1da4c5bc86178c38c1c6db0578bdd9fb7021e3", features = ["opendal-memory", "opendal-fs", "opendal-s3", "opendal-gcs", "opendal-oss", "opendal-azdls"] }
reqsign-core = "3"

# Backport DataFusion #25602 without changing Comet's Arrow/Iceberg dependencies.
# Keep both crates together; provenance and removal instructions are in vendor/README.md.
[patch.crates-io]
datafusion-physical-plan = { path = "vendor/datafusion-physical-plan" }
datafusion-pruning = { path = "vendor/datafusion-pruning" }

[profile.release]
debug = true
overflow-checks = false
Expand Down
83 changes: 83 additions & 0 deletions native/core/src/execution/operators/dynamic_filter/join/tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1146,6 +1146,89 @@ async fn broadcast_filter_reaches_parquet_reader_after_complete_build() {
assert_eq!(unclustered_enabled_filter_rows, 400);
}

#[tokio::test]
async fn bitmap_filter_prunes_parquet_row_groups_inside_build_bounds() {
for mode in [PartitionMode::CollectLeft, PartitionMode::Partitioned] {
let mut scan_bytes = Vec::new();
for enabled in [false, true] {
let mut config = SessionConfig::new()
.with_target_partitions(1)
.with_parquet_page_index_pruning(false);
// Isolate metadata pruning from filtering decoded rows or pages.
config.options_mut().execution.parquet.pushdown_filters = false;
let session = Arc::new(SessionContext::new_with_config(config));
let (_file, scan) = parquet_probe((0..400).collect(), &session, 100);
let filter = filtered_probe(&scan);
let schema = Arc::new(Schema::new(vec![Field::new("key", DataType::Int32, false)]));
let build = memory_exec(
[50, 350]
.into_iter()
.map(|key| {
RecordBatch::try_new(
Arc::clone(&schema),
vec![Arc::new(Int32Array::from(vec![key]))],
)
.unwrap()
})
.collect(),
);
let join =
single_key_join_plans(build, Arc::clone(&filter) as Arc<dyn ExecutionPlan>, mode);
let plan = PhysicalPlanner::apply_join_dynamic_filter(
Arc::new(join),
enabled,
session.copied_config().options(),
)
.unwrap();
let expected = RecordBatch::try_new(
plan.schema(),
vec![
Arc::new(Int32Array::from(vec![50, 350])),
Arc::new(Int32Array::from(vec![50, 350])),
],
)
.unwrap();
let output = collect(Arc::clone(&plan), session.task_ctx())
.await
.unwrap();
assert_eq!(
batches_to_sort_string(&output),
batches_to_sort_string(&[expected]),
"{mode:?}, enabled={enabled}"
);

// Every row group overlaps the build's [50, 350] bounds. Only the
// bitmap can exclude the interior groups [100, 199] and [200, 299].
// Comet's runtime join forces map membership rather than IN lists.
assert_eq!(
pruning_metric(&scan, "row_groups_pruned_statistics"),
if enabled { 2 } else { 0 },
"{mode:?}, enabled={enabled}"
);
assert_eq!(
filter.metrics().unwrap().output_rows().unwrap(),
if enabled { 200 } else { 400 }
);
if enabled {
assert_eq!(metric(&plan, "dynamic_filter_join_filters_attached"), 1);
}
scan_bytes.push(
scan.metrics()
.unwrap()
.sum_by_name("bytes_scanned")
.unwrap()
.as_usize(),
);
}
assert!(
scan_bytes[1] < scan_bytes[0],
"{mode:?}: bitmap pruning should avoid data reads: enabled={}, disabled={}",
scan_bytes[1],
scan_bytes[0]
);
}
}

fn limited_session(bytes: usize) -> (SessionContext, Arc<dyn MemoryPool>) {
let pool: Arc<dyn MemoryPool> = Arc::new(GreedyMemoryPool::new(bytes));
let runtime = RuntimeEnvBuilder::new()
Expand Down
91 changes: 91 additions & 0 deletions native/vendor/README.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,91 @@
<!--
Licensed to the Apache Software Foundation (ASF) under one
or more contributor license agreements. See the NOTICE file
distributed with this work for additional information
regarding copyright ownership. The ASF licenses this file
to you under the Apache License, Version 2.0 (the
"License"); you may not use this file except in compliance
with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing,
software distributed under the License is distributed on an
"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
KIND, either express or implied. See the License for the
specific language governing permissions and limitations
under the License.
-->

# DataFusion bitmap pruning backport

These two crates backport [DataFusion #25602](https://github.com/apache/datafusion/pull/25602)
to DataFusion 55.1.0. The native workspace patches both crates together, so hash joins and
Parquet pruning use the same bitmap type. All other DataFusion dependencies remain on the
existing locked release, as do Arrow, Parquet, and Iceberg.

The upstream implementation first landed at
`4bb9f8d3192f6c945f2bf933231380e60341af66`, whose dependency graph uses Arrow 60.
Comet's pinned Iceberg revision uses Arrow 59. This temporary backport makes the bitmap
available without also requiring an Iceberg migration or an unmerged Iceberg dependency.
It does not enable Comet's experimental join runtime filtering by default or add Delta scans.

## Provenance

The source archives are the published crates.io 55.1.0 packages, based on DataFusion release
commit `7d3835c71f30cbd3c3ae4041732267f1f453097a`. Their SHA-256 checksums match the
original entries in Comet's lockfile:

| Package | Archive SHA-256 |
| -------------------------- | ------------------------------------------------------------------ |
| `datafusion-physical-plan` | `1265d58e5bce07d154e642a51ff43033b576a6ae40d50a29ac2b9f311004eb52` |
| `datafusion-pruning` | `74b06b333405c05015836ed36cea02cc29d1e2ee8fb414a77532d8346fb56492` |

The crates retain their licenses, notices, normalized Cargo manifests, and original manifests.
Manifest license headers and the upstream Rust formatting configuration are included for
Comet's repository checks. Deprecation warnings are allowed in the physical-plan crate so
its pre-Rust-1.99 atomic API remains compatible with DataFusion 55.1.0's minimum Rust version.
Package-local lockfiles and Cargo cache markers are omitted; `native/Cargo.lock` controls the
build. [bitmap-backport.patch](bitmap-backport.patch) records every source and manifest change
relative to those archives.

## Adaptations to 55.1.0

- Preserve the four-argument `HashTableLookupExpr::new` API. The new
`with_pruning_bitmap` builder attaches the optional summary without changing other callers.
- Use 55.1.0's existing hash-join build path. It has no prepared-build API; skip the summary
whenever dynamic filters are not computed. Failure to reserve bitmap memory only disables
this optional optimization.
- Retain the merged PR's conservative handling of `CASE` without `ELSE`, which must preserve
implicit NULLs during Parquet full-match inference.
- Keep existing IN-list pruning unchanged. Port only the bitmap statistics rewrite and CASE
support, adapting the recursive call to the 55.1.0 signature. Recognize lossless casts from
unsigned integers to larger signed integers locally, since 55.1.0's cast helper predates
that support.
- Include upstream bitmap, expression-sharing, and memory-limit tests, plus regressions for
missing statistics and implicit NULLs. Comet's join test verifies that both supported native
hash-join modes skip interior row groups that min/max filtering alone cannot exclude.

## Validation

From `native/`, run:

```sh
cargo test -p datafusion-pruning --lib
cargo test -p datafusion-physical-plan --lib key_range_bitmap
cargo test -p datafusion-physical-plan --lib pruning_bitmap
cargo test -p datafusion-comet --lib dynamic_filter
```

The Comet regression is named `bitmap_filter_prunes_parquet_row_groups_inside_build_bounds`.
It checks exact join results, reader attachment, row groups pruned, decoded rows, and bytes
read with filtering enabled and disabled. Page and decode-time row filtering are disabled
in this test to isolate row-group pruning.

## Removal

When Comet can adopt a DataFusion release containing #25602 together with compatible Arrow
and Iceberg dependencies, remove both vendored crates, their workspace-member entries, and
the `[patch.crates-io]` entries in `native/Cargo.toml`, then update the lockfile. Retain the
Comet regression test. Do not update just one patched crate: the producer and pruning reader
must agree on the hash-lookup expression and bitmap implementation.
Loading
Loading