Optimize TPC-DS query plans for streaming executor (q80, q31, q11) - #22070
Merged
Conversation
…16s) The old plan built each channel segment (store/catalog/web) by first performing a LEFT JOIN of the full sales table with the returns table, creating a massive intermediate with nullable return columns. Only then were the selective dimension filters applied (date range, item price > 50, promotion channel_tv = 'N'). The new plan reorders the join chain so that sales rows are filtered through the selective dimension tables first: 1. Join date_dim (pre-filtered to 30-day range) — eliminates ~97% of rows 2. Join item (pre-filtered to i_current_price > 50) — further reduction 3. Join promotion (pre-filtered to p_channel_tv = 'N') — further reduction 4. Join id_dim (store/catalog_page/web_site) 5. LEFT JOIN returns — now operates on the already-reduced sales set This means the expensive LEFT JOIN with returns processes far fewer rows, and the post-join .filter() call is eliminated entirely since all predicates are now pushed into pre-filtered dimension table joins. SF1000 benchmark: 50.3s → 15.8s (3.2x speedup), validation passed.
The old plan built two broad CTEs (ss and ws) that joined store_sales
or web_sales with date_dim and customer_address, then grouped by
(ca_county, d_qoy, d_year) across ALL quarters and years. Six filtered
views (ss1..ss3, ws1..ws3) were then derived by post-filtering each
CTE to a specific quarter of a single year. This meant the group_by
processed every row in the sales tables regardless of which quarter
was ultimately needed.
The new plan pre-filters date_dim to the target year and builds six
independent per-quarter aggregation pipelines directly:
1. Filter date_dim to target year, then filter to specific quarter
2. Join the quarter-specific date keys to sales first
3. Join customer_address
4. Group by ca_county only (d_qoy and d_year removed from group_by
since each pipeline is quarter/year-specific)
5. Inject d_year as a literal in the final select
Each pipeline processes ~1/12th the data (1 quarter of 1 year) instead
of all quarters of all years, and the group_by key width is reduced
from 3 columns to 1.
SF1000 benchmark: 22.1s → 5.7s (3.9x speedup), validation passed.
…0s → 5s) The old plan called create_year_total 4 times (store_sales × year1, store_sales × year2, web_sales × year1, web_sales × year2). Each call joined the full customer table and grouped by 8 columns (7 customer display columns + d_year), producing wide intermediate frames. The customer table was effectively scanned and joined 4 separate times. The new plan aggregates by c_customer_sk only (a single integer key): 1. Join sales directly to year-filtered date_dim (no customer table) 2. Group by the customer SK foreign key only 3. Join all four aggregates on customer_sk (narrow integer joins) 4. Apply the growth-rate filter 5. Join customer table once at the very end for display columns This reduces group_by key width from 8 columns to 1, eliminates 3 of 4 customer table scans, and narrows all intermediate joins from customer_id strings to integer SKs. SF1000 benchmark: 20.1s → 5.5s (3.7x speedup), validation passed.
github-actions
Bot
added
Python
vyasr
added
improvement
Matt711
reviewed
Apr 8, 2026
Comment on lines
207
to
208
| (pl.col("s_first_year_total") > 0) | ||
| & (pl.col("w_first_year_total") > 0) |
Member
There was a problem hiding this comment.
See if you can push these filters up down. See #22051
- Filter t_s_first and t_w_first on year_total > 0 before joining, rather than after all four tables are joined. - Since both first-year totals are guaranteed > 0 after the pushed-down filters, simplify the CASE/WHEN ternary expressions to plain division.
Matt711
approved these changes
Apr 10, 2026
Contributor
Author
|
/merge |
vyasr
added a commit
to vyasr/cudf
that referenced
this pull request
Apr 13, 2026
Optimize three TPC-DS query plans for the cudf-polars streaming executor, completing the "big 8" set of queries that show significant improvement. Five queries (q11, q31, q64, q75, q80) were already optimized in NVIDIA#22011 and NVIDIA#22070. This PR adds the remaining three. q9: Replace 5 separate store_sales scans with single-pass conditional aggregation. Uses pl.when() guards per CASE bucket so only one scan feeds all five aggregations. SF1k: 5.5s -> 3.7s (1.5x). q4: Rewrite to 6 separate year_total computations (one per channel x year) with FK-only aggregation by customer_sk, joining the customer dimension table last. Avoids expensive customer join before aggregation. SF1k: 7.0s -> 6.8s (1.03x, but prevents 2.7x regression from naive plan). q74: FK-only aggregation by customer_sk for both store_sales and web_sales pipelines, joining customer dimension last. Eliminates early customer join that inflates intermediate row counts. SF1k: 5.1s -> 3.4s (1.5x). Combined with the prior PRs, the full "big 8" set delivers 1.17x overall speedup on the 99-query SF1k suite (591s -> 508s), saving 83.8s (14.2%) with zero regressions or validation failures.
3 tasks
vyasr
added a commit
to vyasr/cudf
that referenced
this pull request
Apr 13, 2026
Optimize three TPC-DS query plans for the cudf-polars streaming executor, completing the "big 8" set of queries that show significant improvement. Five queries (q11, q31, q64, q75, q80) were already optimized in NVIDIA#22011 and NVIDIA#22070. This PR adds the remaining three. q9: Replace 5 separate store_sales scans with single-pass conditional aggregation. Uses pl.when() guards per CASE bucket so only one scan feeds all five aggregations. SF1k: 5.5s -> 3.7s (1.5x). q4: Rewrite to 6 separate year_total computations (one per channel x year) with FK-only aggregation by customer_sk, joining the customer dimension table last. Avoids expensive customer join before aggregation. SF1k: 7.0s -> 6.8s (1.03x, but prevents 2.7x regression from naive plan). q74: FK-only aggregation by customer_sk for both store_sales and web_sales pipelines, joining customer dimension last. Eliminates early customer join that inflates intermediate row counts. SF1k: 5.1s -> 3.4s (1.5x). Combined with the prior PRs, the full "big 8" set delivers 1.17x overall speedup on the 99-query SF1k suite (591s -> 508s), saving 83.8s (14.2%) with zero regressions or validation failures.
vyasr
added a commit
to vyasr/cudf
that referenced
this pull request
Apr 13, 2026
Optimize three TPC-DS query plans for the cudf-polars streaming executor, completing the "big 8" set of queries that show significant improvement. Five queries (q11, q31, q64, q75, q80) were already optimized in NVIDIA#22011 and NVIDIA#22070. This PR adds the remaining three. q9: Replace 5 separate store_sales scans with single-pass conditional aggregation. Uses pl.when() guards per CASE bucket so only one scan feeds all five aggregations. SF1k: 5.5s -> 3.7s (1.5x). q4: Rewrite to 6 separate year_total computations (one per channel x year) with FK-only aggregation by customer_sk, joining the customer dimension table last. Avoids expensive customer join before aggregation. SF1k: 7.0s -> 6.8s (1.03x, but prevents 2.7x regression from naive plan). q74: FK-only aggregation by customer_sk for both store_sales and web_sales pipelines, joining customer dimension last. Eliminates early customer join that inflates intermediate row counts. SF1k: 5.1s -> 3.4s (1.5x). Combined with the prior PRs, the full "big 8" set delivers 1.17x overall speedup on the 99-query SF1k suite (591s -> 508s), saving 83.8s (14.2%) with zero regressions or validation failures.
shrshi
pushed a commit
to shrshi/cudf
that referenced
this pull request
May 12, 2026
…VIDIA#22070) ## Summary Optimizes three TPC-DS query implementations to produce better plans for the streaming executor, validated against DuckDB SF1000 golden results. - **q80** (50s → 16s, 3.2×): Reorder joins so that date_dim, item, and promotion filters are applied BEFORE the LEFT JOIN with store_returns, avoiding materializing unfiltered wide intermediates - **q31** (22s → 6s, 3.9×): Split CTEs into per-quarter pipelines instead of building all-quarters data then self-joining by quarter, and remove d_year/d_qoy from group_by since each pipeline already targets a single quarter - **q11** (20s → 6s, 3.7×): Aggregate by customer_sk (integer FK) instead of wide display columns (customer_id, first_name, last_name, etc.), then join the customer dimension table once at the end to retrieve display columns ## Optimization patterns These follow the same playbook established in NVIDIA#22011: 1. **Filter pushdown before expensive joins**: Move selective dimension-table joins earlier in the pipeline 2. **Year/quarter-split CTEs**: When a CTE builds data across multiple periods then self-joins by period, split into per-period pipelines 3. **Aggregate by FK, join dimension last**: Group by integer surrogate keys instead of wide display columns, join dimension once at end Authors: - Vyas Ramasubramani (https://github.com/vyasr) - Matthew Murray (https://github.com/Matt711) Approvers: - Matthew Murray (https://github.com/Matt711) URL: NVIDIA#22070
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
3 participants
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.
Summary
Optimizes three TPC-DS query implementations to produce better plans for the streaming executor, validated against DuckDB SF1000 golden results.
Optimization patterns
These follow the same playbook established in #22011: