Repository navigation
Accept DataFusion dynamic filters in VortexSource, with join membership lists - #10415
joseph-isaacs wants to merge 18 commits into
Conversation
Convert TopK and hash-join dynamic filters into live `vortex.dynamic` comparisons so the scan prunes zones as the filter tightens. Only `col op literal` bounds are tracked; anything else is relaxed to `true`, so the Vortex filter never drops rows the operator needs. Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk> Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01FL15XgcyJgZ6xkaMo8p5o3
Hash-join dynamic filters are complete by the time the probe side opens a file, so read their bounds once and apply them as static comparisons, but only when the file's min/max statistics suggest they keep at most half of the column's value range. Non-selective join bounds (e.g. TPC-H order and customer keys) otherwise add a row-level filter that costs more than it saves. Filters that are still changing (TopK) keep the live template. TPC-H SF1, DataFusion over Vortex, 30 samples per mode: geomean on/off went from 1.065 to 0.981; q07 0.70x, q21 0.71x. Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk> Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01FL15XgcyJgZ6xkaMo8p5o3
When a file's type differs from the table's, the expression adapter wraps a dynamic filter's columns in casts. The opener then treated the filter as an unpushable static filter and failed the query (ClickBench q23, `ORDER BY "EventTime" LIMIT 10`). Send every dynamic filter down the dynamic path, dropping any that can't be converted, and look through casts to the underlying column. Bounds on a cast column are only applied when the literal converts to the file type and back unchanged, so a lossy cast can never tighten the filter. Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk> Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01FL15XgcyJgZ6xkaMo8p5o3
Multi-column TopK filters have the form `a < v OR (a = v AND b < w)`, which previously yielded no bounds, leaving every row to pay for an always-true template (ClickBench q26 regressed 1.24x). Read bounds recursively: `=` gives both `<=` and `>=`, and an OR keeps, per column and direction, the looser bound when both sides have one, so the example yields `a <= v`. Incomparable values yield no bound. Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk> Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01FL15XgcyJgZ6xkaMo8p5o3
Partitioned hash joins publish `CASE route(key) WHEN i THEN bounds_i ... END`, which previously yielded no bounds, leaving probe-side files with an always-true filter. Bounds are now the hull over a CASE's results, with `false`/`NULL` branches (empty partitions) contributing nothing. When a still-changing filter already has bounds at file open, emit only the comparisons and columns those bounds use rather than the full template. Tidy the opener's conjunct split and share a values helper across the end-to-end tests, adding a hash-join case whose bounds are selective enough to be applied. Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk> Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01FL15XgcyJgZ6xkaMo8p5o3
The cached bounds are read on every evaluation of the Vortex filter but replaced only when the DataFusion filter changes, a few dozen times per query for TopK. Hold them in an ArcSwap instead of a mutex so concurrent reads don't contend; a refresh never replaces a newer generation. With four threads reading the same filter, a lookup drops from ~1.3 us to ~0.3 us. Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk> Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01FL15XgcyJgZ6xkaMo8p5o3
Report each converted dynamic filter with its DataFusion predicate, whether it is complete, and the resulting Vortex expression, plus the kept fraction estimated for complete filters. This makes it possible to see which scans actually receive TopK and join filters in a benchmark run. Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk> Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01XGkjCDMVSGtEWhEHHW5wGL
A constant list against an array of needles was answered with one equality comparison over the whole array per element, OR-ed together, so an `IN` list of a hundred keys cost a hundred passes over the column. For integer needles and lists of at least four non-null elements, sort the elements once and binary search per needle instead. Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk> Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01XGkjCDMVSGtEWhEHHW5wGL
Complete hash-join filters carry `col IN (keys)` next to their min/max bounds whenever the build side has few distinct keys. Convert those lists into a static `in_list` when they prune a meaningful share of the rows the bounds keep: a list that is dense in the bounded range adds nothing but a decode of the column. Partitioned joins route keys through a CASE, whose branch lists are unioned. Before a live filter has any bounds, only its leading column gets the dynamic comparison template. A TopK filter over several sort keys is a lexicographic disjunction whose hull only ever bounds the first key, so the other columns cost evaluations on every batch for nothing; on TPC-DS q10 that was 32 dynamic comparisons per batch. Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk> Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01XGkjCDMVSGtEWhEHHW5wGL
A misplaced `?` dropped every column that had bounds but no membership list, so complete join filters that only carry min/max bounds were never applied. Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk> Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01XGkjCDMVSGtEWhEHHW5wGL
Keeps `expect` out of the rstest case expressions, which are inlined into a test returning `Result`. Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk> Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01XGkjCDMVSGtEWhEHHW5wGL
Merging this PR will regress 1 benchmark
|
| Mode | Benchmark | BASE |
HEAD |
Efficiency | |
|---|---|---|---|---|---|
| ❌ | WallTime | starts_with_avx2[64, VarBin] |
54.7 µs | 62.9 µs | -13.09% |
| ⚡ | Simulation | i64_random_chunked[256] |
15,817.7 µs | 783.2 µs | ×20 |
| ⚡ | Simulation | i64_random[256] |
4,360.6 µs | 274 µs | ×16 |
Tip
Investigate this regression by commenting @codspeedbot fix this regression on this PR, or directly use the CodSpeed MCP with your agent.
Comparing ji/nifty-wright-ahdfl0 (83102a8) with develop (731a231)
Footnotes
-
359 benchmarks were skipped, so the baseline results were used instead. If they were deleted from the codebase, click here and archive them to remove them from the performance reports. ↩
…ilter Clippy's cognitive complexity limit rejected the closure that built a column's bounds and membership comparisons inline. Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk> Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01XGkjCDMVSGtEWhEHHW5wGL
Polar Signals Profiling ResultsLatest Run
Powered by Polar Signals Cloud |
Benchmarks: String Encoding 📖Commits: PR vortex / vortex-file-compressed / ms (1.014x ➖, 0↑ 0↓)
vortex / vortex-file-compressed / % (1.000x ➖, 0↑ 0↓)
|
Benchmarks: PolarSignals Profiling 📖Commits: PR datafusion / vortex-file-compressed / ns (1.003x ➖, 0↑ 0↓)
No file size changes detected. |
Benchmarks: FineWeb S3 📖Commits: PR How to read Verdict and Engines
datafusion / vortex-file-compressed / ns (0.984x ➖, 1↑ 1↓)
datafusion / vortex-compact / ns (1.032x ➖, 1↑ 0↓)
datafusion / parquet / ns (0.932x ➖, 1↑ 0↓)
duckdb / vortex-file-compressed / ns (0.995x ➖, 0↑ 0↓)
duckdb / vortex-compact / ns (1.061x ➖, 0↑ 1↓)
duckdb / parquet / ns (0.949x ➖, 0↑ 0↓)
|
Benchmarks: TPC-H SF=10 on NVME 📖Commits: PR How to read Verdict and Engines
datafusion / vortex-file-compressed / ns (1.007x ➖, 1↑ 0↓)
datafusion / vortex-compact / ns (1.005x ➖, 1↑ 1↓)
datafusion / parquet / ns (1.000x ➖, 0↑ 0↓)
duckdb / vortex-file-compressed / ns (0.995x ➖, 0↑ 0↓)
duckdb / vortex-compact / ns (0.995x ➖, 0↑ 0↓)
duckdb / parquet / ns (1.004x ➖, 0↑ 0↓)
No file size changes detected. |
Benchmarks: Clickbench on NVME 📖Commits: PR How to read Verdict and Engines
datafusion / vortex-file-compressed / ns (0.984x ➖, 3↑ 0↓)
datafusion / vortex-compact / ns (0.954x ➖, 3↑ 0↓)
datafusion / parquet / ns (0.998x ➖, 0↑ 0↓)
duckdb / vortex-file-compressed / ns (1.009x ➖, 1↑ 3↓)
duckdb / vortex-compact / ns (1.006x ➖, 0↑ 0↓)
duckdb / parquet / ns (0.993x ➖, 1↑ 0↓)
No file size changes detected. |
Benchmarks: FineWeb NVMe 📖Commits: PR How to read Verdict and Engines
datafusion / vortex-file-compressed / ns (0.974x ➖, 0↑ 0↓)
datafusion / vortex-compact / ns (0.989x ➖, 2↑ 1↓)
datafusion / parquet / ns (1.005x ➖, 0↑ 0↓)
duckdb / vortex-file-compressed / ns (1.011x ➖, 4↑ 2↓)
duckdb / vortex-compact / ns (1.021x ➖, 1↑ 3↓)
duckdb / parquet / ns (0.991x ➖, 0↑ 0↓)
No file size changes detected. |
Benchmarks: Compression 📖Commits: PR vortex / vortex-file-compressed / ns (0.984x ➖, 0↑ 0↓)
vortex / vortex-file-compressed / bytes (1.000x ➖, 0↑ 0↓)
vortex / vortex-file-compressed / ratio (0.989x ➖, 0↑ 0↓)
vortex / parquet / ns (1.000x ➖, 0↑ 0↓)
vortex / parquet / bytes (1.000x ➖, 0↑ 0↓)
vortex / arrow-ipc / ns (1.027x ➖, 0↑ 5↓)
vortex / arrow-ipc / bytes (1.000x ➖, 0↑ 0↓)
|
Benchmarks: TPC-H SF=1 on NVME 📖Commits: PR How to read Verdict and Engines
datafusion / vortex-file-compressed / ns (1.027x ➖, 0↑ 2↓)
datafusion / vortex-compact / ns (1.029x ➖, 0↑ 2↓)
datafusion / parquet / ns (1.003x ➖, 0↑ 0↓)
duckdb / vortex-file-compressed / ns (1.016x ➖, 0↑ 0↓)
duckdb / vortex-compact / ns (0.994x ➖, 1↑ 0↓)
duckdb / parquet / ns (1.017x ➖, 0↑ 1↓)
No file size changes detected. |
Benchmarks: Clickbench Sorted on NVME 📖Commits: PR How to read Verdict and Engines
datafusion / vortex-file-compressed / ns (0.954x ➖, 1↑ 0↓)
datafusion / vortex-compact / ns (0.913x ➖, 2↑ 0↓)
datafusion / parquet / ns (0.995x ➖, 0↑ 0↓)
duckdb / vortex-file-compressed / ns (1.020x ➖, 1↑ 2↓)
duckdb / vortex-compact / ns (1.017x ➖, 0↑ 2↓)
duckdb / parquet / ns (0.967x ➖, 1↑ 0↓)
File Size Changes (200 files changed, +0.0% overall, 96↑ 104↓)
Totals:
|
Benchmarks: TPC-H SF=1 on S3 📖Commits: PR How to read Verdict and Engines
datafusion / vortex-file-compressed / ns (0.988x ➖, 1↑ 1↓)
datafusion / vortex-compact / ns (1.050x ➖, 0↑ 5↓)
datafusion / parquet / ns (0.999x ➖, 1↑ 1↓)
duckdb / vortex-file-compressed / ns (1.004x ➖, 0↑ 1↓)
duckdb / vortex-compact / ns (0.980x ➖, 1↑ 0↓)
duckdb / parquet / ns (0.952x ➖, 0↑ 1↓)
|
Benchmarks: Appian on NVME 📖Commits: PR How to read Verdict and Engines
datafusion / vortex-compact / ns (1.004x ➖, 0↑ 0↓)
datafusion / parquet / ns (1.002x ➖, 0↑ 0↓)
duckdb / vortex-compact / ns (0.991x ➖, 0↑ 0↓)
duckdb / parquet / ns (0.986x ➖, 0↑ 0↓)
No file size changes detected. |
Benchmarks: TPC-DS SF=1 on NVME 📖Commits: PR How to read Verdict and Engines
datafusion / vortex-file-compressed / ns (0.999x ➖, 9↑ 8↓)
datafusion / vortex-compact / ns (1.146x ❌, 3↑ 49↓)
datafusion / parquet / ns (0.998x ➖, 0↑ 0↓)
duckdb / vortex-file-compressed / ns (1.005x ➖, 1↑ 4↓)
duckdb / vortex-compact / ns (0.986x ➖, 7↑ 5↓)
duckdb / parquet / ns (0.995x ➖, 4↑ 2↓)
No file size changes detected. |
Benchmarks: Statistical and Population Genetics 📖Commits: PR How to read Verdict and Engines
datafusion / vortex-file-compressed / ns (0.986x ➖, 0↑ 0↓)
datafusion / vortex-compact / ns (0.992x ➖, 1↑ 0↓)
datafusion / parquet / ns (1.002x ➖, 0↑ 0↓)
duckdb / vortex-file-compressed / ns (0.997x ➖, 0↑ 0↓)
duckdb / vortex-compact / ns (0.998x ➖, 0↑ 0↓)
duckdb / parquet / ns (1.011x ➖, 0↑ 0↓)
No file size changes detected. |
Benchmarks: TPC-H SF=10 on S3 📖Commits: PR How to read Verdict and Engines
datafusion / vortex-compact / ns (0.968x ➖, 1↑ 0↓)
datafusion / parquet / ns (0.947x ➖, 1↑ 1↓)
duckdb / vortex-compact / ns (0.892x ➖, 0↑ 0↓)
duckdb / parquet / ns (0.962x ➖, 0↑ 0↓)
|
Benchmarks: Random Access 📖Commits: PR How to read Verdict and Engines
vortex / arrow-ipc / ns (1.002x ➖, 1↑ 1↓)
random-access / vortex-file-compressed / ns (0.984x ➖, 0↑ 0↓)
random-access / parquet / ns (0.997x ➖, 0↑ 0↓)
random-access / lance / ns (0.993x ➖, 0↑ 0↓)
|
Benchmarks: Random Access (S3) 📖Commits: PR How to read Verdict and Engines
random-access / vortex-file-compressed / ns (1.040x ➖, 0↑ 2↓)
random-access / parquet / ns (0.986x ➖, 0↑ 0↓)
random-access / lance / ns (1.096x ➖, 0↑ 2↓)
|
Vortex scans now accept DataFusion's dynamic filters, so every probe-side scan under a hash join or TopK shows them in EXPLAIN output. Regenerated with `--complete` and the byte ranges replaced per the README. Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk> Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01XGkjCDMVSGtEWhEHHW5wGL
Same regeneration as the TPC-DS plans: probe-side scans under hash joins and TopK now show DataFusion's dynamic filters in EXPLAIN output. Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk> Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01XGkjCDMVSGtEWhEHHW5wGL
SearchSortedPrimitiveArray read every element with a one-shot scalar probe, so an encoding that must decode a block to serve a single element paid that decode on every step of the binary search. RunEnd slicing searches its ends this way, and with Pco-compressed ends each search decoded the same Pco page about log2(n) times. A dynamic filter that re-slices those arrays per batch made this dominate TPC-DS on the compact file variant: q96 spent 69% of worker CPU there. Hold a RepeatedArrayProbe for the search instead, so the encoding's probe state (Pco's decoded page) is built once per search. DCO Remediation Commit for Claude <noreply@anthropic.com> I, Claude <noreply@anthropic.com>, hereby add my Signed-off-by to this commit: e9e1ae8 I, Claude <noreply@anthropic.com>, hereby add my Signed-off-by to this commit: 89de46e I, Claude <noreply@anthropic.com>, hereby add my Signed-off-by to this commit: 9651809 I, Claude <noreply@anthropic.com>, hereby add my Signed-off-by to this commit: e61228c I, Claude <noreply@anthropic.com>, hereby add my Signed-off-by to this commit: e745df0 I, Claude <noreply@anthropic.com>, hereby add my Signed-off-by to this commit: a831e08 I, Claude <noreply@anthropic.com>, hereby add my Signed-off-by to this commit: 1946187 I, Claude <noreply@anthropic.com>, hereby add my Signed-off-by to this commit: 8d19c9c Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk> Signed-off-by: Claude <noreply@anthropic.com> Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01XGkjCDMVSGtEWhEHHW5wGL
|
The Generated by Claude Code |
Each comparison in a filter expression decodes its column on its own, so `col >= a AND col <= b` decoded the column twice before the scan decoded it again for output. Counted on a Pco column: two decodes for the range, one for `between`. Emit a single `between` for a lower and upper bound, and when a membership list applies, emit the list alone: its values already lie within the bounds. Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk> Signed-off-by: Claude <noreply@anthropic.com> Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01XGkjCDMVSGtEWhEHHW5wGL
|
No fix applies here. The changes this PR does cause are improvements: Generated by Claude Code |
Summary
Follow-up branch on top of #10375 (Accept DataFusion dynamic filters in VortexSource). It carries that PR's commits plus the changes below, merged with
develop, so the benchmark workflows can measure the combined result againstdevelop.Changes
list_containssorted search (vortex-array): a constant list against an array of integer needles was one equality pass over the whole column per list element, OR-ed together. For integer needles and lists of at least four non-null elements the elements are sorted once and each row is answered by a binary search.vortex-datafusion): complete hash-join filters carrycol IN (keys)next to their min/max bounds when the build side has few distinct keys. Those lists become a staticin_listin the Vortex scan filter when they prune at least half of the rows the bounds keep. Partitioned joins route keys through aCASE, whose branch lists are unioned.vortex_datafusion::convert::dynamic.Local SF1 results (4-core VM, hot median, ratio vs
develop, < 1 is faster): TPC-H geomean 0.92 (q18 0.46, q21 0.57, q7 0.66), TPC-DS geomean 0.78 (q72 0.11, q15/q21 0.26, q40 0.32, q17 0.34). q70 and q94 regress by 1.2x to 1.3x on both Vortex and Parquet once a scan accepts a hash-join dynamic filter; investigation ongoing.Tests:
convert::dynamicunit tests cover membership list conversion, CASE-routed lists, and the leading-column TopK template;list_containstests cover the sorted search path.🤖 Generated with Claude Code
https://claude.ai/code/session_01XGkjCDMVSGtEWhEHHW5wGL
Generated by Claude Code