I'm Xiaoju Wu.
I'm from eBay.
I'm a big data engineering at eBay now.
Today I'm going to share some of the details
of our work on DPP, which is Dynamic Partition Pruning,
and runtime filter.
Okay, today's agenda is, so firstly,
I will introduce on how,
I think you should already know the
dynamic partition pruning in spark SQL engine optimizer.
And,but the current implementation
in spark community version,
we can not support dynamic partition pruning
in adaptive execution.
So in eBay's spark SQL engine.
We optimize this part to support DPP in AE.
This is the first improvement I want to introduce
and secondly, I'm going to introduce
the Equivalence Inferring that we optimized for DPP.
And the third part is Runtime Filter,
which though semantic is a little bit similar with DPP.
And I want to introduce how implement runtime filter and
some tips on runtime filter for example,
like when should we infer right on filter and how we support
multiple columns, join and how we plan
a runtime filter sub query and how to
keep the capability with DVP and also introduce a little bit
on the exchange or use implementation and at last
the performance results.
So how we support DPP in AE
I think in AE the sub query reuse logic, planning logic
and in the exchange or use logic is totally different
because in AE, we use the sub query map
which is generated in insert adaptive spot plan.
And for, for reuse exchange, we don't use
the reuse exchange rule.
We use the stageCash, which is a logic.
So to support DPP in AE, firstly,
we should extend that Butte saccharin map in
InsertAdaptiveSparkPlan rule.
We need to put the DPP subquery into the subquery map.
And then when we plan the Adaptive Subqueries,
we use that subquery from the subquery map
And also the exchange reuse is happened,
and the stage is replaced by stageCache in AE logic
at runtime.
So we need to, put the DPP exchange reuse logic
into the AE logic.
But one thing to notice is that
the stage we use maybe fails in the runtime, AE logic.
We cannot know just like the pre-calculated rules,
when wilL the exchange be reused?
So if the stage reuse fails, we need to fall back.
So in this way, we import a new Exec class,
which is named the AdaptiveSubquery Excc
This is similar with the adaptive spark plan EXCC,
which is to wrap the plan and
the plan may change around time.
So this class is to wrap the DPP Subquery
and separate on the, this now they will change dynamically
and even fall back to true.
So by this way, to support DPP in AE,
we can benefit from DPP and AE at the same time.
And another improvement on DPP is the equivalence inferring.
We should know that DPP is also a kind of FilterExpression,
and so he should be applied your pipe.
So, so it's okay to apply the DPP
for filter inferring
and push down, the larger should be similar but.
one thing which is different is that
there is equivalence class.
If there's equivalent class for joint condition,
for example, if we have a joint condition, which is
from table A and B, and we also have a DPP on table A,
so we should infer that the, the DPP subquery filtering
can also be applied on table B.
So, but the logic is different from the field,
from the simple expressions filter inferring and push down
because we need to change the pruning key
in the DPP subquery from table A to table B.
So this is the way.
But there are some tricky cases because
for down pole, we have a joint structure like this.
We generate a DPP subquery from A,
and then push down to the sub tree.
But we can only push down to table B
For table A even there is a equivalent
scarce joint condition, we can cannot push down
because this is the same table.
So we have some logic to avoid.
useless DPP inferred and applied.
And Runtime Filter is another optimization that we employ
maintaining an even statistical engine.
Runtime filter semantic is similar with DPP,
but it's not on the partition column,
is on non partition column.
Since its on non partition column so it's high cardinality
and we need to,
so we cannot filter
just the table partitions directly,
so we apply a so important range and
BloomFilter for example, we can create a small table
which can be broadcasted,
and we calculate it's mimex
and the Bloomfilter of the joint columns,
and then we pushed down to the big table as a left
and filter the
big table packet trunks with range and the Bloomfilter.
This is how we do is runtime filter.
But currently we only implement the logic for BroadcastJoin
because it's easier to be implemented.
And, another trickiest things in runtime filter is that,
if there's many joint conditions, how to apply
the runtime filter.
So let's talk about when to infer in the runtime filter.
We have some checks in the limitations to avoid,
the bad performance.
So firstly, if there's already a RuntimeFilter on one side,
we don't need to inferring.
If both sides be broadcasted,
there's no need to infer.
And we also chat the joint type with down pole
Inner/LeftSemi/RightOuter can prune the left
and Inner/LeftOuter can prune the right.
We also have some limitation configurations,
for example, by default only single table can be
runtime filter build side.
If allow multiple tables tables build side cannot have
subqueries to avoid too many nested subqueries.
If we allow the build side to have subqueries,
build side leaves should not have intersection
with other side leaves in case of loop dependency.
If there's an existing single column and equal to filter
on pruned side plan, no need to apply runtime filter
for this column.
If a column data type is not orderable, to apply
runtime range filter for this column.
If the column data type is not supported in BloomFilter,
no need to apply runtime bloom filter for this column.
Still, the next question is how to support
multiple columns.
Because if we apply the runtime filter column by column,
we will have many subqueries and which will make the
performance even worse.
And to support the multiple columns,
we put the main marks and the BloomFilter
into a StructType
because spark supports ordering of StructType
if only the types in a struct can all
be comparable, so we put all the multiple columns
into our genericInternalRow and compare the Struct
as a whole when we do the evaluation.
And one trick is seen in to notice is when we push down the
StrutType, we need to split it into expressions
of single column.
For example, if we have three columns,
we put it into abstract, firstly, and when we push down,
we will generate six filters
for the three columns, which is greater than or equal to,
less than or equal to.
And the next topic is about
how to plan a Runtimefilter Subquery.
We define two more
physical notes which is in
RangeSubqueryExec and
InBloomFilterSubquery Exec
So in the InRangeSubquery node, we calculate a min/max
from broadcast HashRelation, generate the codegen code
for min/max comparing expressions.
And also do the similar scene in InBloomFilterSubquery,
we calculate the Bloomfilter from the broadcast
HashRelation, generate the gen code
for InBoomFilter expression.
We know that we have both DPP
and the runtime field in our engine optimization.
So we need to avoid the conflicts
between the two optimization.
For example, we should not infer runtime filter
with partition keys.
And if we have a DVP on table one from table two,
we should not infer runtime filter from table one
to table two to avoid loop.
We should also have exchange reuse logic
in for runtime filter subquery.
The exchange reuse logic is separated into
with AE and without AE.
Without AE, we just extend the ReuseExchange rule to handle
the InRangeSubquery and InBloomFilterSubquery logic
And with AE, the implementation is similar with AE + DPP.
Firstly, we generate the subqueries map in InsertAdaptive
Sparkplan rule, and then we reuse stage in
AE logic, and then we fall
where we need to fall back the
stage reuse, we fallback the subquery
filter expression to true,
which is implemented in the AdaptiveSubqueryExec.
Here's our performance where it does wave down the ground
time field to performance tests on the TPCDS-5T
that are set, and you can see,
we have big performance increasing some of the queries,
for example, like the Q-72.
Yeah, and that's all, I offer my content.
Thank you for listening and looking forward the
questions and communications.
Thank you