[jira] [Comment Edited] (IMPALA-8262) Join cardinality not decreased by join filter selectivity

["Zoltán Borók-Nagy (Jira)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/IMPALA-8262?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18125058#comment-18125058 ] 

Zoltán Borók-Nagy edited comment on IMPALA-8262 at 10/8/26 2:02 PM:
--------------------------------------------------------------------

Experimented with Opus 5.5 on a 5-node cluster:

Another example from TPC-DS q72 (SF100, master 1f34ea915e), and notes from a prototype.

{noformat}
15:HASH JOIN [INNER JOIN, PARTITIONED]
|  hash predicates: cs_sold_date_sk = d1.d_date_sk, cs_item_sk = inv_item_sk
|  other predicates: inv_quantity_on_hand < cs_quantity
   est. 9.69B rows, actual 682.80K

18:HASH JOIN [INNER JOIN, BROADCAST]
|  hash predicates: cs_ship_date_sk = d3.d_date_sk
|  other predicates: CAST(d3.d_date AS TIMESTAMP) > CAST(d1.d_date AS TIMESTAMP) + INTERVAL 5 days
   est. 311.50M rows, actual 651.11K
{noformat}

Today this mostly inflates memory estimates. It becomes critical once join NDV estimates are fixed (IMPALA-15520, FK/PK NDV cap): with that fix alone, q72's per-host memory estimate goes from 7.09 GB to 197 GB. Applying the selectivity of these conjuncts brings it to 11.6 GB.

*Rules that worked in the prototype (all 104 TPC-DS queries):*
* Candidates:
** the join's conjuncts_, which filter the output for every join type;
** the non-equi ON conjuncts of inner and cross joins only. For outer, semi and anti joins they only decide which rows match. For anti joins they can even raise the output.
* Only binary comparisons, using the analyzed selectivity, or 1 - 1/max(NDV) for a two-sided <>. Everything else counts as unknown. Combined like scan conjuncts with PlanNode.computeCombinedSelectivity(): one DEFAULT_SELECTIVITY (0.1) for all unknowns, plus exponential backoff.
* Skip conjuncts that reference a tuple nullable at this join. For example, q78's sr_ticket_number IS NULL after a LEFT JOIN would get selectivity of about 0 from the column's null count.
* Skip disjunctions. In a first version that included them, q13, q48 and q85 dropped to 2-8 estimated rows, because the single-table parts of their OR clauses are already pushed down to the scans. Broadcast joins then flipped to partitioned. The TPC-H Q7 example in the description is that kind of OR, so it needs care: only the part that is not already applied below the join may count.
* Runtime filter selectivity (RuntimeFilterGenerator.getJoinNodeSelectivity) must keep using the estimate without these conjuncts, because a filter only applies the equi-join keys.


was (Author: boroknagyz):
Another example from TPC-DS q72 (SF100, master 1f34ea915e), and notes from a prototype.

{noformat}
15:HASH JOIN [INNER JOIN, PARTITIONED]
|  hash predicates: cs_sold_date_sk = d1.d_date_sk, cs_item_sk = inv_item_sk
|  other predicates: inv_quantity_on_hand < cs_quantity
   est. 9.69B rows, actual 682.80K

18:HASH JOIN [INNER JOIN, BROADCAST]
|  hash predicates: cs_ship_date_sk = d3.d_date_sk
|  other predicates: CAST(d3.d_date AS TIMESTAMP) > CAST(d1.d_date AS TIMESTAMP) + INTERVAL 5 days
   est. 311.50M rows, actual 651.11K
{noformat}

Today this mostly inflates memory estimates. It becomes critical once join NDV estimates are fixed (IMPALA-15520, FK/PK NDV cap): with that fix alone, q72's per-host memory estimate goes from 7.09 GB to 197 GB. Applying the selectivity of these conjuncts brings it to 11.6 GB.

*Rules that worked in the prototype (all 104 TPC-DS queries):*
* Candidates:
** the join's conjuncts_, which filter the output for every join type;
** the non-equi ON conjuncts of inner and cross joins only. For outer, semi and anti joins they only decide which rows match. For anti joins they can even raise the output.
* Only binary comparisons, using the analyzed selectivity, or 1 - 1/max(NDV) for a two-sided <>. Everything else counts as unknown. Combined like scan conjuncts with PlanNode.computeCombinedSelectivity(): one DEFAULT_SELECTIVITY (0.1) for all unknowns, plus exponential backoff.
* Skip conjuncts that reference a tuple nullable at this join. For example, q78's sr_ticket_number IS NULL after a LEFT JOIN would get selectivity of about 0 from the column's null count.
* Skip disjunctions. In a first version that included them, q13, q48 and q85 dropped to 2-8 estimated rows, because the single-table parts of their OR clauses are already pushed down to the scans. Broadcast joins then flipped to partitioned. The TPC-H Q7 example in the description is that kind of OR, so it needs care: only the part that is not already applied below the join may count.
* Runtime filter selectivity (RuntimeFilterGenerator.getJoinNodeSelectivity) must keep using the estimate without these conjuncts, because a filter only applies the equi-join keys.

> Join cardinality not decreased by join filter selectivity
> ---------------------------------------------------------
>
>                 Key: IMPALA-8262
>                 URL: https://issues.apache.org/jira/browse/IMPALA-8262
>             Project: IMPALA
>          Issue Type: Bug
>          Components: Frontend
>    Affects Versions: Impala 3.1.0
>            Reporter: Paul Rogers
>            Priority: Major
>              Labels: performance
>
> Consider a subset of the plan for TPC-H query 7. (See {{tpch-all.test}} for details.)
> {noformat}
> 11:AGGREGATE [FINALIZE]
> |  output: sum(l_extendedprice * (1 - l_discount))
> |  group by: n1.n_name, n2.n_name, year(l_shipdate)
> |  row-size=58B cardinality=575.77K
> |
> 10:HASH JOIN [INNER JOIN]
> |  hash predicates: c_nationkey = n2.n_nationkey
> |  other predicates: ((n1.n_name = 'FRANCE' AND n2.n_name = 'GERMANY') OR (n1.n_name = 'GERMANY' AND n2.n_name = 'FRANCE'))
> |  row-size=132B cardinality=575.77K
> |
> |--05:SCAN HDFS [tpch.nation n2]
> |     row-size=21B cardinality=25
> |
> 09:HASH JOIN [INNER JOIN]
> |  hash predicates: s_nationkey = n1.n_nationkey
> |  row-size=111B cardinality=575.77K
> {noformat}
> Here, we have join 09 feeding 576K rows into join 10. All 576K rows pass along to the aggregate 11. Notice, however, that join 10 has a that picks out 2 of the 25 countries in each of two paths. The selectivity of the filters should be something like 2 * 2/25 = 0.16. Thus, the output cardinality of the 10 join should be 577K * 0.16 = 92K.
> The problem is that the join cardinality calculations don't consider join filter selectivity.
> It may be that this was done to handle the outer join case, in which filters applied in the outer-side scan must be re-applied on the join. Omitting the filters avoids duplicate accounting for the selectivity.
> But, that case is special and should be handled specially as part of IMPALA-8213. Except for correlated filters, the planner *should* apply join filter selectivity to the join output cardinality calculations.
> This error has consequences. The filter should reduce the number of rows though the join. Because it does so, it should come early in the join tree to reduce the set of rows processed. But, because selectivity is ignored, the planner does not see the join as a filter, and ends up putting the join 10 at the top of the join tree. (See the test file for the full plan.) The result is that Impala schleps around many more rows than necessary, only to discard them near the top of the DAG.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)

---------------------------------------------------------------------
To unsubscribe, e-mail: issues-all-unsubscribe@impala.apache.org
For additional commands, e-mail: issues-all-help@impala.apache.org