Re: Bloom filter Pushdown Optimization for Merge Join

On Fri, Sep 30, 2022 at 9:20 PM Zhihong Yu <zyu(at)yugabyte(dot)com> wrote:

>
>
> On Fri, Sep 30, 2022 at 8:40 PM Zhihong Yu <zyu(at)yugabyte(dot)com> wrote:
>
>>
>>
>> On Fri, Sep 30, 2022 at 3:44 PM Zheng Li <zhengli10(at)gmail(dot)com> wrote:
>>
>>> Hello,
>>>
>>> A bloom filter provides early filtering of rows that cannot be joined
>>> before they would reach the join operator, the optimization is also
>>> called a semi join filter (SJF) pushdown. Such a filter can be created
>>> when one child of the join operator must materialize its derived table
>>> before the other child is evaluated.
>>>
>>> For example, a bloom filter can be created using the the join keys for
>>> the build side/inner side of a hash join or the outer side of a merge
>>> join, the bloom filter can then be used to pre-filter rows on the
>>> other side of the join operator during the scan of the base relation.
>>> The thread about “Hash Joins vs. Bloom Filters / take 2” [1] is good
>>> discussion on using such optimization for hash join without going into
>>> the pushdown of the filter where its performance gain could be further
>>> increased.
>>>
>>> We worked on prototyping bloom filter pushdown for both hash join and
>>> merge join. Attached is a patch set for bloom filter pushdown for
>>> merge join. We also plan to send the patch for hash join once we have
>>> it rebased.
>>>
>>> Here is a summary of the patch set:
>>> 1. Bloom Filter Pushdown optimizes Merge Join by filtering rows early
>>> during the table scan instead of later on.
>>> -The bloom filter is pushed down along the execution tree to
>>> the target SeqScan nodes.
>>> -Experiments show that this optimization can speed up Merge
>>> Join by up to 36%.
>>>
>>> 2. The planner makes the decision to use the bloom filter based on the
>>> estimated filtering rate and the expected performance gain.
>>> -The planner accomplishes this by estimating four numbers per
>>> variable - the total number of rows of the relation, the number of
>>> distinct values for a given variable, and the minimum and maximum
>>> value of the variable (when applicable). Using these numbers, the
>>> planner estimates a filtering rate of a potential filter.
>>> -Because actually creating and implementing the filter adds
>>> more operations, there is a minimum threshold of filtering where the
>>> filter would actually be useful. Based on testing, we query to see if
>>> the estimated filtering rate is higher than 35%, and that informs our
>>> decision to use a filter or not.
>>>
>>> 3. If using a bloom filter, the planner also adjusts the expected cost
>>> of Merge Join based on expected performance gain.
>>>
>>> 4. Capability to build the bloom filter in parallel in case of
>>> parallel SeqScan. This is done efficiently by populating a local bloom
>>> filter for each parallel worker and then taking a bitwise OR over all
>>> the local bloom filters to form a shared bloom filter at the end of
>>> the parallel SeqScan.
>>>
>>> 5. The optimization is GUC controlled, with settings of
>>> enable_mergejoin_semijoin_filter and force_mergejoin_semijoin_filter.
>>>
>>> We found in experiments that there is a significant improvement
>>> when using the bloom filter during Merge Join. One experiment involved
>>> joining two large tables while varying the theoretical filtering rate
>>> (TFR) between the two tables, the TFR is defined as the percentage
>>> that the two datasets are disjoint. Both tables in the merge join were
>>> the same size. We tested changing the TFR to see the change in
>>> filtering optimization.
>>>
>>> For example, let’s imagine t0 has 10 million rows, which contain the
>>> numbers 1 through 10 million randomly shuffled. Also, t1 has the
>>> numbers 4 million through 14 million randomly shuffled. Then the TFR
>>> for a join of these two tables is 40%, since 40% of the tables are
>>> disjoint from the other table (1 through 4 million for t0, 10 million
>>> through 14 million for t4).
>>>
>>> Here is the performance test result joining two tables:
>>> TFR: theoretical filtering rate
>>> EFR: estimated filtering rate
>>> AFR: actual filtering rate
>>> HJ: hash join
>>> MJ Default: default merge join
>>> MJ Filter: merge join with bloom filter optimization enabled
>>> MJ Filter Forced: merge join with bloom filter optimization forced
>>>
>>> TFR EFR AFR HJ MJ Default MJ Filter MJ Filter Forced
>>>
>>> -------------------------------------------------------------------------------------
>>> 10 33.46 7.41 6529 22638 21949 23160
>>> 20 37.27 14.85 6483 22290 21928 21930
>>> 30 41.32 22.25 6395 22374 20718 20794
>>> 40 45.67 29.7 6272 21969 19449 19410
>>> 50 50.41 37.1 6210 21412 18222 18224
>>> 60 55.64 44.51 6052 21108 17060 17018
>>> 70 61.59 51.98 5947 21020 15682 15737
>>> 80 68.64 59.36 5761 20812 14411 14437
>>> 90 77.83 66.86 5701 20585 13171 13200
>>> Table. Execution Time (ms) vs Filtering Rate (%) for Joining Two
>>> Tables of 10M Rows.
>>>
>>> Attached you can find figures of the same performance test and a SQL
>>> script
>>> to reproduce the performance test.
>>>
>>> The first thing to notice is that Hash Join generally is the most
>>> efficient join strategy. This is because Hash Join is better at
>>> dealing with small tables, and our size of 10 million is still small
>>> enough where Hash Join outperforms the other join strategies. Future
>>> experiments can investigate using much larger tables.
>>>
>>> However, comparing just within the different Merge Join variants, we
>>> see that using the bloom filter greatly improves performance.
>>> Intuitively, all of these execution times follow linear paths.
>>> Comparing forced filtering versus default, we can see that the default
>>> Merge Join outperforms Merge Join with filtering at low filter rates,
>>> but after about 20% TFR, the Merge Join with filtering outperforms
>>> default Merge Join. This makes intuitive sense, as there are some
>>> fixed costs associated with building and checking with the bloom
>>> filter. In the worst case, at only 10% TFR, the bloom filter makes
>>> Merge Join less than 5% slower. However, in the best case, at 90% TFR,
>>> the bloom filter improves Merge Join by 36%.
>>>
>>> Based on the results of the above experiments, we came up with a
>>> linear equation for the performance ratio for using the filter
>>> pushdown from the actual filtering rate. Based on the numbers
>>> presented in the figure, this is the equation:
>>>
>>> T_filter / T_no_filter = 1 / (0.83 * estimated filtering rate + 0.863)
>>>
>>> For example, this means that with an estimated filtering rate of 0.4,
>>> the execution time of merge join is estimated to be improved by 16.3%.
>>> Note that the estimated filtering rate is used in the equation, not
>>> the theoretical filtering rate or the actual filtering rate because it
>>> is what we have during planning. In practice the estimated filtering
>>> rate isn’t usually accurate. In fact, the estimated filtering rate can
>>> differ from the theoretical filtering rate by as much as 17% in our
>>> experiments. One way to mitigate the power loss of bloom filter caused
>>> by inaccurate estimated filtering rate is to adaptively turn it off at
>>> execution time, this is yet to be implemented.
>>>
>>> Here is a list of tasks we plan to work on in order to improve this
>>> patch:
>>> 1. More regression testing to guarantee correctness.
>>> 2. More performance testing involving larger tables and complicated
>>> query plans.
>>> 3. Improve the cost model.
>>> 4. Explore runtime tuning such as making the bloom filter checking
>>> adaptive.
>>> 5. Currently, only the best single join key is used for building the
>>> Bloom filter. However, if there are several keys and we know that
>>> their distributions are somewhat disjoint, we could leverage this fact
>>> and use multiple keys for the bloom filter.
>>> 6. Currently, Bloom filter pushdown is only implemented for SeqScan
>>> nodes. However, it would be possible to allow push down to other types
>>> of scan nodes.
>>> 7. Explore if the Bloom filter could be pushed down through a foreign
>>> scan when the foreign server is capable of handling it – which could
>>> be made true for postgres_fdw.
>>> 8. Better explain command on the usage of bloom filters.
>>>
>>> This patch set is prepared by Marcus Ma, Lyu Pan and myself. Feedback
>>> is appreciated.
>>>
>>> With Regards,
>>> Zheng Li
>>> Amazon RDS/Aurora for PostgreSQL
>>>
>>> [1]
>>> https://www.postgresql.org/message-id/flat/c902844d-837f-5f63-ced3-9f7fd222f175%402ndquadrant.com
>>
>>
>> Hi,
>> In the header of patch 1:
>>
>> In this prototype, the cost model is based on an assumption that there is
>> a linear relationship between the performance gain from using a semijoin
>> filter and the estimated filtering rate:
>> % improvement to Merge Join cost = 0.83 * estimated filtering rate -
>> 0.137.
>>
>> How were the coefficients (0.83 and 0.137) determined ?
>> I guess they were based on the results of running certain workload.
>>
>> Cheers
>>
> Hi,
> For patch 1:
>
> +bool enable_mergejoin_semijoin_filter;
> +bool force_mergejoin_semijoin_filter;
>
> How would (enable_mergejoin_semijoin_filter = off,
> force_mergejoin_semijoin_filter = on) be interpreted ?
> Have you considered using one GUC which has three values: off, enabled,
> forced ?
>
> + mergeclauses_for_sjf = get_actual_clauses(path->path_mergeclauses);
> + mergeclauses_for_sjf =
> get_switched_clauses(path->path_mergeclauses,
> +
> path->jpath.outerjoinpath->parent->relids);
>
> mergeclauses_for_sjf is assigned twice and I don't
> see mergeclauses_for_sjf being reference in the call
> to get_switched_clauses().
> Is this intentional ?
>
> + /* want at least 1000 rows_filtered to avoid any nasty edge
> cases */
> + if (force_mergejoin_semijoin_filter || (filteringRate >= 0.35
> && rows_filtered > 1000))
>
> The above condition is narrower compared to the enclosing condition.
> Since there is no else block for the second if block, please merge the two
> if statements.
>
> + int best_filter_clause;
>
> Normally I would think `clause` is represented by List*. But
> best_filter_clause is an int. Please use another variable name so that
> there is less chance of confusion.
>
> For evaluate_semijoin_filtering_rate():
>
> + double best_sj_selectivity = 1.01;
>
> How was 1.01 determined ?
>
> + debug_sj1("SJPD: start evaluate_semijoin_filtering_rate");
>
> There are debug statements in the methods.
> It would be better to remove them in the next patch set.
>
> Cheers
>
Hi,
Still patch 1.

