[
https://issues.apache.org/jira/browse/IMPALA-8018?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Paul Rogers updated IMPALA-8018:
--------------------------------
Description:
The math is wrong when we estimate the cardinality for both M:1 (FK/PK joins)
(IMPALA-8014) and for M:N (generic) joins (IMPALA-8015). General practice is
that the two cases are treated the same as described here.
TL;DR: skip to the last section for the proposed change. See IMPALA-8014 and
IMPALA-8015 for details about the bugs in the current code.
h4. Background
Join logic is complex. To ensure that the analysis is sound we work it out from
first principles, then verify against the Swami & Schiefer, [On the Estimation
of Join Result
Sizes|https://pdfs.semanticscholar.org/2735/672262940a23cfce8d4cc1e25c0191de7da7.pdf]
(S&S). Note especially section 3, Background.
h4. Definitions
The following terms are used below:
* _Relation:_ either a _base table_ or the result of a join. (Relational
theory says that both are simply relations.) We use an upper case letter,
typically R, for a relation. E.g.: {{R1}}, {{R1}}
* _Key:_ A column used in a join. We use a lower case letter for a column, and
typically use “k” for a key, along with the relation number. E.g. {{k1}} and
{{ON R1.k1a = R2.k2a}}.
* _Compound key:_ A join key made up of multiple columns. We use a letter to
denote each column. E.g. {{k1a}} and {{ON R1.k1a = R2.k2a AND R1.k1b = R2.k2b}}
* _Join:_ the join of two relations. Impala supports the join of two base
tables (a table-to-table join), or the join of a base table and another join (a
table-to-join join.) We use the symbol {{⋈}} to denote a join. Impala typically
uses a hash join.
* _Relation cardinality:_ the number of rows in the relation, denoted {{|R|}}.
* _Join cardinality:_ the cardinality of the relation produced by the join.
That is, the join’s output cardinality. Denoted as {{|R1 ⋈ R2|}}.
* _Key column cardinality:_ the number of unique values in a column, denoted
{{|k|}}. Also known as the NDV of the column or {{ndv(k)}}.
In Impala, HMS provides the relation and join key cardinality as table and
column statistics respectively.
Since Impala typically uses hash joins, it is helpful to use terminology
specific to that case:
* _Probe_ side of a join: the larger side (ideally) of a hash join. Also
called the “left” side. Is known as {{chiild(0)}} in the code. Appears at the
same level as the join in the query plan (which would be to the left if the
plan were rotated 90 degrees counter-clockwise.)
* _Build_ side of a join: the smaller side (ideally) of a hash join. Also
called the “right” side. Is known as {{child(1)}} in the code. Appears indented
directly under the join in the query plan.
In Impala:
* The detail table in a M:1 relationship is always on the probe (left) side of
a join. Represented as {{child(0)}} in the code.
* The master table in a M:1 relationship is always on the build (right) side
of a join. Represented as {{child(1)}} in the code.
Finally, we also need:
* The _scan_ of a table: the (possibly smaller) relation produced by applying
one or more predicates while scanning a table. We are concerned with the
cardinality of the scan, denoted as {{|R1'|}}. We assume that Impala has
already used rules (not discussed here) to estimate the cardinality of the
selection.
h4. Deriving the Join Formula For the M:1 Case
In RDBMS, a primary key (PK) is a column (or, more typically, set of columns)
that uniquely identify a row in the master table (M). Primary keys are
generally indexed via a unique index. Since keys are unique:
{noformat}
|M.pk| = |M|
{noformat}
The Detail table forms a M:1 relationship with the master table. Each foreign
key (FK) in the detail table (D) references one primary key in the master
table. Because of the M:1 relationship:
{noformat}
|D.fk| <= |M| << |D|
{noformat}
If we read all rows from both tables, and all primary keys appear as foreign
keys, the the join cardinality is simply:
{noformat}
|D ⋈ M| = |D| [Equation 1]
{noformat}
h5. Filtering Master Rows
Let’s consider what happens when we filter out master (M) table rows producing
a subset M'. To do so, we make three assumptions:
* A uniform distribution of foreign keys, (the “Uniformity” assumption in the
S&S paper cited above),
* Every primary key is referenced by an equal number of foreign keys (implied
by the uniformity assumption),
* The filter results in a random sampling of master rows not correlated with
the join keys.
A later section will relax the third assumption.
Then the probability of any particular primary key appearing is:
{noformat}
p(pk) = |M'| / |M|
{noformat}
The result value is a probability (hence the {{p()}} function) given by the
ratio of selected rows to total table rows (from basic probability theory.) The
value runs from 0 (no master rows match the scan predicates) to 1 (all rows
present).
The revised join cardinality is:
{noformat}
|D ⋈ M'| = |D| * p(pk)
= |D| * |M'| / |M| [Equation 2]
{noformat}
h5. Filtering Detail Rows
Suppose we instead filter detail rows to produce a new subset D'. Again we make
some assumptions:
* Uniform distribution of foreign key values across primary keys.
* The "Containment" assumption from the S&S paper that the set of foreign keys
is a (possibly full) subset of the set of primary keys.
* The filter results in a random sampling of detail rows not correlated with
the join keys.
Again, a later section will relax the third assumption.
With these assumptions, we can see that if we join D' with the master table M,
every row that remains in D' will still find a match in M, so:
{noformat}
|D' ⋈ M| = |D'|
{noformat}
h5. Filtering Both Master and Detail Rows
Continuing with the assumption that filters are not correlated with keys, we
can combine the two selection models to produce:
{noformat}
|D' ⋈ M'| = |D'| * |M'| / |M| [Equation 3]
{noformat}
h5. Non-Containment
The above is based on the S&S Containment assumption: that the set of foreign
keys is a subset of the set of primary keys. In a normal RDBMS with integrity
constraints, this is a valid assumption. But, in Big Data, things are messy and
we can’t actually make this assumption. Fortunately, there is a way loosen the
containment assumption.
Suppose the master table has half the number of keys as the detail table. Using
the uniformity assumption, half the foreign keys will go unmatched. If there
are four times as many foreign keys as primary keys, only a quarter of the
detail rows will find matches. The probability of any one foreign key finding a
match is:
{noformat}
p(match) = |M.pk| / |D.fk|
{noformat}
The above applies only if {{|D.fk| > |M.pk|}}; no adjustment is needed if
{{|D.fk| < |M.pk|}}. We can express this mathematically as:
{noformat}
p(match) = |M.pk| / max(|D.fk|, |M.pk|) [Equation 4]
{noformat}
Combining this with Equation 3, recalling that {{|M.pk| = |M|}}:
{noformat}
|D' ⋈ M'| = |D'| * (|M'| / |M|) * p(match)
= |D'| * |M'| / |M| * |M.pk| / max(|D.fk|, |M.pk|)
= |D'| * |M'| * (|M| / |M|) / max(|D.fk|, |M.pk|)
= |D'| * |M'| / max(|D.fk|, |M|)
{noformat}
If we rearrange the above we get:
{noformat}
|D'| * |M'|
|D' ⋈ M'| = ---------------- [Equation 5]
max(|D.fk|, |M|)
{noformat}
h4. The S&S Equation
Let’s check out answer against the S&S paper cited above by assuming the
master/detail relationship and removing from our equation the affect of scans.
The paper’s equation:
{noformat}
|R1 ⋈ R2| = min( |k1|, |k2| ) * ( |R1| / |k1| ) * ( |R2| * |k2| )
= |R1| * |R2| / max( |k1|, |k2| )
{noformat}
If we assume:
* {{R1 = D}}
* {{R2 = M}}
* {{k1 = D.fk}}
* {{k2 = M.pk}}
* {{|D.fk| > |M.pk|}}
We get:
{noformat}
|D ⋈ M| = |D| * |M| / max( |D.fk|, |M.pk| )
{noformat}
Which is the same as Equation 5 (assuming no selection). All good.
h4. Compound Keys
The discussion thus far has assumed a single key column per table so that we
have NDV values for each. The next challenge occurs if the query has compound
columns:
{noformat}
SELECT …
FROM T1, T2
WHERE T1.k1a = T2.k2a AND T1.k1b = T2.k2b
{noformat}
We will use the symbol {{j}} to mean a joint (compound) key. The joint key for
table {{T1}} is {{T1.j1}} and {{||T1.j1|}} is the cardinality (NDV) of that
key.
In an RDBMS, if an index exists on (k1a, k1b) and (k2a, k2b) then it is
possible to determine the joint NDV (which is just the number of index
entries.) In HMS, we have the individual NDVs, but not the joint NDV.
The S&S paper supplies the require assumption: “Independence: Within each
relation, values chosen from distinct columns that are involved in join
predicates are independent.” This lets us estimate the joint NDV:
One way to think of this is the the columns form a matrix, say (year, month).
Every year has 12 months, so there is no correlation between the two. If we
have data for 10 years, we will have 10 * 12 = 120 distinct key values.
{noformat}
|T.j| = |T.k1| * |T.k2| * … * |T.kn|
= ∏ |T.ki|
{noformat}
Where {{∏}} is the product operator. (S&S, section 3.3, calls this the
"Multiplicative Rule.")
Of course, keys are not always independent. Suppose the keys are (state, city).
In this case, there are 50 states. But most city names occur in only one state,
so NDV(state) * NDV(city) >> NDV(state, city). Still, we know that, regardless
of the correlation, we can never have more keys than rows. We can adjust the
expression to:
{noformat}
|T.j| = min( ∏ |T.ki|, |T| )
{noformat}
We can also assume that joins are typically formulated to be as selective as
possible. So the assumption may not be too far off. Still, if a file contains
addresses, so the key is (state, city, street), and we just the (state, city)
prefix, the joint NDV will smaller than the table cardinality. So we need both
terms above.
Adding this to the expression derived above:
{noformat}
|D'| * |M'|
|D' ⋈ M'| = --------------------------------------------- [Equation 6]
max(min(∏ |D.fki|, |D|), min(∏ |M.pki|, |M|))
{noformat}
h4. Correlated Filters
Suppose that we relax the assumption that filters are not correlated with join
keys and assume they are. For example:
{noformat}
SELECT ...
FROM master m, detail d
WHERE m.pk = d.fk
AND m.pk = 10
{noformat}
The simplified math used above no longer applies: the filter affects the number
of join keys available. Applying Equation 5 in this case results in a "double
accounting" for the primary key filter.
Equation 5 has four terms. Consider a predicate on the master table:
* The predicate reduces the number of table rows from {{|M|}} to {{|M'|}}.
* The predicate also reduces the number of keys from {{|M.pk| (= |M|)}} to
{{|M.pk'| (= |M'|)}}.
The above equation does not consider the reduction in key cardinality. Let's
add that now:
{noformat}
|D'| * |M'|
|D' ⋈ M'| = --------------------- [Equation 7]
max(|D.fk'|, |M.pk'|)
{noformat}
For the specific example above:
{noformat}
|M.pk'| = |M.pk| * sel(M.pk = 10)
= |M.pk| * 1/|M.pk|
= 1
|M'| = |M| * sel(M.pk = 10)
= |M| * 1/|M.pk|
= 1
|D ⋈ M'| = |D| * |M'| / max(|D.fk|, |M.pk'|)
= |D| * 1 / max(|D.fk|, 1)
= |D| / |D.fk|
= |D| / |M|
{noformat}
Which is correct: we'll return a single master record and its details which is
the average number of details per master. The same logic works if we filter on
the detail side:
{noformat}
SELECT ...
FROM master m, detail d
WHERE m.pk = d.fk
AND d.fk = 10
{noformat}
So:
{noformat}
|D.fk'| = |D.fk| * sel(D.fk = 10|
= |D.fk| * 1/|D.fk|
= 1
|D'| = |D| * sel(D.fk = 10|
= |D| / |D.fk|
|D' ⋈ M| = |D'| * |M| / max(|D.fk'|, |M.pk|)
= |D| / |D.fk| / * |M| / |M.pk|
= |D| / |M|
{noformat}
It also works if we filter on both sides (which Impala does implicitly):
{noformat}
SELECT ...
FROM master m, detail d
WHERE m.pk = d.fk
AND m.pk = 10
AND d.fk = 10
{noformat}
Some of the steps are the same as earlier, let's focus on the join:
{noformat}
|D' ⋈ M'| = |D'| * |M'| / max(|D.fk'|, |M.pk'|)
= (|D| / |D.fk|) * 1 / max(1, 1)
= |D| / |M|
{noformat}
Now, suppose we place a broad filter on the master table:
{noformat}
SELECT ...
FROM master m, detail d
WHERE m.pk = d.fk
AND m.balance > 1234
{noformat}
Let's apply the same math, assuming {{sel(m.pk > 1234) = 1/3}}.
{noformat}
|M'| = |M| * sel(m.balance > 1234)
= |M| / 3
|M.pk'| = |M.pk| * sel(m.balance > 1234)
= |M.pk| / 3
= |M'|
|D ⋈ M'| = |D| * |M'| / max(|D.fk|, |M.pk'|)
= |D| * (|M| / 3) / max(|D.fk|, |M.pk| / 3)
= (|D| / 3) * |M| / |D.fk|
= |D| / 3
{noformat}
Remembering, again, that {{|D.fk| = |M.pk| = |M|}}.
h4. Compound Keys with Correlated Filters
If the join key is compound (two or more keys), then we apply the selectivity
to either each key as we multiply them, or the the final product:
{noformat}
|T.k'| = ∏ |T.ki| * sel(T)
= sel(T) * ∏ |T.ki|
{noformat}
That is, we assume that the filter reduces the cardinality of all keys equally.
This may require more thought to verify.
The key cardinality is available in metadata. The predicate selectivity is not
currently available. Since the scan computes it to compute the scan
cardinality, the scan node can simply save the selectivity for use by the join.
Plugging this into Equation 6:
{noformat}
|L'| * |R'|
|L' ⋈ R'| = ---------------------------------------- [Equation 8]
max(sel(L) * ∏ |L.ki|, sel(R) * ∏ |R.ki|)
{noformat}
With that, it is a simple matter to compute the join cardinality.
h4. Generalizing the Correlated Filter Case
A complication then occurs. The next join will have the present join as one of
its input relations. What selectivity should this join report?
We need the following:
* Table cardinalities: {{|L|}} and {{|R|}}, which can be base tables or joins.
* The local predicates for each table, {{sel(L)}} and {{sel(|R|}} along with
their selectivities.
* The NDV values for the join predicates {{|L.k1|}} and {{|R.k2|}}.
>From these we can work out the math. As it turns out, the scan nodes already
>compute the scan cardinalities for us: {{|L'|}} and {{|R'|}}. The join must
>work out the other terms.
The key cardinalities are worked out by applying the scan predicate selectivity
to the join key NDVs:
{noformat}
|T.k'| = |T.k| * sel(T)
|D'| * |M'|
|D' ⋈ M'| = ---------------------
max(|D.fk'|, |M.pk'|)
|D| * |M|
sel(join) = ---------
|D' ⋈ M'|
|D'| = |D| * sel(D)
|M'| = |M| * sel(M)
{noformat}
The top term then becomes:
{noformat}
|D| * |M| = (|D'| / sel(D)) * |M'| / sel(M))
{noformat}
So the whole equation is:
{noformat}
sel(join) = (|D'| / sel(D)) * |M'| / sel(M)) / |D' ⋈ M'|
{noformat}
Or
{noformat}
|D'| * |M'|
sel(join) = --------------------------- [Equation 9]
|D' ⋈ M'| * sel(D) * sel(M)
{noformat}
All the above terms are available in the planner. The selectivity computed
above then becomes the input to the next join.
h4. M:N (Generic) Case
Join discussions are more intuitive when discussed in the M:1 (FK/PK) case. But
the math works just as well for the M:N (generic) case.
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.k1 = R.k2}}. We can now use a hash
join in which L is on the probe side and R is on the build side.
Because we are concerned with the M:N (generic) case, we assume that {{|R.k2| <
|R|}}. This means that multiple build rows have the same key, say {{R.k2 = x}}.
This then implies that each row of the L (probe) side will potentially match
multiple rows on the R (build) side.
We want to know, how many rows will each probe-side row match?
Let’s focus on a table-to-table join and assume we can obtain the following
from HMS:
* Probe and build table cardinalities: {{|L|}} and {{|R|}}
* Key cardinalities (NDVs): {{|L.k1|}} and {{|R.k2|}}
The probe side will match rows on the build side where {{R 𝜎 R.k2 = L.k1}}.
Using the uniformity assumption from the S&S paper (see IMPALA-8014) we can see
that the the values of R.k1 divide the R table into a set of {{|R.k2|}} groups,
the size of each must be:
{noformat}
|R'| = |R| / |R.k2|
{noformat}
Let’s assume that every row on the L side matches some row on the R side. Then,
the join cardinality is just:
{noformat}
|L ⋈ R| = |L| * |R| / |R.k2|
{noformat}
Both the L and R tables may be subject to selection during scan (see
IMPALA-8014) that is an unbiased sampling (given the uniformity assumption) of
the rows of both tables. This reduces the rows available to join, but does not
reduce the population of groups from which the sample is drawn. So:
{noformat}
|L' ⋈ R'| = |L'| * |R'| / |R.k2|
{noformat}
Intuitively, however may rows are scanned, they are still divided into the same
set of groups.
The above assumes that all rows from L match rows in R. (This is called the
“containment assumption” in the S&S paper.) But, Big Data is messy. Perhaps
there are more key values in L than R or visa-versa. We can make some
reasonable assumptions:
* If there are fewer values in L.k1 than in R.k2, we can assume all probe rows
will match a build key.
* If there are more values in L.k1 than in R.k2, we can assume we'll match all
keys on the build side, then discard the extra probe values that don't match.
Again using the uniformity assumption, the probability is simply the ratio of
the the number of keys available for matching (the right or probe side) divided
by the number of keys we want to match (the left or probe side):
{noformat}
p(match) = / |R.k2| / |L.k1| if |L.k1| > |R.k2|,
\ 1 otherwise
= |R.k2| / max( |L.k1|, |R.k2| )
{noformat}
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 probe keys ({{|L.k1|}}) is half that of build keys ({{|R.k2|}}) then all
probe rows will find a mach, so the probability is 1 (though half of the build
side rows will go unmatched.)
* If we have twice as many probe keys {{|L.k1|}} as build keys ({{|R.k2|}})
then half probe rows won’t find a match and the probability of a match is 0.5.
All good.
Putting it all together:
{noformat}
|L’ ⋈ R’| = |L’| * |R’| / |R.k2| * p(match)
= (|L’| * |R’| / |R.k2|) * |R.k2| / max(|R.k2|, |L.k1|)
= |L’| * |R’| / max(|R.k2|, |L.k1|)
{noformat}
Rearranging terms, we get the M:N cardinality estimation expression:
{noformat}
|L’| * |R’|
|L’ ⋈ R’| = -------------------
max(|L.k1|, |R.k2|)
{noformat}
As it turns out, this is exactly Equation 2 in the S&S paper which provides
confirmation that the derivation is correct. It is also the same (except for
names) as Equation 5 above, showing that the M:1 and M:N cases are the same
mathematically.
Said another way, as the M:N groups on the "N" side get ever smaller, they will
converge on a group size of 1, which is the M:1 case. So, the M:1 (FK/PK)
formula *must* be a special case of the more generic M:N (generic) formula.
h4. Proposed Fix
All of this leads to the proposed fix. To gather the details from the above
discussion, modify the planner as follows:
* Add node selectivity as state on each node that can be obtained by the join
node.
* For the Scan node, retain the selectivity already computed.
* For the Join node, compute selectivity as shown below.
* Modify the Join node to compute join cardinality using the equations below.
We assume the following are available from the input nodes:
* {{|T'|}}, the node output cardinality
* {{sel(T)}}, the selectivity of that node
{noformat}
|L.k'| = min(sel(L) * ∏ |L.ki'|, |L'|)
|R.k'| = min(sel(R) * ∏ |R.ki'|, |R'|)
|L'| * |R'|
|L' ⋈ R'| = -------------------
max(|L.k'|, |R.k'|)
|L'| * |R'|
sel(join) = ---------------------------
|L' ⋈ R'| * sel(L) * sel(R)
{noformat}
was:
The math is wrong when we estimate the cardinality of a join we've labeled as
"FK/PK" (commonly known as many-to-one or M:1.)
TL;DR: skip to the end for the proposed change. See IMPALA-8018 for the
detailed explanation of the correct math.
h4. Join Estimation in Impala
Join cardinality is estimated in {{JoinNode.computeStats()}}, and especially in
{{getJoinCardinality()}}, which implements the infamous "FK/PK" heuristics.
"The FK/PK detection logic is based on the assumption that most joins are
FK/PK. ... In the absence of relevant stats, we assume FK/PK with a join
selectivity of 1." As noted above, M:1 is more common name for "FK/PK".
During execution, the code first ensures that the larger table is on the probe
(left) side. So, {{getJoinCardinality()}} assumes that the L is the detail
table with the foreign key (FK), while the smaller build-side table is the
master table with the primary key (PK). That is L:R is M:1, joined on {{L.<FK>
= R.<PK>}}.
Next, let's translate the code's naming to the terms defined in IMPALA-8018.
A comment in the code explains the cardinality estimation model:
{noformat}
cardinality = |child(0)| * (|child(1)| / |R|) * (NDV(R.d) / NDV(L.c))
{noformat}
We can check how the code implements the expression:
{code:java}
ndvRatio = slots.rhsNdv() / slots.lhsNdv();
double rhsSelectivity = rhsCard / slots.rhsNumRows();
long joinCard = (long) Math.ceil(lhsCard * rhsSelectivity * ndvRatio);
{code}
So, the comment accurately describes the code.
Let’s translate the code’s expression to the terms defined above:
{noformat}
|L'| |R'|
----^----- ----^-----
cardinality = |child(0)| * (|child(1)| / |R|) * (NDV(R.d) / NDV(L.c))
{noformat}
Using:
* {{cardinality}} here means the cardinality of the join, hence {{|𝜎D ⋈ 𝜎M|}}
* {{L}} = Left = Probe table (before scan) = Detail table, hence {{D}}
* {{R}} = Right = Build table (before scan) = Master table, hence {{M}}
* {{child(0)}} is the output of the scan the left table or {{D}}, hence {{𝜎D}}
* {{child(1)}} is the output of the scan the right table or {{M}}, hence {{𝜎M}}
* {{R.d}} is the join key of the R (master) table, hence {{M.pk}}
* {{L.c}} is the join key of the L (detail) table, hence {{D.fk}}
* NDV is, as we said, the cardinality of the key column, hence {{|k|}}.
Gives:
{noformat}
|D' ⋈ M'| = |D'| * (|M'| / |M|) * (|M.pk| / |D.fk|). [Sic]
{noformat}
The code expression is actually overly complex. Recall that, by definition,
{{|M| = |M.pk|}}:
{noformat}
|D' ⋈ M'| = |D'| * (|M'| / |M|) * (|M.pk| / |D.fk|)
= |D'| * (|M'| / |D.fk|) * (|M.pk| / |M|)
= |D'| * (|M'| / |D.fk|) * (|M| / |M|)
= |D'| * |M'| / |D.fk|
{noformat}
As we’ll see, this formula is *NOT* correct.
h4. Bug in the Code
Compare IMPALA-8018, Equation 5 to the one in the code:
{noformat}
|D' ⋈ M'| = |D'| * |M'| / |D.fk| — Code
|D' ⋈ M'| = |D'| * |M'| / max(|D.fk|, |M.pk|) — Derived
{noformat}
Notice the code contains a severe error: it divides by the foreign key
cardinality, even when that cardinality is *smaller* than that of the primary
key, causing the join cardinality to be larger than the probe cardinality,
which cannot occur if we assume a master/detail (PK/FK) relationship.
To ensure we are reading all this correctly, let' work out an example. Let's
assume that our detail table has 10K rows, our master table 100 rows, and that
only 25 of the master records are referenced. Let's assume the scans remove no
rows. That is:
* {{|D'| = |D| = 10,000}}
* {{|M'| = |M| = = |M.pk| = 100}}
* {{|D.fk| = 25}}
Using Equation 5:
{noformat}
|D' ⋈ M'| = |D'| * |M'| / max( |D.fk|, |M.pk| )
= |D| * |M| / max( |D.fk|, |M.pk| )
= 10,000 * 100 / max( 100, 25 )
= 10,000 * 100 / 100
= 10,000
{noformat}
Makes sense: all 10K detail rows match some master row, though 3/4 of the
master rows are unreferenced.
Using the formula from the code:
{noformat}
cardinality = |D| * |M| / |D.fk|
= 10,000 * 100 / 25
= 10,000 * 4
= 40,000
{noformat}
As a result, the fewer of the primary keys are referenced, the more we expect
the input rows to be replicated and the greater the (mis)estimated output
cardinality. This, of course, leads to poor join selection and thus a poor plan.
Let's check this empirically by printing out the `ndvRatio` (see code snippet
above) when running some queries:
{noformat}
ndv ratio = 2.0977582147609786
ndv ratio = 3.3183831043488135
ndv ratio = 2.0977582147609786
ndv ratio = 2.0977582147609786
{noformat}
So, this seems like a real bug.
h4. Compound Keys
IMPALA-8018 discusses the correct way to handle compound keys and correlated
filters. Impala has an implementation: in {{JoinNode.getFkPkEqJoinConjuncts()}}:
{code:java}
double jointNdv = 1.0;
for (EqJoinConjunctScanSlots slots: fkPkCandidate) jointNdv *=
slots.rhsNdv();
{code}
However, the calculations become muddled in the actual cardinality estimation
in {{JoinNode.getFkPkJoinCardinality()}}
{code:java}
long result = -1;
for (EqJoinConjunctScanSlots slots: eqJoinConjunctSlots) {
// Adjust the join selectivity based on the NDV ratio to avoid
underestimating
// the cardinality if the PK side has a higher NDV than the FK side.
double ndvRatio = 1.0;
if (slots.lhsNdv() > 0) ndvRatio = slots.rhsNdv() / slots.lhsNdv();
double rhsSelectivity = Double.MIN_VALUE;
if (slots.rhsNumRows() > 0) rhsSelectivity = rhsCard / slots.rhsNumRows();
long joinCard = (long) Math.ceil(lhsCard * rhsSelectivity * ndvRatio);
if (result == -1) {
result = joinCard;
} else {
result = Math.min(result, joinCard);
}
}
{code}
The above tries to work to the key cardinality after selection ({{|T.k'|}}),
but the math is off. Rather than using the actual selectivity from the input
node, the code attempts to use the standard formula to derive the selectivity.
Unfortunately, for join nodes, we cannot recover the original table
selectivity, so the above gives the wrong answers.
h4. Proposed Fix
All of this leads to the proposed fix, which is to implement the changes
proposed in IMPALA-8018.
> Unify cardinality math for PK/FK and generic cases
> --------------------------------------------------
>
> Key: IMPALA-8018
> URL: https://issues.apache.org/jira/browse/IMPALA-8018
> Project: IMPALA
> Issue Type: Improvement
> Components: Frontend
> Affects Versions: Impala 3.1.0
> Reporter: Paul Rogers
> Priority: Minor
>
> The math is wrong when we estimate the cardinality for both M:1 (FK/PK joins)
> (IMPALA-8014) and for M:N (generic) joins (IMPALA-8015). General practice is
> that the two cases are treated the same as described here.
> TL;DR: skip to the last section for the proposed change. See IMPALA-8014 and
> IMPALA-8015 for details about the bugs in the current code.
> h4. Background
> Join logic is complex. To ensure that the analysis is sound we work it out
> from first principles, then verify against the Swami & Schiefer, [On the
> Estimation of Join Result
> Sizes|https://pdfs.semanticscholar.org/2735/672262940a23cfce8d4cc1e25c0191de7da7.pdf]
> (S&S). Note especially section 3, Background.
> h4. Definitions
> The following terms are used below:
> * _Relation:_ either a _base table_ or the result of a join. (Relational
> theory says that both are simply relations.) We use an upper case letter,
> typically R, for a relation. E.g.: {{R1}}, {{R1}}
> * _Key:_ A column used in a join. We use a lower case letter for a column,
> and typically use “k” for a key, along with the relation number. E.g. {{k1}}
> and {{ON R1.k1a = R2.k2a}}.
> * _Compound key:_ A join key made up of multiple columns. We use a letter to
> denote each column. E.g. {{k1a}} and {{ON R1.k1a = R2.k2a AND R1.k1b =
> R2.k2b}}
> * _Join:_ the join of two relations. Impala supports the join of two base
> tables (a table-to-table join), or the join of a base table and another join
> (a table-to-join join.) We use the symbol {{⋈}} to denote a join. Impala
> typically uses a hash join.
> * _Relation cardinality:_ the number of rows in the relation, denoted
> {{|R|}}.
> * _Join cardinality:_ the cardinality of the relation produced by the join.
> That is, the join’s output cardinality. Denoted as {{|R1 ⋈ R2|}}.
> * _Key column cardinality:_ the number of unique values in a column, denoted
> {{|k|}}. Also known as the NDV of the column or {{ndv(k)}}.
> In Impala, HMS provides the relation and join key cardinality as table and
> column statistics respectively.
> Since Impala typically uses hash joins, it is helpful to use terminology
> specific to that case:
> * _Probe_ side of a join: the larger side (ideally) of a hash join. Also
> called the “left” side. Is known as {{chiild(0)}} in the code. Appears at the
> same level as the join in the query plan (which would be to the left if the
> plan were rotated 90 degrees counter-clockwise.)
> * _Build_ side of a join: the smaller side (ideally) of a hash join. Also
> called the “right” side. Is known as {{child(1)}} in the code. Appears
> indented directly under the join in the query plan.
> In Impala:
> * The detail table in a M:1 relationship is always on the probe (left) side
> of a join. Represented as {{child(0)}} in the code.
> * The master table in a M:1 relationship is always on the build (right) side
> of a join. Represented as {{child(1)}} in the code.
> Finally, we also need:
> * The _scan_ of a table: the (possibly smaller) relation produced by
> applying one or more predicates while scanning a table. We are concerned with
> the cardinality of the scan, denoted as {{|R1'|}}. We assume that Impala has
> already used rules (not discussed here) to estimate the cardinality of the
> selection.
> h4. Deriving the Join Formula For the M:1 Case
> In RDBMS, a primary key (PK) is a column (or, more typically, set of columns)
> that uniquely identify a row in the master table (M). Primary keys are
> generally indexed via a unique index. Since keys are unique:
> {noformat}
> |M.pk| = |M|
> {noformat}
> The Detail table forms a M:1 relationship with the master table. Each foreign
> key (FK) in the detail table (D) references one primary key in the master
> table. Because of the M:1 relationship:
> {noformat}
> |D.fk| <= |M| << |D|
> {noformat}
> If we read all rows from both tables, and all primary keys appear as foreign
> keys, the the join cardinality is simply:
> {noformat}
> |D ⋈ M| = |D| [Equation 1]
> {noformat}
> h5. Filtering Master Rows
> Let’s consider what happens when we filter out master (M) table rows
> producing a subset M'. To do so, we make three assumptions:
> * A uniform distribution of foreign keys, (the “Uniformity” assumption in the
> S&S paper cited above),
> * Every primary key is referenced by an equal number of foreign keys (implied
> by the uniformity assumption),
> * The filter results in a random sampling of master rows not correlated with
> the join keys.
> A later section will relax the third assumption.
> Then the probability of any particular primary key appearing is:
> {noformat}
> p(pk) = |M'| / |M|
> {noformat}
> The result value is a probability (hence the {{p()}} function) given by the
> ratio of selected rows to total table rows (from basic probability theory.)
> The value runs from 0 (no master rows match the scan predicates) to 1 (all
> rows present).
> The revised join cardinality is:
> {noformat}
> |D ⋈ M'| = |D| * p(pk)
> = |D| * |M'| / |M| [Equation 2]
> {noformat}
> h5. Filtering Detail Rows
> Suppose we instead filter detail rows to produce a new subset D'. Again we
> make some assumptions:
> * Uniform distribution of foreign key values across primary keys.
> * The "Containment" assumption from the S&S paper that the set of foreign
> keys is a (possibly full) subset of the set of primary keys.
> * The filter results in a random sampling of detail rows not correlated with
> the join keys.
> Again, a later section will relax the third assumption.
> With these assumptions, we can see that if we join D' with the master table
> M, every row that remains in D' will still find a match in M, so:
> {noformat}
> |D' ⋈ M| = |D'|
> {noformat}
> h5. Filtering Both Master and Detail Rows
> Continuing with the assumption that filters are not correlated with keys, we
> can combine the two selection models to produce:
> {noformat}
> |D' ⋈ M'| = |D'| * |M'| / |M| [Equation 3]
> {noformat}
> h5. Non-Containment
> The above is based on the S&S Containment assumption: that the set of foreign
> keys is a subset of the set of primary keys. In a normal RDBMS with integrity
> constraints, this is a valid assumption. But, in Big Data, things are messy
> and we can’t actually make this assumption. Fortunately, there is a way
> loosen the containment assumption.
> Suppose the master table has half the number of keys as the detail table.
> Using the uniformity assumption, half the foreign keys will go unmatched. If
> there are four times as many foreign keys as primary keys, only a quarter of
> the detail rows will find matches. The probability of any one foreign key
> finding a match is:
> {noformat}
> p(match) = |M.pk| / |D.fk|
> {noformat}
> The above applies only if {{|D.fk| > |M.pk|}}; no adjustment is needed if
> {{|D.fk| < |M.pk|}}. We can express this mathematically as:
> {noformat}
> p(match) = |M.pk| / max(|D.fk|, |M.pk|) [Equation 4]
> {noformat}
> Combining this with Equation 3, recalling that {{|M.pk| = |M|}}:
> {noformat}
> |D' ⋈ M'| = |D'| * (|M'| / |M|) * p(match)
> = |D'| * |M'| / |M| * |M.pk| / max(|D.fk|, |M.pk|)
> = |D'| * |M'| * (|M| / |M|) / max(|D.fk|, |M.pk|)
> = |D'| * |M'| / max(|D.fk|, |M|)
> {noformat}
> If we rearrange the above we get:
> {noformat}
> |D'| * |M'|
> |D' ⋈ M'| = ---------------- [Equation 5]
> max(|D.fk|, |M|)
> {noformat}
> h4. The S&S Equation
> Let’s check out answer against the S&S paper cited above by assuming the
> master/detail relationship and removing from our equation the affect of
> scans. The paper’s equation:
> {noformat}
> |R1 ⋈ R2| = min( |k1|, |k2| ) * ( |R1| / |k1| ) * ( |R2| * |k2| )
> = |R1| * |R2| / max( |k1|, |k2| )
> {noformat}
> If we assume:
> * {{R1 = D}}
> * {{R2 = M}}
> * {{k1 = D.fk}}
> * {{k2 = M.pk}}
> * {{|D.fk| > |M.pk|}}
> We get:
> {noformat}
> |D ⋈ M| = |D| * |M| / max( |D.fk|, |M.pk| )
> {noformat}
> Which is the same as Equation 5 (assuming no selection). All good.
> h4. Compound Keys
> The discussion thus far has assumed a single key column per table so that we
> have NDV values for each. The next challenge occurs if the query has compound
> columns:
> {noformat}
> SELECT …
> FROM T1, T2
> WHERE T1.k1a = T2.k2a AND T1.k1b = T2.k2b
> {noformat}
> We will use the symbol {{j}} to mean a joint (compound) key. The joint key
> for table {{T1}} is {{T1.j1}} and {{||T1.j1|}} is the cardinality (NDV) of
> that key.
> In an RDBMS, if an index exists on (k1a, k1b) and (k2a, k2b) then it is
> possible to determine the joint NDV (which is just the number of index
> entries.) In HMS, we have the individual NDVs, but not the joint NDV.
> The S&S paper supplies the require assumption: “Independence: Within each
> relation, values chosen from distinct columns that are involved in join
> predicates are independent.” This lets us estimate the joint NDV:
> One way to think of this is the the columns form a matrix, say (year, month).
> Every year has 12 months, so there is no correlation between the two. If we
> have data for 10 years, we will have 10 * 12 = 120 distinct key values.
> {noformat}
> |T.j| = |T.k1| * |T.k2| * … * |T.kn|
> = ∏ |T.ki|
> {noformat}
> Where {{∏}} is the product operator. (S&S, section 3.3, calls this the
> "Multiplicative Rule.")
> Of course, keys are not always independent. Suppose the keys are (state,
> city). In this case, there are 50 states. But most city names occur in only
> one state, so NDV(state) * NDV(city) >> NDV(state, city). Still, we know
> that, regardless of the correlation, we can never have more keys than rows.
> We can adjust the expression to:
> {noformat}
> |T.j| = min( ∏ |T.ki|, |T| )
> {noformat}
> We can also assume that joins are typically formulated to be as selective as
> possible. So the assumption may not be too far off. Still, if a file contains
> addresses, so the key is (state, city, street), and we just the (state, city)
> prefix, the joint NDV will smaller than the table cardinality. So we need
> both terms above.
> Adding this to the expression derived above:
> {noformat}
> |D'| * |M'|
> |D' ⋈ M'| = --------------------------------------------- [Equation
> 6]
> max(min(∏ |D.fki|, |D|), min(∏ |M.pki|, |M|))
> {noformat}
> h4. Correlated Filters
> Suppose that we relax the assumption that filters are not correlated with
> join keys and assume they are. For example:
> {noformat}
> SELECT ...
> FROM master m, detail d
> WHERE m.pk = d.fk
> AND m.pk = 10
> {noformat}
> The simplified math used above no longer applies: the filter affects the
> number of join keys available. Applying Equation 5 in this case results in a
> "double accounting" for the primary key filter.
> Equation 5 has four terms. Consider a predicate on the master table:
> * The predicate reduces the number of table rows from {{|M|}} to {{|M'|}}.
> * The predicate also reduces the number of keys from {{|M.pk| (= |M|)}} to
> {{|M.pk'| (= |M'|)}}.
> The above equation does not consider the reduction in key cardinality. Let's
> add that now:
> {noformat}
> |D'| * |M'|
> |D' ⋈ M'| = --------------------- [Equation 7]
> max(|D.fk'|, |M.pk'|)
> {noformat}
> For the specific example above:
> {noformat}
> |M.pk'| = |M.pk| * sel(M.pk = 10)
> = |M.pk| * 1/|M.pk|
> = 1
> |M'| = |M| * sel(M.pk = 10)
> = |M| * 1/|M.pk|
> = 1
> |D ⋈ M'| = |D| * |M'| / max(|D.fk|, |M.pk'|)
> = |D| * 1 / max(|D.fk|, 1)
> = |D| / |D.fk|
> = |D| / |M|
> {noformat}
> Which is correct: we'll return a single master record and its details which
> is the average number of details per master. The same logic works if we
> filter on the detail side:
> {noformat}
> SELECT ...
> FROM master m, detail d
> WHERE m.pk = d.fk
> AND d.fk = 10
> {noformat}
> So:
> {noformat}
> |D.fk'| = |D.fk| * sel(D.fk = 10|
> = |D.fk| * 1/|D.fk|
> = 1
> |D'| = |D| * sel(D.fk = 10|
> = |D| / |D.fk|
> |D' ⋈ M| = |D'| * |M| / max(|D.fk'|, |M.pk|)
> = |D| / |D.fk| / * |M| / |M.pk|
> = |D| / |M|
> {noformat}
> It also works if we filter on both sides (which Impala does implicitly):
> {noformat}
> SELECT ...
> FROM master m, detail d
> WHERE m.pk = d.fk
> AND m.pk = 10
> AND d.fk = 10
> {noformat}
> Some of the steps are the same as earlier, let's focus on the join:
> {noformat}
> |D' ⋈ M'| = |D'| * |M'| / max(|D.fk'|, |M.pk'|)
> = (|D| / |D.fk|) * 1 / max(1, 1)
> = |D| / |M|
> {noformat}
> Now, suppose we place a broad filter on the master table:
> {noformat}
> SELECT ...
> FROM master m, detail d
> WHERE m.pk = d.fk
> AND m.balance > 1234
> {noformat}
> Let's apply the same math, assuming {{sel(m.pk > 1234) = 1/3}}.
> {noformat}
> |M'| = |M| * sel(m.balance > 1234)
> = |M| / 3
> |M.pk'| = |M.pk| * sel(m.balance > 1234)
> = |M.pk| / 3
> = |M'|
> |D ⋈ M'| = |D| * |M'| / max(|D.fk|, |M.pk'|)
> = |D| * (|M| / 3) / max(|D.fk|, |M.pk| / 3)
> = (|D| / 3) * |M| / |D.fk|
> = |D| / 3
> {noformat}
> Remembering, again, that {{|D.fk| = |M.pk| = |M|}}.
> h4. Compound Keys with Correlated Filters
> If the join key is compound (two or more keys), then we apply the selectivity
> to either each key as we multiply them, or the the final product:
> {noformat}
> |T.k'| = ∏ |T.ki| * sel(T)
> = sel(T) * ∏ |T.ki|
> {noformat}
> That is, we assume that the filter reduces the cardinality of all keys
> equally. This may require more thought to verify.
> The key cardinality is available in metadata. The predicate selectivity is
> not currently available. Since the scan computes it to compute the scan
> cardinality, the scan node can simply save the selectivity for use by the
> join.
> Plugging this into Equation 6:
> {noformat}
> |L'| * |R'|
> |L' ⋈ R'| = ---------------------------------------- [Equation 8]
> max(sel(L) * ∏ |L.ki|, sel(R) * ∏ |R.ki|)
> {noformat}
> With that, it is a simple matter to compute the join cardinality.
> h4. Generalizing the Correlated Filter Case
> A complication then occurs. The next join will have the present join as one
> of its input relations. What selectivity should this join report?
> We need the following:
> * Table cardinalities: {{|L|}} and {{|R|}}, which can be base tables or
> joins.
> * The local predicates for each table, {{sel(L)}} and {{sel(|R|}} along with
> their selectivities.
> * The NDV values for the join predicates {{|L.k1|}} and {{|R.k2|}}.
> From these we can work out the math. As it turns out, the scan nodes already
> compute the scan cardinalities for us: {{|L'|}} and {{|R'|}}. The join must
> work out the other terms.
> The key cardinalities are worked out by applying the scan predicate
> selectivity to the join key NDVs:
> {noformat}
> |T.k'| = |T.k| * sel(T)
> |D'| * |M'|
> |D' ⋈ M'| = ---------------------
> max(|D.fk'|, |M.pk'|)
> |D| * |M|
> sel(join) = ---------
> |D' ⋈ M'|
> |D'| = |D| * sel(D)
> |M'| = |M| * sel(M)
> {noformat}
> The top term then becomes:
> {noformat}
> |D| * |M| = (|D'| / sel(D)) * |M'| / sel(M))
> {noformat}
> So the whole equation is:
> {noformat}
> sel(join) = (|D'| / sel(D)) * |M'| / sel(M)) / |D' ⋈ M'|
> {noformat}
> Or
> {noformat}
> |D'| * |M'|
> sel(join) = --------------------------- [Equation 9]
> |D' ⋈ M'| * sel(D) * sel(M)
> {noformat}
> All the above terms are available in the planner. The selectivity computed
> above then becomes the input to the next join.
> h4. M:N (Generic) Case
> Join discussions are more intuitive when discussed in the M:1 (FK/PK) case.
> But the math works just as well for the M:N (generic) case.
> 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.k1 = R.k2}}. We can now use a
> hash join in which L is on the probe side and R is on the build side.
> Because we are concerned with the M:N (generic) case, we assume that {{|R.k2|
> < |R|}}. This means that multiple build rows have the same key, say {{R.k2 =
> x}}. This then implies that each row of the L (probe) side will potentially
> match multiple rows on the R (build) side.
> We want to know, how many rows will each probe-side row match?
> Let’s focus on a table-to-table join and assume we can obtain the following
> from HMS:
> * Probe and build table cardinalities: {{|L|}} and {{|R|}}
> * Key cardinalities (NDVs): {{|L.k1|}} and {{|R.k2|}}
> The probe side will match rows on the build side where {{R 𝜎 R.k2 = L.k1}}.
> Using the uniformity assumption from the S&S paper (see IMPALA-8014) we can
> see that the the values of R.k1 divide the R table into a set of {{|R.k2|}}
> groups, the size of each must be:
> {noformat}
> |R'| = |R| / |R.k2|
> {noformat}
> Let’s assume that every row on the L side matches some row on the R side.
> Then, the join cardinality is just:
> {noformat}
> |L ⋈ R| = |L| * |R| / |R.k2|
> {noformat}
> Both the L and R tables may be subject to selection during scan (see
> IMPALA-8014) that is an unbiased sampling (given the uniformity assumption)
> of the rows of both tables. This reduces the rows available to join, but does
> not reduce the population of groups from which the sample is drawn. So:
> {noformat}
> |L' ⋈ R'| = |L'| * |R'| / |R.k2|
> {noformat}
> Intuitively, however may rows are scanned, they are still divided into the
> same set of groups.
> The above assumes that all rows from L match rows in R. (This is called the
> “containment assumption” in the S&S paper.) But, Big Data is messy. Perhaps
> there are more key values in L than R or visa-versa. We can make some
> reasonable assumptions:
> * If there are fewer values in L.k1 than in R.k2, we can assume all probe
> rows will match a build key.
> * If there are more values in L.k1 than in R.k2, we can assume we'll match
> all keys on the build side, then discard the extra probe values that don't
> match.
> Again using the uniformity assumption, the probability is simply the ratio of
> the the number of keys available for matching (the right or probe side)
> divided by the number of keys we want to match (the left or probe side):
> {noformat}
> p(match) = / |R.k2| / |L.k1| if |L.k1| > |R.k2|,
> \ 1 otherwise
> = |R.k2| / max( |L.k1|, |R.k2| )
> {noformat}
> 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 probe keys ({{|L.k1|}}) is half that of build keys ({{|R.k2|}}) then all
> probe rows will find a mach, so the probability is 1 (though half of the
> build side rows will go unmatched.)
> * If we have twice as many probe keys {{|L.k1|}} as build keys ({{|R.k2|}})
> then half probe rows won’t find a match and the probability of a match is
> 0.5.
> All good.
> Putting it all together:
> {noformat}
> |L’ ⋈ R’| = |L’| * |R’| / |R.k2| * p(match)
> = (|L’| * |R’| / |R.k2|) * |R.k2| / max(|R.k2|, |L.k1|)
> = |L’| * |R’| / max(|R.k2|, |L.k1|)
> {noformat}
> Rearranging terms, we get the M:N cardinality estimation expression:
> {noformat}
> |L’| * |R’|
> |L’ ⋈ R’| = -------------------
> max(|L.k1|, |R.k2|)
> {noformat}
> As it turns out, this is exactly Equation 2 in the S&S paper which provides
> confirmation that the derivation is correct. It is also the same (except for
> names) as Equation 5 above, showing that the M:1 and M:N cases are the same
> mathematically.
> Said another way, as the M:N groups on the "N" side get ever smaller, they
> will converge on a group size of 1, which is the M:1 case. So, the M:1
> (FK/PK) formula *must* be a special case of the more generic M:N (generic)
> formula.
> h4. Proposed Fix
> All of this leads to the proposed fix. To gather the details from the above
> discussion, modify the planner as follows:
> * Add node selectivity as state on each node that can be obtained by the join
> node.
> * For the Scan node, retain the selectivity already computed.
> * For the Join node, compute selectivity as shown below.
> * Modify the Join node to compute join cardinality using the equations below.
> We assume the following are available from the input nodes:
> * {{|T'|}}, the node output cardinality
> * {{sel(T)}}, the selectivity of that node
> {noformat}
> |L.k'| = min(sel(L) * ∏ |L.ki'|, |L'|)
> |R.k'| = min(sel(R) * ∏ |R.ki'|, |R'|)
> |L'| * |R'|
> |L' ⋈ R'| = -------------------
> max(|L.k'|, |R.k'|)
> |L'| * |R'|
> sel(join) = ---------------------------
> |L' ⋈ R'| * sel(L) * sel(R)
> {noformat}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
