[jira] [Created] (IMPALA-8015) Incorrect cardinality calculation for non-FK/PK case

["Paul Rogers (JIRA)" <ji...@apache.org>]

Paul Rogers created IMPALA-8015:
-----------------------------------

             Summary: Incorrect cardinality calculation for non-FK/PK case
                 Key: IMPALA-8015
                 URL: https://issues.apache.org/jira/browse/IMPALA-8015
             Project: IMPALA
          Issue Type: Bug
          Components: Frontend
    Affects Versions: Impala 3.1.0
            Reporter: Paul Rogers
            Assignee: Paul Rogers


Please see the background information in IMPALA-8014, including notation. The material here is a version of that in [these note|https://github.com/paul-rogers/impala/wiki/Planner].

h4. Background

Impala uses two distinct ways to estimate join cardinality: FK/PK and the "generic" case. Both are in {{JoinNode.getJoinCardinality()}}.

The generic case handles the M:N case, multiple rows in the left table join with multiple rows in the right table. The {{getGenericJoinCardinality()}} works out the estimated cardinality.

h4. Deriving the Cardinality

Assume a join of two tables, left ({{l}}) and right ({{r}}), with no predicate. We have a Cartesian product:

{noformat}
|l >< r| = |l| * |r|
{noformat}

Suppose we have an equi-join predicate: {{l.a = r.b}} and that we have the NDV values for both columns.

Lets assume the left is the probe side. How many rows will it match? It will match all the rows with a given value, or {{|r 𝜎 b=x|}}, (which we'll abbreviate to {{|r.b=x|}} which is:

{noformat}
|r.b=x| = |r| / ndv(r.b)
{noformat}

Intuitively, the {{r.b}} keys are uniformly distributed, each left-hand row matches a group of right-hand rows the size of which is given by the equation above. If all values in r are unique, then {{|r| / ndv(r.b)}} is one and we match one row. If all values are the same, {{|r| / ndv(r.b) = |r|}}, so we match all rows (or none).

The total number of output rows, to a first approximation, is:

{noformat}
|l >< r| = |l| * |r| / ndv(r.b)
{noformat}

Next, how many rows will appear on the left (probe) side? It is the output of the previous join, or the scan of the table in that node, thus {{scan(l)}}.

How many rows on the right? Impala requires that the right side be a table, so the number is {{|scan(r)|}}. (Actually, either side can be a table scan, but the reasoning is symmetrical.) If we randomly sample the {{r}} table to get {{scan(r)}}, then the probability of getting any one row from r is:

{noformat}
p(scan(r) : row = x) = |scan(r)| / |r|
{noformat}

(Read this as the probability or a row x appearing in the scan result of r.)

Let's check the math. Using the above, the total number of scanned rows is:

{noformat}
|scan(r)| = |r| * p(scan(r) : row = x)
          = |r| * |scan(r)| / |r|
          = |scan(r)|
{noformat}

The total number of keys in each group in the scanned result is the group size adjusted for the probability that any one group member appears:

{noformat}
|scan(r.b)| = |r| / ndv(r.b) * p(r.b)
            = |r| / ndv(r.b) * |scan(r)| / |r|
            = |scan(r)| / ndv(r.b)
{noformat}

Intuitively, however may rows are scanned, they are still divided into the same set of groups.

Next we can work out the total join cardinality given the number of rows on the left, and the number of matching rows on the right:

{noformat}
|l >< r| = |scan(l)| * |scan(r.b)|
         = |scan(l)| * |scan(r)| / ndv(r.b)
{noformat}

The above assumes that all rows from l match rows in r. But, if there are more key values in l than r, some can't match, and visa-versa. What is the probability that a row from l will find a match? We can reason it out like this:

* If there are fewer values in l.a than in r.b, we can assume all will match.
* If there are more values in l.a than in r.b, we can assume we'll match all values in l.a, then discard the extra values that don't match.

The details are already worked out in IMPALA-8014. The probability is:

{noformat}
p(match) = ndv(l.a) / ndv(r.b) if ndv(l.a) < ndv(r.b),
           ndv(r.b) / ndv(l.a) otherwise.
{noformat}

Intuitively, if there are more l rows than r, then a match is the ratio between them and visa-versa.

Let's check. If the ndv's are equal, the probability of a match is 1. If either table is empty, the probability is 0. If ndv(l.a) is half that of ndv(l.b) the probability is 0.5 and visa versa. All good.

Putting it all together:

{noformat}
|l >< r| = |scan(l)| * |scan(r)| / ndv(r.b) * p(match)
{noformat}

For the case that {{ndv(l.a) > ndv(r.b)}}:

{noformat}
|l >< r| = |scan(l)| * (|scan(r)| / ndv(r.b)) * (ndv(r.b) / ndv(l.a))
         = |scan(l)| * |scan(r)| / ndv(l.a)
         = (|scan(l)| / ndv(l.a)) * |scan(r)|
{noformat}

Think if this as the inverse of the above: every row in r matches the set of rows in l with the same key value.

h4. Code Bug

The code uses the following:

{code:java}
      double lhsAdjNdv = slots.lhsNdv();
      if (slots.lhsNumRows() > lhsCard) lhsAdjNdv *= lhsCard / slots.lhsNumRows();
      double rhsAdjNdv = slots.rhsNdv();
      if (slots.rhsNumRows() > rhsCard) rhsAdjNdv *= rhsCard / slots.rhsNumRows();
      long joinCard = Math.round((lhsCard / Math.max(1, Math.max(lhsAdjNdv, rhsAdjNdv))) *
          rhsCard);
{code}

Where:

* {{lhsCard}} is the output cardinality of the l (previous join) table or {{|l|}}
* {{rhsCard}} is the output cardinality of the r (scanned) table or {{|scan(r)|}}
* {slots.lhsNdv()}} is the NDV of the left key or {{ndv(l.a)}}
* {slots.lnsNumRows()}} is the cardinality of the LHS table, or {{|l|}}
* {slots.rhsNdv()}} is the NDV of the right key or {{ndv(r.a)}}
* {slots.rnsNumRows()}} is the cardinality of the LHS table, or {{|r|}}

