[jira] [Created] (IMPALA-8218) Use "simple bin model" to estimate M:N, FK join cardinality

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

Paul Rogers created IMPALA-8218:
-----------------------------------

             Summary: Use "simple bin model" to estimate M:N, FK join cardinality
                 Key: IMPALA-8218
                 URL: https://issues.apache.org/jira/browse/IMPALA-8218
             Project: IMPALA
          Issue Type: Improvement
          Components: Frontend
    Affects Versions: Impala 3.1.0
            Reporter: Paul Rogers
            Assignee: Paul Rogers


When computing join cardinality, the planner must determine how much a filters reduce the cardinality of join keys. For example, in a M:1 (FK/PK) join, filtering on the left (M, FK) side will reduce the number of rows available to join. How much does that filtering reduce the keys?

The "selectivity" of a filter is the probability that any one row will pass through the filter:

{noformat}
|T'| = |T| * sel(f)

sel(f) = |T'| / |T|
{noformat}

Where:

* \{{|T|}} is the cardinality of some table or relation T.
* \{{|T'|}} is the cardinality of the new relation T' after filtering.
* \{{sel(f)}} is the selectivity of some filter f.

The current model makes the standard assumption that the NDV of key columns reduces by the same amount as the table cardinality. That is:

{noformat}
|T.k'| = |T.k| * sel(f)
{noformat}

This assumption is incorrect as explained below. The correct expression is explained in Swami & Schiefer (S&S), [On the Estimation of Join Result Sizes|https://pdfs.semanticscholar.org/2735/672262940a23cfce8d4cc1e25c0191de7da7.pdf] (S&S), section 5.


h4. Motivation

It is easiest to reason about these issues in the M:1 (FK/PK) case, but the logic also applies to both sides of the M:N (many-to-many, generic) case.

Consider an example. We have a M:1 relationship with two tables: D (for detail) and M (for master). A M:1 relationship exists:

{noformat}
D.fk --> M.pk
{noformat}

The foreign key (fk) in the detail table points to the primary key (pk) in the master table. Note that keys in the master table are unique, so the simple filtering expression works fine:

{noformat}
|M.pk'| = |M.pk| * sel(f)
{noformat}

On the detail side, there are multiple rows that have the same key. What we want to know is, when we apply the filtering above, how many of the foreign keys survive? Since there are multiple keys, the answer is not obvious. To see, this, let's assume that the detail table has 1000 rows, and uses only one FK value. If we select 500 rows, how many foreign keys are left? Obviously, the answer is that there is still one FK value.


h4. Correlated Filtering

There is one case where simple filtering is the right answer: if the filter is on the key column itself. Suppose we know that the left (detail) scan applied a filter on the foreign key column. Maybe \{{D.fk = 123}}. In this case (as explained in the S&S paper above) we know that \{{|D.fk'| = 1}}. In the general case, we may have any operator (!=, <, etc.) in the filter so the cardinality of the foreign key is simply the result of applying that filter:

{noformat}
|D.fk'| = |D.fk| * sel(f)
{noformat}


h4. Uncorrelated Filtering

In the next case, we filter on columns other than the primary key column. For simplicity, let us adopt the uniformity assumption from S&S paper: that applying a filter on a column other than the key results in a random sampling of the key column. As explained in S&S, section 5, we must use the "simple urn model":

{noformat}
|T.k'| = urn( |T.k|, |T'| )

url(c, n) = c * (1 - (1 - 1/c)^n)
{noformat}

Where:

* \{{urn(c, n)}} is the urn model expression which gives the estimated cardinality of a column as the result of a scan
* \{{c}} is the cardinality of a key column
* \{{n}} is the cardinality of the output relation of a scan

See the paper for the reasoning behind this expression.


h4. Combined Filtering

To make the above work, we observe that, in relational theory, we get the same result whether we apply all filters in one go, or apply them one-by-one. We get the same result if we apply the filters during the scan (as Impala does), or by first scanning all rows, then applying the filters afterward (as some other engines do.)

This observation allows us to break the calculation into two parts:

* Determine the key NDV and table cardinality produced by just the correlated filters.
* Use the urn model to predict key cardinality from the uncorrelated filters.

We start by sorting scan filters into two categories:

* Correlated filters applied to a given key column (here, \{{D.fk}}).
* All other filters (the uncorrelated filters.)

We do this by first gathering all predicates applied to the detail (left) relation. The detail for doing so is explained in IMPALA-8217 which describes the need to adjust for filters applied to both sides of a join. Given that set, we can do the following:

* Remove from the set all those predicates which reference \{{D.fk}}. These are the correlated predicates.
* Remove correlated filtering from the scan cardinality to get the cardinality due just to the uncorrelated predicates.
* Apply correlated filtering to the key column NDV to get the reduced NDV after filtering.
* Apply the urn model to compute key cardinality after uncorrelated filtering.

Suppose we have the following:

* \{{F(D)}}: the set of all filters applied to the scan of D: \{{(f1(D), f2(D), ... fn(D))}}.
* \{{F(D.fk)}}: the subset of the above that apply to only the key column \{{D.fk}}.

Then we can compute or define the selectivity of these filters:

{noformat}
sel(scan(D)) = ∏ sel(Fi(D))

= |D'| / |D|

sel(F(D.fk)) = ∏ sel(Fi(D.fk))
{noformat}

Next we can work out the cardinalities required for inputs to the urn model:

{noformat}
|D.fk''| = |D.fk| * sel(F(D.fk))

|D''| = |D| * sel(F(D.fk))

= |D'| / sel(scan(D) * sel(F(D.fk))
{noformat}

Where:

* \{{|D.fk''|}} is the key cardinality obtained by applying just the correlated filters.
* \{{|D''|}} is the scan cardinality that would result if we applied only the correlated filters.

Then, we can apply the uncorrelated filtering using the urn model:

{noformat}
|D.fk'| = urn( |D.fk''|, |D''| )
{noformat}


h4. Compound Keys

The discussion above is for simple keys. If a key is compound (see IMPALA-8014) we can refine the above as follows:

* Compute each column in the key as described above.
* Multiply the adjusted NDV's to get the cardinality of the key as a whole, adjusting as explained in IMPALA-8014.


h4. Complication: Compound Joins

The calculation above is sound, but is complicated by several factors in Impala's implementation. First is that the left side of a join is usually not a table; it is usually another join. (This is a result of the left-deep pattern adopted by most query optimizers.) Because of this, we actually do not know the original table cardinality which we've called \{{|D|}}. Yes, we do know the size of the tables that went into a join, and we know the size of the relation that comes out of the join, but there is no reasonable number for the base table since there are multiple.

Instead, the above calculations have been based on what we do know:

* \{{|T'|}} the result of all filtering (and joins). This is the cardinality input to the present join.
* \{{F}}, the set of all filters applied anywhere in the subtree below a join input.
* \{{|D.fk|}}, the key column NDV as reported from HMS stats.

Using just these values we can calculate key cardinality as:

{noformat}
|D.fk''| = |D.fk| * sel(F(D.fk))

|D''| = |D'| / sel(scan(D) * sel(F(D.fk))

|D.fk'| = urn( |D.fk''|, |D''| )
{noformat}


h4. Tracking Adjusted NDV

The above recomputes adjusted NDV at each join from first principals. It is possible to simplify the task by working only one level at a time. Each scan or join would maintain a running adjusted NDV for each column:

* The scan node starts with the input NDVs as reported by HMS.
* The scan node computes output NDVs by applying filters to each column and tracking the result.
* The join node uses the scan output NDVs as |T.k''| (the NDV after correlated filtering.)
* The join node tracks its own output NDVs that result from applying uncorrelated filtering.

With this approach, changes bubble up the operator tree one step at a time. Debugging is easier since we can visualize the adjusted NDVs.

In the approach described in above sections, we track the combined set of all filters applied up the tree. With the approach described in this section, we track the result of applying the filters rather than the filters themselves.


h4. Complication: Non-Linear Filter Combination

The approach above assumes we can apply filters in any order which is a foundational assumption of relational theory. Unfortunately, Impala takes a different approach: it uses an exponential back-off:

{noformat}
|T'| = |T| * ∏(i =0..) Fi^i
{noformat}

This means that the filters cannot be applied in any order, nor can we easily back one filter out of the combined filter.

Impala does this to compensate, in part, for the fact that Impala does not compute selectivity for any but simple \{{col = const}} predicate.

To allow the calculations above, we must remove the exponential back-off, which requires computing selectivity for all predicates -- something that is a good idea anyway. See IMPALA-8217.


h4. Generalizing to the M:N (Generic) Case

The discussion above discusses foreign keys in a M:1 join. The same logic applies to both sides of a join in a M:N (many-to-many, "generic") join. Calculations for the right-side table are a bit simpler because, in Impala, the right input is always a base table.



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