Bonnie Varghese created FLINK-41310:
---------------------------------------
Summary: BatchExecAdaptiveJoin sizes the external buffer from the
runtime config instead of the compiled plan
Key: FLINK-41310
URL: https://issues.apache.org/jira/browse/FLINK-41310
Project: Flink
Issue Type: Bug
Components: Table SQL / Planner
Affects Versions: 2.5.0
Reporter: Bonnie Varghese
Assignee: Bonnie Varghese
Fix For: 2.5.0
When a batch job runs from a compiled plan, \{{BatchExecAdaptiveJoin}} reads
the external buffer size and the managed memory budget from two different
configs.
* At translation time, \{{translateToPlanInternal}} computes \{{managedMemory}}
via \{{JoinUtil.getManagedMemory}} from the node's \{{ExecNodeConfig}}. For a
compiled plan, that config is the one persisted in the plan. The plan always
persists \{{table.exec.resource.hash-join.memory}},
\{{table.exec.resource.sort.memory}} and
\{{table.exec.resource.external-buffer-memory}}, even when they are set to
their defaults.
* At runtime, \{{PostProcessAdaptiveJoinStrategy}} builds the actual join
operator with \{{context.getStreamGraph().getConfiguration()}}. That is the
config of the session that compiled the job, not the persisted plan config.
\{{HashJoinOperatorUtil}} and \{{SorMergeJoinOperatorUtil}} then compute the
external buffer fraction as:
{code:java}
config.get(TABLE_EXEC_RESOURCE_EXTERNAL_BUFFER_MEMORY).getBytes() /
managedMemory
{code}
If the session that compiles the job sets
\{{table.exec.resource.external-buffer-memory}} to a different value than the
session that compiled the plan, the two values don't match:
* With a higher value, the fraction can exceed 1, and
\{{SortMergeJoinFunction#open}} fails with "Memory size is too small". With the
defaults the budget is max(128 MB, 10 MB + 2 * 128 MB) = 266 MB, so a job
compiled with the buffer set to 300 MB fails.
* With a smaller difference the job doesn't fail, but the buffer quietly takes
memory from the sorters.
{\{BatchExecHashJoin}} and \{{BatchExecSortMergeJoin}} are not affected,
because they build their operators at translation time and read both values
from the same \{{ExecNodeConfig}}. The adaptive join is the only join that
postpones building the operator to runtime. This only became reachable once
\{{BatchExecAdaptiveJoin}} could be serialized into a compiled plan
(FLINK-41263).
*Proposed fix*
Read \{{table.exec.resource.external-buffer-memory}} from the
\{{ExecNodeConfig}} in \{{BatchExecAdaptiveJoin#translateToPlanInternal}}, pass
it into \{{AdaptiveJoinOperatorGenerator}} next to \{{managedMemory}}, and use
it in place of the runtime config value. The budget and the buffer then always
come from the same persisted config. The compiled plan JSON doesn't change,
because the generator is serialized into the operator factory, not the plan.
*Test*
Compile a plan with an adaptive join (converted from a sort merge join), raise
\{{table.exec.resource.external-buffer-memory}} in the session above the
persisted budget, then load and execute the plan. The job should succeed and
produce the expected results. It fails without the fix.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
