Fabian Hueske created FLINK-41410:
-------------------------------------
Summary: Push nested field accesses below joins to reduce join
shuffle and state size
Key: FLINK-41410
URL: https://issues.apache.org/jira/browse/FLINK-41410
Project: Flink
Issue Type: Improvement
Components: Table SQL / Legacy Planner
Reporter: Fabian Hueske
Assignee: Fabian Hueske
h2. Problem
When a query accesses only a few leaf fields of a ROW column on either side of
a join, the whole ROW is carried into the join. The nested field access is
evaluated in a Calc/Project *above* the join, so the full struct is shuffled
and stored in the join state, even though only a leaf (or a few) is needed. The
same is true for nested field accesses in the join condition.
Existing rules push projections through joins only at the granularity of
top-level columns (e.g. {{{}FlinkProjectJoinTransposeRule{}}}), and Calcite's
{{PushProjector}} records a field access like {{a.b}} as a reference to the
whole column {{{}a{}}}. For wide structs this can dominate the row size, which
directly increases the network volume of the join's Exchange and the size of
its state.
h2. Reproduction
{code:sql}
CREATE TABLE L (id INT, nested ROW<name STRING, `value` INT, other STRING>)
WITH (...);
CREATE TABLE R (id INT, info ROW<city STRING, zip INT, other STRING>)
WITH (...);
SELECT L.nested.name, R.info.city FROM L JOIN R ON L.id = R.id;
{code}
Current plan (abridged): the full structs go through the Exchange and into the
join.
{noformat}
Calc(select=[nested.name AS name, info.city AS city])
+- Join(joinType=[InnerJoin], where=[(id = id0)], select=[id, nested, id0,
info], ...)
:- Exchange(distribution=[hash[id]])
: +- TableSourceScan(table=[[L, project=[id, nested]]])
+- Exchange(distribution=[hash[id]])
+- TableSourceScan(table=[[R, project=[id, info]]])
{noformat}
Expected plan (with a source supporting nested projection pushdown): only the
referenced leaves enter the join.
{noformat}
Calc(select=[nested_name AS name, info_city AS city])
+- Join(joinType=[InnerJoin], where=[(id = id0)], select=[id, nested_name, id0,
info_city], ...)
:- Exchange(distribution=[hash[id]])
: +- TableSourceScan(table=[[L, project=[id, nested_name]]])
+- Exchange(distribution=[hash[id]])
+- TableSourceScan(table=[[R, project=[id, info_city]]])
{noformat}
The benefit does not depend on the source supporting nested projection
push-down: even if the scan still reads the whole struct, the leaf is extracted
in a Calc below the join, so the Exchange and the join state only carry the
leaf.
h2. Proposal
Add a rule that, for a Project above a Join, collects the nested leaf accesses
on each side (including those in the join condition), pushes them below the
join as flat top-level columns, and rewrites the join condition and the top
projection accordingly. Whole-struct references keep the struct whole. Outer
joins widen the pushed leaves to nullable on the null-producing side. Semi/anti
joins are not touched.
The rule is guarded by a new option
{{table.optimizer.nested-projection-join-transpose-enabled}} (default
{{{}false{}}}).
h2. Compatibility
Enabling the option changes the row type entering the join operator, and
therefore the layout of its state. A job planned without the option cannot be
restored from a savepoint after enabling it (and vice versa). This is why the
option defaults to false and should be documented accordingly (release note).
h2. Related
* FLINK-33322: ProjectWatermarkAssignerTransposeRule does not support nested
projection push-down
* FLINK-39804: nested projection not pushed to the source when a filter sits
between project and scan
* FLINK-32940: projection push-down through UNNEST / UDTF
These cover other places where nested pruning is lost; this issue is only about
joins.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