+ if (!outer_arg_md->is_or_maps_to_base_column
+ && !inner_arg_md->is_or_maps_to_constant)
+ {
+ debug_sj2("SJPD: outer equijoin arg does not map %s",
+ "to a base column nor a constant; semijoin is not
valid");

Looks like there is a typo: inner_arg_md->is_or_maps_to_constant should be
outer_arg_md->is_or_maps_to_constant

+ if (outer_arg_md->est_col_width > MAX_SEMIJOIN_SINGLE_KEY_WIDTH)
+ {
+ debug_sj2("SJPD: outer equijoin column's width %s",
+ "was excessive; condition rejected");

How is the value of MAX_SEMIJOIN_SINGLE_KEY_WIDTH determined ?

For verify_valid_pushdown():

+ Assert(path);
+ Assert(target_var_no > 0);
+
+ if (path == NULL)
+ {
+ return false;

I don't understand the first assertion. Does it mean path would always be
non-NULL ? Then the if statement should be dropped.

+ if (path->parent->relid == target_var_no)
+ {
+ /*
+ * Found source of target var! We know that the pushdown
+ * is valid now.
+ */
+ return true;
+ }
+ return false;

The above can be simplified as: return path->parent->relid == target_var_no;

+ * True if the given con_exprs, ref_exprs and operators will exactlty

Typo: exactlty -> exactly

+ if (!bms_equal(all_vars, matched_vars))
+ return false;
+ return true;

The above can be simplified as: return bms_equal(all_vars, matched_vars);

Cheers