Translated to our notation:

{noformat}
adjNdv(l.a) = ndv(l.a) * ( |scan(l)| / |l| if |l| > |scan(l)|, 1 otherwise )
adjNdv(r.b) = ndv(r.b) * ( |scan(r)| / |r| if |r| > |scan(r)|, 1 otherwise )
|join| = |scan(l)| * |scan(r)| / max( adjNdv(l.a), adjNdv(r.b) )
{noformat}

Since {{|scan\(x)| <= |x|}} we can reduce the above to:

{noformat}
adjNdv(l.a) = ndv(l.a) * |scan(l)| / |l|
adjNdv(r.b) = ndv(r.b) * |scan(r)| / |r|
|join| = |scan(l)| * |scan(r)| / max( adjNdv(l.a), adjNdv(r.b) )
{noformat}

For one case:

{noformat}
|join| = |scan(l)| * |scan(r)| / adjNdv(l.a)
       = |scan(l)| * |scan(r)| / (ndv(l.a) * |scan(l)| / |l|)
       = |scan(l)| * |scan(r)| * |l| / (ndv(l.a) * |scan(l)|)
       = |scan(l)| * |scan(r)| * (|l| / |scan(l)|) / ndv(l.a)
{noformat}

Comparing the two results

{noformat}
|l >< r|  = |scan(l)| * |scan(r)| / ndv(l.a)
         != |scan(l)| * |scan(r)| * (|l| / |scan(l)|) / ndv(l.a)
{noformat}

The only way we get the code's answer is if we assume {{|l| / |scan(r)| = 1}} which seems unlikely. That is, the code applies an unnecessary multiplication factor that increases as the ratio of scanned rows decreases.

So, a bug. 

h4. Worked Example

Let's check.

* {{|l| = 10000}}
* {{|r| = 100}}
* {{scan(l) = 3000}}
* {{scan(r) = 50}}
* {{ndv(l.a) = 40}}
* {{ndv(r.a) = 20}}

Let's work this out intuitively first. The join is the normal cross join with adjustments:

{noformat}
|l >< r| = |l| * |r| * adjustments
{noformat}

Adjustments:

* The left side scan returns only 30% of the total rows. 
* The left hand has twice the key values of the right, so we have to discard half of the left rows.
* Of the l keys with a match in r, it will match, on average, 100/20 = 5 rows.
* But half of r is filtered out, reducing the available matching rows.

And:

{noformat}
|l >< r| = (|l| / 2) * (|r| / 20 / 2)
         = 10000 * 30% / 2 * 100 * / 20 / 2
         = 1500 * 2.5
         = 3750
{noformat}

Using the formula derived above:

{noformat}
p(match) = ndv(r.b) / ndv(l.a) if ndv(r.b) < ndv(l.a),
           ndv(l.a) / ndv(r.b) otherwise
|l >< r| = |scan(l)| * |scan(r)| * (|r| / ndv(r.b)) * p(match)
         = |scan(l)| * |scan(r)| * (|r| / ndv(r.b)) * (ndv(r.b) / ndv(l.a))
         = |scan(l)| * |scan(r)| * |r| / ndv(l.a)
         = 3000 * 50 * 100 / 40
         = 3750
{noformat}

Which all works out.

Now let's check the code's formula:

{noformat}
|l >< r| = |scan(l)| * |scan(r)| * (|l| / |scan(l)|) / ndv(l.a)
         = (|scan(l)| / |scan(l)|) * |scan(r)| * |l| / ndv(l.a)
         = |scan(r)| * |l| / ndv(l.a)
         = 50 * 10,000 / 50
         = 10,000
{noformat}

Which does not quite work out, verifying the bug.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

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