Repository navigation
Sync upstream (#524–#527): adopt upstream's file-id fix - #60
Merged
Merged
Conversation
The liquid parquet reader ignores
`datafusion.execution.parquet.bloom_filter_on_read`. Whenever a scan has
a pruning predicate, it loads and applies every remaining row group's
bloom filters, even with the option set to `false`.
```rust
let config = SessionConfig::new()
.set_bool("datafusion.execution.parquet.bloom_filter_on_read", false);
let ctx = SessionContext::new_with_config(config);
// register a parquet table `t` with bloom filters, then:
// SELECT * FROM t WHERE a = 2
```
With `ParquetSource`, this scan reads no bloom filters. Once the scan
goes through `LiquidParquetSource`, it reads them anyway, and
`row_groups_pruned_bloom_filter` reports row groups pruned by a filter
the session turned off.
## Why it matters
Each bloom filter is a separate ranged read per row group and predicate
column, and the reads are not cached across queries. Callers turn the
option off when bloom checks cannot pay for themselves — for example,
many small files where file-level pruning has already selected exactly
the files that match. Today they have no way to avoid those reads.
The option was honoured in the original opener and dropped in d7f0741
("remove some flags"). If leaving it out was deliberate, I'm happy to
hear the reasoning; otherwise this restores it.
## The fix
`LiquidParquetSource` already carries the session's
`TableParquetOptions`. Pass `global.bloom_filter_on_read` into the
morselizer and skip the bloom-filter load when it is off. As in
DataFusion's own opener, the remaining row groups are then counted as
matched by bloom filters, so the pruning metrics stay consistent between
the two sources.
## Tests
The existing `prunes_row_group_with_bloom_filter` test becomes a helper
parameterised on the option, and a new test covers the off case:
- `prunes_row_group_with_bloom_filter` (on): 1 row, 1 row group pruned,
1 matched.
- `skips_bloom_filter_when_bloom_filter_on_read_is_off` (off): 1 row, 0
pruned, 2 matched. The predicate alone still produces the right answer.
With the test applied to `main` and the fix left out:
```
$ cargo test -p liquid-cache-datafusion --lib bloom
test optimizers::tests::prunes_row_group_with_bloom_filter ... ok
test optimizers::tests::skips_bloom_filter_when_bloom_filter_on_read_is_off ... FAILED
assertion `left == right` failed
left: 1
right: 0
```
With the fix, `cargo test -p liquid-cache-datafusion` passes (63 tests).
…b#525) Setting `datafusion.execution.batch_size` to anything other than the cache's batch size makes liquid-cache scans fail or return the wrong rows. ```sql -- cache batch size is the default 8192 SET datafusion.execution.batch_size = 2145; SELECT id FROM t; -- errors SELECT id FROM t LIMIT 4290; -- succeeds, rows from the wrong offsets ``` `LiquidCacheLocalBuilder::build` pins the session batch size to the cache's, so the default configuration does not hit this. A host that changes it afterwards — per query, or by sizing batches to row width — does. ## Why `LiquidCacheReader` walks the row selection in windows of the session batch size, and numbers each window with `current_batch_id`. But that id is a cache chunk index: the cache read and the parquet fallback both turn it back into rows using the *cache* batch size. When the two sizes differ, batch id N names one range of rows to the reader and a different range to everything that resolves it. - Without a `LIMIT`, the scan runs off the end of the row group: `parquet fallback ended before batch N`. - With a `LIMIT` that stops it first, it returns rows from the wrong offsets and succeeds. - On a fully cached row group, a selection mask sized by the session batch size is applied to a chunk sized by the cache's. `arrow::compute::filter` only rejects a mask *longer* than its target, so a shorter one silently selects the wrong rows. The only guard was a `debug_assert_eq!` in `LiquidCacheReader::new`, so debug builds panic and release builds carry on. ## The fix Window the selection by the cached row group's batch size, and drop the session batch size from `LiquidCacheReaderConfig` and `LiquidRowGroupPlanner`, which no longer need it. Callers still get session-sized batches: `DataSourceExec` re-splits every source stream to `datafusion.execution.batch_size`, and `session_batch_size_still_bounds_the_batches_the_caller_receives` checks that. ## Tests `src/datafusion-local/src/tests/batch_size_alignment.rs` covers a narrowed and a widened session batch size, `LIMIT`, the fully cached path, multiple row groups and multiple partitions, all against a cache at the default 8192. On `main` with the fix left out, in release (the debug build stops at the assert): ``` $ cargo test --release -p liquid-cache-datafusion-local --lib batch_size_alignment test result: FAILED. 0 passed; 7 failed ... "parquet fallback read failed: Parquet error: parquet fallback ended before batch 1" ... narrowed_session_batch_size_with_limit_returns_aligned_rows: assertion `left == right` failed ``` With the fix, all 7 pass in debug and release, and `cargo test -p liquid-cache-datafusion -p liquid-cache-datafusion-local` passes. One case is not covered here: page-index pruning that starts the selection partway into a row group. A scan with page-index pruning plus a pushed-down predicate panics in `boolean_buffer_and_then` on `main` even with matching batch sizes, so it cannot be tested until that is fixed; that is a separate issue.
…afusion-contrib#526) A process that reads more than 65,536 distinct parquet files gets another file's cached data back. A cache key packs the file id into 16 bits (`ParquetArrayID`), but `LiquidCacheParquet::register_or_get_file` hands out file ids from a counter that only climbs. The 65,537th distinct `(object store, path)` gets id 65536, which narrows to 0, so every key it computes is a key of the first file. Reads then return the first file's arrays: - silently wrong rows when the column types match; - a panic in `CachedColumn::array_to_record_batch` (`RecordBatch::try_new(..).unwrap()`) when they don't; - and `contains` reports the key as cached, so the newer file never caches its own data. The narrowing is guarded only by `debug_assert!`, so release builds hit this with no signal. A long-running server reading a stream of new files reaches it in normal use. ## Repro `files_whose_keys_collide_do_not_share_entries` (in `src/datafusion/src/cache/mod.rs`) caches a batch for one file, registers 65,535 more, then reads the same batch through the 65,537th. On `main`, with the test alone: - debug: fails at `debug_assert!(file_id <= u16::MAX as u64)` (`id.rs:150`); - release: fails with "a file was served another file's cached data" — the read returns the first file's array. ## Fix Store the full file id (an "identity") with every cached entry and check it on every lookup. A mismatch reads as a miss, so the caller falls back to parquet. An insert replaces whatever held the key. - **Index:** each slot holds `(identity, entry)`. - **Disk:** every object gets an 8-byte identity trailer. Objects stay keyed by entry id. An eviction/flush/hydration that lands after the key changed hands leaves bytes naming the old identity, and the new owner's read of them misses, so these races end in a miss rather than in wrong data and need no extra coordination. The trailer follows the payload so the payload keeps its alignment; the disk budget is charged for it. Nothing reads objects across a restart (the index is in-memory), so there is no on-disk compatibility concern. - **DataFusion layer:** `CachedFile` already holds the full `u64` file id; `CachedColumn` now carries it and passes it with every call. The key width is unchanged: the index (`congee`) supports 8-byte keys only, so widening the key would mean replacing the index. ## API change `LiquidCache::insert`, `get`, `eval_predicate`, `prefetch`, `try_read_liquid`, `contains`, `read_entry` and `eval_predicate_on_entry` take an `identity: u64`. Callers whose keys are already unique can pass any constant; the README, example and study bench pass `0`. `read_entry` / `eval_predicate_on_entry` act on an entry the caller already holds, so they trust that entry and check the identity only for data read from disk. ## Cost 8 bytes per index slot and per disk object, and one integer comparison per lookup. ## Tests - New: the repro above, `an_entry_is_served_only_to_its_identity`, `disk_bytes_written_for_another_identity_are_a_miss`, `a_slot_is_visible_only_to_its_identity`. - Replaced: `test_cache_entry_id_new_panic_file_id` pinned the debug-only guard; `test_cache_entry_id_new_keeps_low_16_bits_of_file_id` pins the narrowing the identity check relies on. - Two disk-budget tests now size the budget to include the trailer. - Workspace tests, doctests and the shuttle tests pass; test inventory is otherwise unchanged. ## Not in this PR - After 65,536 files, two files that share a narrowed id and are both being read keep replacing each other's entries. Results are correct; caching just stops helping for that pair. - `register_or_get_file`'s map keeps one entry per distinct file forever. - Replacing a disk entry (by hydration, or now by another identity) does not release its disk reservation. That predates this change and is separate.
…sion-contrib#527) Closes datafusion-contrib#522. In-process mode (`liquid-cache-datafusion-local`) no longer pulls in the client/server network stack. Its normal dependency graph drops from 387 to 343 crates; `arrow-flight`, `tonic`, `axum`, `hyper`, `h2`, `reqwest` and `rustls` are all gone. ## Changes - **`liquid-cache-common`: new `rpc` feature, off by default.** It gates `pub mod rpc` and makes `arrow-flight` and `prost` optional. `liquid-cache-datafusion-client` and `liquid-cache-datafusion-server` enable it. - **`ExecutionMetricsResponse` moves to `liquid_cache_common::metrics`.** It is a plain serde struct and the only part of `rpc` the in-process crates used. `extract_execution_metrics` stays where it is, with the same signature. - **`liquid-cache-datafusion` no longer enables `object_store`'s `http` feature.** Nothing in that crate uses it, and it was what pulled in `reqwest`/`hyper`/`h2`/`rustls`, so gating `rpc` alone would leave those in place. `liquid-cache-datafusion-client` calls `object_store::parse_url_opts` and was relying on this transitively, so it now enables `http` itself. The server, benchmark and examples already did. - **`liquid-cache-common` gets `tokio` with `macros` as a dev-dependency.** Its `#[tokio::test]`s only compiled before because `tonic` enabled that feature. - **docs.rs builds `liquid-cache-common` with `rpc`**, and its README notes the feature. `prost` remains in the graph through `vortex`; that is outside this change. ## Breaking changes - **The metrics struct has a new path.** `liquid_cache_common::rpc::ExecutionMetricsResponse` is now `liquid_cache_common::metrics::ExecutionMetricsResponse`. The serialized JSON is unchanged. - **`rpc` is now opt-in.** Code that uses `liquid_cache_common::rpc` directly, or through the `common` re-exports of `liquid-cache` / `liquid-cache-datafusion` / `liquid-cache-datafusion-local`, needs `liquid-cache-common = { ..., features = ["rpc"] }`. - **No more transitive `http`.** Users of `liquid-cache-datafusion` or `-local` who registered `http(s)://` object stores and got `object_store`'s `http` feature through this crate need to enable it themselves. ## Testing - **Workspace checks.** `cargo test --workspace --exclude dev-tools` gives 236 passed, 0 failed. The test list is identical to `main` (239 entries). `cargo clippy --workspace --all-targets --all-features -D warnings`, `cargo fmt --check`, `cargo doc` and `cargo shear` are clean. - **Each crate built on its own.** `cargo check -p <crate> --all-targets` passes for every crate. Workspace feature unification would otherwise hide a missing feature declaration. - **Standalone in-process crate.** A crate outside the workspace depending only on `liquid-cache-datafusion-local`: - It builds and runs the `examples/datafusion-local.rs` flow on `nano_hits.parquet`; the second run is served from cache and `extract_execution_metrics` reports correctly. - `cargo tree` shows none of the crates listed above. - **Standalone client crate.** A crate depending only on `liquid-cache-datafusion-client` registers `http://` and `https://` object stores. - With the client's new `http` line removed, the same build fails at runtime with `feature for Http not enabled`, so that line is required. - **End-to-end client/server.** - `bench_server` with `clickbench_client` returns correct query results. - The example client/server reads an `https://` parquet file. - The admin `/execution_metrics` endpoint returns the same JSON keys as before.
…10-16 Brings in datafusion-contrib#524, datafusion-contrib#525, datafusion-contrib#526 and datafusion-contrib#527. datafusion-contrib#526 is upstream's fix for file-id aliasing in cache keys: each entry records the full file id, in the index and as a trailer on its disk object, and a lookup under any other id misses. It replaces the fork's own fix for the same bug, so the cache code takes upstream's version and the fork's is removed: identity-scoped disk keys, the reclamation layer built on them (DiskResidue, settle, reclaim_orphaned_disk), WriteIdentity, and file-id leasing (file_id.rs). Kept on top of upstream's version: - The index releases a removed or replaced entry's payload immediately rather than through crossbeam-epoch's deferred drop, which kept multi-megabyte arrays alive past the budget that stopped counting them. Ported onto upstream's slot as a cell the index empties. - leased_file_ids, file_ids_over_key_width and identity_mismatches on LiquidCacheParquet, which downstream exports as metrics. Without leasing, leased_file_ids counts every file ever registered rather than the files being read. Dropped with the reclamation layer: the fixes that release a disk copy displaced by hydration or overwritten in place. Upstream has that leak too; it is to be fixed there and synced. The four datafusion-local snapshots return to upstream's disk_evictions: 0 for that reason. datafusion-contrib#525 is the fork's batch-size alignment fix returning from upstream. batch_size_alignment.rs takes upstream's text and keeps the fork's two ignored pruned-scan tests and the page_rows helper they use.
![claude[bot]](https://avatars.githubusercontent.com/in/1236702?s=60&v=4){kind=small}
There was a problem hiding this comment.
The full diff was omitted from the review prompt. The whole patch was fetched with gh pr diff and read for all 35 files. CI checks were still pending when this review ran, so this approval does not rely on test results. Merge with a merge commit, as the PR description states.
Codecov Report❌ Patch coverage is 📢 Thoughts on this report? Let us know! |
📊 Benchmark ComparisonCurrent:
Compared Liquid vs DataFusionDefault on the same runner |
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
1 participant
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.
Syncs upstream datafusion-contrib#524, datafusion-contrib#525, datafusion-contrib#526 and datafusion-contrib#527. Merge with a merge commit, not a squash, so upstream's history enters
main.Main change: upstream's file-id fix replaces ours
Upstream datafusion-contrib#526 fixes file-id aliasing in cache keys (the 65,537th file being served the first file's data). It does this by storing the full file id with each entry, in the index and as a disk trailer. We already had our own, more complex fix for the same bug, so the cache code now takes upstream's version and ours is removed:
DiskResidue,settle,reclaim_orphaned_disk)WriteIdentityfile_id.rs)Kept on top of upstream
removed_and_replaced_entries_are_released_immediately.leased_file_ids,file_ids_over_key_width,identity_mismatchesonLiquidCacheParquet, which a downstream consumer exports as metrics. Meaning changes:leased_file_idscounts every file ever registered, not only the files being read.identity_mismatchescounts lookups only. The fork's version also counted a write that took over a key held by another file, so the exported value will be lower than before for the same workload.Dropped
mainhas the leak too. The fourdatafusion-localsnapshots go back to upstream'sdisk_evictions: 0for this reason.Other
batch_size_alignment.rs(fix(reader): read the cache in cache-sized batches datafusion-contrib/liquid-cache#525, our own fix coming back from upstream) takes upstream's text and keeps our two ignored pruned-scan tests.Verified
mainis missing. The tests that disappear from forkmainare all tests of the removed machinery.