Repository navigation
perf: opt substr into make_scalar_function dict fast path - #25874
Rich-T-kid wants to merge 6 commits into
Conversation
71ad106 to
0f9e4cd
Compare
|
will wait until we can enable the dictionary flag to run the benchmarks. This should have a significant affect on query performance |
0f9e4cd to
497b608
Compare
6318fb4 to
d172c89
Compare
For scalar start/length, substr can run over the dict's unique values and reuse the keys instead of casting the dict to its value type. - Signature uses EncodingPreservation::dictionary() so coercion keeps the dict wrapper. - Fast path in invoke_with_args runs substr on dict.values() and rewraps via with_values(..). - Per-row start/length still materializes the dict. - return_field_from_args declares Dictionary for the scalar case and the value type for the per-row case so the planner's expected schema matches runtime output. Closes apache#26170
d172c89 to
c11a73c
Compare
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #25874 +/- ##
==========================================
- Coverage 82.80% 82.80% -0.01%
==========================================
Files 1149 1149
Lines 452234 452296 +62
Branches 452234 452296 +62
==========================================
+ Hits 374493 374527 +34
- Misses 54946 54963 +17
- Partials 22795 22806 +11 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
|
run benchmark tpch |
1 similar comment
|
run benchmark tpch |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing rich-T-kid/substr-dict-encoding-preservation (601afda) to b295a67 (merge-base) diff Run configurationrun benchmark tpch
env:
DATAFUSION_RUNTIME_MEMORY_LIMIT: "32G"
baseline:
env:
DATAFUSION_EXECUTION_PARQUET_ENABLE_RLE_TO_DICTIONARY: "true"
changed:
env:
DATAFUSION_EXECUTION_PARQUET_ENABLE_RLE_TO_DICTIONARY: "true"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing rich-T-kid/substr-dict-encoding-preservation (601afda) to b295a67 (merge-base) diff Run configurationrun benchmark tpch
env:
DATAFUSION_RUNTIME_MEMORY_LIMIT: "32G"
baseline:
env:
DATAFUSION_EXECUTION_PARQUET_ENABLE_RLE_TO_DICTIONARY: "true"
changed:
env:
DATAFUSION_EXECUTION_PARQUET_ENABLE_RLE_TO_DICTIONARY: "true"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing rich-T-kid/substr-dict-encoding-preservation (601afda) to b295a67 (merge-base) diff Run configurationrun benchmark tpch
env:
DATAFUSION_RUNTIME_MEMORY_LIMIT: "32G"
baseline:
env:
DATAFUSION_EXECUTION_PARQUET_ENABLE_RLE_TO_DICTIONARY: "true"
changed:
env:
DATAFUSION_EXECUTION_PARQUET_ENABLE_RLE_TO_DICTIONARY: "true"CPU Details (lscpu)Details
Memory Pool PeaksPeak Base:
Pool accounting vs. process RSS Max pool peak is the largest reservation any single query in the run reached; peak RSS covers the whole invocation, including data loading and allocator retention, and the two high-water marks need not coincide in time. The gap is therefore an upper bound on what the pool did not account for, not a measurement of it.
Resource Usagetpch — base (merge-base)
tpch — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing rich-T-kid/substr-dict-encoding-preservation (601afda) to b295a67 (merge-base) diff Run configurationrun benchmark tpch
env:
DATAFUSION_RUNTIME_MEMORY_LIMIT: "32G"
baseline:
env:
DATAFUSION_EXECUTION_PARQUET_ENABLE_RLE_TO_DICTIONARY: "true"
changed:
env:
DATAFUSION_EXECUTION_PARQUET_ENABLE_RLE_TO_DICTIONARY: "true"CPU Details (lscpu)Details
Memory Pool PeaksPeak Base:
Pool accounting vs. process RSS Max pool peak is the largest reservation any single query in the run reached; peak RSS covers the whole invocation, including data loading and allocator retention, and the two high-water marks need not coincide in time. The gap is therefore an upper bound on what the pool did not account for, not a measurement of it.
Resource Usagetpch — base (merge-base)
tpch — branch
File an issue against this benchmark runner |
|
🤔 this got slower because the cardinaility of the column is neaer unqiue. will revisit this after we implement use defined thresholds |
Would you mind marking this as a draft for now? Feel free to mark it ready again once it's good for another look. |
|
sounds good to me |
|
@jayzhan211 this should be good for review. the benchmarks show a slow down for |
jayzhan211
left a comment
jayzhan211
left a comment
There was a problem hiding this comment.
I checked the speedup locally on TPC-H SF1 with enable_rle_to_dictionary=true. substr(l_shipinstruct, 1, 4) grouped is about 3.7x faster (10–11 ms vs 37–41 ms), and Q22 is a bit slower (16–18 ms vs 15–16 ms), matching the bot. Since the config is opt-in, I'm fine with handling the high-cardinality case in the follow-up threshold work you mentioned. A note about the trade-off in the PR description would help.
|
run benchmark tpch |
|
Hi @Rich-T-kid, your benchmark configuration could not be parsed (#25874 (comment)). Error: Usage: Any benchmark name is accepted: Per-side configuration ( env:
# shared env is inherited by BOTH the build and the run, so build
# flags go here. Builds default to no debuginfo for speed; opt back
# in for hung-job gdb dumps and cap jobs to stay within memory:
CARGO_PROFILE_RELEASE_DEBUG: "1"
CARGO_BUILD_JOBS: "1"
baseline:
ref: v45.0.0
env:
# per-side env only reaches the benchmark run, not the build
DATAFUSION_RUNTIME_MEMORY_LIMIT: 1G
changed:
ref: v46.0.0
env:
DATAFUSION_RUNTIME_MEMORY_LIMIT: 2GPod size and CPU architecture (each key is optional and keeps the default when absent; requests and limits are set equal): resources:
cpu: "16"
memory: "128Gi"
arch: arm64 # arm64 or amd64Independent benchmark shards (optional; 1–8, default 1): run benchmark arrow_writer
shards: 4Each shard builds both revisions, measures its assigned cases, and posts its own report. Resources and the job deadline apply per shard. DataFusion requests specifying a shard count greater than 1 fail with an error. Omit shards or use shards: 1 for DataFusion. File an issue against this benchmark runner |
|
run benchmark tpch |
|
Hi @Rich-T-kid, your benchmark configuration could not be parsed (#25874 (comment)). Error: Usage: Any benchmark name is accepted: Per-side configuration ( env:
# shared env is inherited by BOTH the build and the run, so build
# flags go here. Builds default to no debuginfo for speed; opt back
# in for hung-job gdb dumps and cap jobs to stay within memory:
CARGO_PROFILE_RELEASE_DEBUG: "1"
CARGO_BUILD_JOBS: "1"
baseline:
ref: v45.0.0
env:
# per-side env only reaches the benchmark run, not the build
DATAFUSION_RUNTIME_MEMORY_LIMIT: 1G
changed:
ref: v46.0.0
env:
DATAFUSION_RUNTIME_MEMORY_LIMIT: 2GPod size and CPU architecture (each key is optional and keeps the default when absent; requests and limits are set equal): resources:
cpu: "16"
memory: "128Gi"
arch: arm64 # arm64 or amd64Independent benchmark shards (optional; 1–8, default 1): run benchmark arrow_writer
shards: 4Each shard builds both revisions, measures its assigned cases, and posts its own report. Resources and the job deadline apply per shard. DataFusion requests specifying a shard count greater than 1 fail with an error. Omit shards or use shards: 1 for DataFusion. File an issue against this benchmark runner |
|
run benchmark tpch |
|
@jayzhan211 I've addressed your comments as well as updated the PR description, plz take a look 🙇 |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing rich-T-kid/substr-dict-encoding-preservation (3a90a9f) to b295a67 (merge-base) diff Run configurationrun benchmark tpch
env:
DATAFUSION_RUNTIME_MEMORY_LIMIT: "32G"
baseline:
env:
DATAFUSION_EXECUTION_PARQUET_ENABLE_RLE_TO_DICTIONARY: "true"
changed:
env:
DATAFUSION_EXECUTION_PARQUET_ENABLE_RLE_TO_DICTIONARY: "true"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing rich-T-kid/substr-dict-encoding-preservation (3a90a9f) to b295a67 (merge-base) diff Run configurationrun benchmark tpch
env:
DATAFUSION_RUNTIME_MEMORY_LIMIT: "32G"
baseline:
env:
DATAFUSION_EXECUTION_PARQUET_ENABLE_RLE_TO_DICTIONARY: "true"
changed:
env:
DATAFUSION_EXECUTION_PARQUET_ENABLE_RLE_TO_DICTIONARY: "true"CPU Details (lscpu)Details
Memory Pool PeaksPeak Base:
Pool accounting vs. process RSS Max pool peak is the largest reservation any single query in the run reached; peak RSS covers the whole invocation, including data loading and allocator retention, and the two high-water marks need not coincide in time. The gap is therefore an upper bound on what the pool did not account for, not a measurement of it.
Resource Usagetpch — base (merge-base)
tpch — branch
File an issue against this benchmark runner |
## Which issue does this PR close? - N/A ## Rationale for this change The semver check previously compared the PR against the current state of `apache/main`, and did so by doing a full `git` checkout, including the entire repository history. This approach has two downsides: 1. It is expensive: cloning the entire repo history takes ~9 sec locally, versus ~0.7 for a shallow checkout. 2. There's a race condition which results in comparing the PR against the wrong base when assessing whether the PR breaks semantic versioning. On (2): when a PR is opened or updated, GitHub creates a test merge commit that merges the PR into main as of that moment (T0). Now suppose that there is a delay between the CI jobs being queued and the semver job actually running, so that when the semver job runs, `HEAD` is now at T1. The previous version of this check would compare the PR with T1, not T0. For example, if a commit between T0 and T1 added a new API, the semver check would claim that the PR _removes_ that API. Fix this by comparing the merge commit with its first parent instead, which is T0 in the notation above. ## What changes are included in this PR? See above. ## What is the testing strategy for this PR? CI / dev tooling change only. Tested locally on six open PRs (apache#26565, apache#26000, apache#26582, apache#25874, apache#26534, apache#26576): fetched each PR's merge ref at depth 2 and ran both steps against `HEAD^1`. The crate lists and semver results match CI, including two PRs that branched 66 and 154 commits behind main. ## Are there any user-facing changes? No.
Per-row start and length values can create more dictionary entries than Int8, UInt8, Int16, or UInt16 keys can represent. Decode those inputs while retaining dictionary preservation for wide key types, and cover the Int8 overflow regression in the dictionary SQLLogicTest.
| } | ||
| } | ||
|
|
||
| fn dictionary_substr<K: ArrowDictionaryKeyType>( |
There was a problem hiding this comment.
Thanks for the Int8 fix, the repro passes now. One more thing on this row-wise path: per-row start/length rarely repeat as combinations, so the HashMap dedup mostly adds hashing cost. On 8192 rows (same release build), substr(dict, start_array) is 1.5x slower than cast to Utf8 + substr (what main does) with 16 distinct values, and 5.7x slower with 8192.
Only wide key types reach this function, so one key per row always fits. Decoding and wrapping the result with identity keys runs at 1.01-1.03x of the decode path, and the existing unit tests and string/ slt still pass:
fn dictionary_substr<K: ArrowDictionaryKeyType>(
dictionary: &DictionaryArray<K>,
args: &[ArrayRef],
) -> Result<ArrayRef> {
// Only wide key types reach here, so one key per row always fits.
let decoded = take(dictionary.values().as_ref(), dictionary.keys(), None)?;
let mut row_args = vec![decoded];
row_args.extend(args.iter().cloned());
let values = substr(&row_args)?;
let keys = (0..values.len())
.map(|i| K::Native::from_usize(i).unwrap())
.collect::<Vec<_>>();
let keys = PrimitiveArray::<K>::new(keys.into(), values.nulls().cloned());
Ok(Arc::new(DictionaryArray::<K>::try_new(keys, values)?))
}Decode wide-key dictionaries once and wrap the row results with identity keys, avoiding HashMap overhead when start and length combinations rarely repeat.
|
run benchmark tpch |
|
Hi @Rich-T-kid, your benchmark configuration could not be parsed (#25874 (comment)). Error: Usage: Any benchmark name is accepted: Per-side configuration ( env:
# shared env is inherited by BOTH the build and the run, so build
# flags go here. Builds default to no debuginfo for speed; opt back
# in for hung-job gdb dumps and cap jobs to stay within memory:
CARGO_PROFILE_RELEASE_DEBUG: "1"
CARGO_BUILD_JOBS: "1"
baseline:
ref: v45.0.0
env:
# per-side env only reaches the benchmark run, not the build
DATAFUSION_RUNTIME_MEMORY_LIMIT: 1G
changed:
ref: v46.0.0
env:
DATAFUSION_RUNTIME_MEMORY_LIMIT: 2GPod size and CPU architecture (each key is optional and keeps the default when absent; requests and limits are set equal): resources:
cpu: "16"
memory: "128Gi"
arch: arm64 # arm64 or amd64Independent benchmark shards (optional; 1–8, default 1): run benchmark arrow_writer
shards: 4Each shard builds both revisions, measures its assigned cases, and posts its own report. Resources and the job deadline apply per shard. DataFusion requests specifying a shard count greater than 1 fail with an error. Omit shards or use shards: 1 for DataFusion. File an issue against this benchmark runner |
|
run benchmark tpch |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing rich-T-kid/substr-dict-encoding-preservation (fbe0593) to b295a67 (merge-base) diff Run configurationrun benchmark tpch
env:
DATAFUSION_RUNTIME_MEMORY_LIMIT: "32G"
baseline:
env:
DATAFUSION_EXECUTION_PARQUET_ENABLE_RLE_TO_DICTIONARY: "true"
changed:
env:
DATAFUSION_EXECUTION_PARQUET_ENABLE_RLE_TO_DICTIONARY: "true"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing rich-T-kid/substr-dict-encoding-preservation (fbe0593) to b295a67 (merge-base) diff Run configurationrun benchmark tpch
env:
DATAFUSION_RUNTIME_MEMORY_LIMIT: "32G"
baseline:
env:
DATAFUSION_EXECUTION_PARQUET_ENABLE_RLE_TO_DICTIONARY: "true"
changed:
env:
DATAFUSION_EXECUTION_PARQUET_ENABLE_RLE_TO_DICTIONARY: "true"CPU Details (lscpu)Details
Memory Pool PeaksPeak Base:
Pool accounting vs. process RSS Max pool peak is the largest reservation any single query in the run reached; peak RSS covers the whole invocation, including data loading and allocator retention, and the two high-water marks need not coincide in time. The gap is therefore an upper bound on what the pool did not account for, not a measurement of it.
Resource Usagetpch — base (merge-base)
tpch — branch
File an issue against this benchmark runner |
|
QQuery 22 slows down 🤔 |
| pub fn new() -> Self { | ||
| let string = Coercion::new_exact(TypeSignatureClass::Native(logical_string())); | ||
| let string = Coercion::new_exact(TypeSignatureClass::Native(logical_string())) | ||
| .with_encoding_preservation(EncodingPreservation::dictionary()); |
There was a problem hiding this comment.
Returning a dictionary also makes a high-cardinality GROUP BY on the result slower, with or without the guard. With enable_rle_to_dictionary=true on ClickBench, SELECT substr("SearchPhrase", 1, 10) s, COUNT(*) c FROM hits WHERE "SearchPhrase" <> '' GROUP BY s ORDER BY c DESC LIMIT 10 is 1.2x slower than main (1.3x with 2 partitions), although substr itself is faster. The time goes into repartitioning and re-aggregating dictionary group keys, the same cost main already pays when grouping on "SearchPhrase" directly (#24822).
Compacting the dictionary when the aggregate slices its output (#26696) fixes the 2-partition case, but the default stays at 1.2x because the repartition coalescer still rebuilds dictionaries. The next step is to dig into that further, and until then we might need to hold this PR.
There was a problem hiding this comment.
I dug into this further. Part of the cost was hashing and dictionary merging doing work for every dictionary value instead of every row. I'll fix that in the shared code rather than in the aggregate (see #26696), and with a prototype of that this query goes from 1.20x to 1.12x of main. The rest is the general cost of dictionary group keys (#24822), which lower/btrim on dictionary columns already pay on main, so I don't think it needs to block this PR. The guard on the fast path is still needed for Q22.
| } | ||
|
|
||
| if let ColumnarValue::Array(array) = &args.args[0] | ||
| && matches!(array.data_type(), DataType::Dictionary(_, _)) |
There was a problem hiding this comment.
This fast path runs substr over every dictionary value, so its cost follows the dictionary size rather than the batch size: filter keeps the full dictionary, and Parquet shares one dictionary across batches. That's what the bot's Q22 regression is. On data where c_phone is Dictionary(Int32, Utf8), Q22 goes from 11.8 ms on main to 15.1 ms here (38.6 → 55.3 ms with one partition).
Only taking the fast path when the dictionary isn't larger than the batch brings it back to 12.1 ms (40.4 ms), with the same results:
if let ColumnarValue::Array(array) = &args.args[0]
&& matches!(array.data_type(), DataType::Dictionary(_, _))
+ && array.as_any_dictionary().values().len() <= array.len()
&& args.args[1..]
Which issue does this PR close?
Closes #26170.
Rationale for this change
substrdecoded dictionary-encoded strings before evaluation. This did unnecessary per-row work and lost the dictionary encoding in the result.What changes are included in this PR?
Keep dictionary encoding through coercion and return type inference.
substrto the dictionary values once and reuse the original keys.(value key, start, length)combination and build new dictionary keys directly.Neither path casts the full column to strings and back.
Are these changes tested?
A unit test covers the scalar fast path.
dictionary_utf8.sltcovers literal and runtime scalar arguments, scalar dictionary inputs, nullable row-wise arguments, and repeated input values that produce different outputs. It also verifies that the result remains dictionary-encoded.Are there any user-facing changes?
Dictionary inputs now produce dictionary-encoded results. The string values are unchanged.