[
https://issues.apache.org/jira/browse/FLINK-41310?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Bonnie Varghese updated FLINK-41310:
------------------------------------
Description:
BatchExecAdaptiveJoin computes the join's managed memory from the compiled
plan's config. It builds the actual join operator at runtime, though, and that
operator reads `table.exec.resource.external-buffer-memory` from the config of
the session that executes the plan.
If the two sessions use different values, the buffer no longer fits the budget.
There are two problems associated with this:
1. A large enough value fails the job with "Memory size is too small".
2. A smaller difference just takes memory from the sorters without any error.
BatchExecHashJoin and BatchExecSortMergeJoin aren't affected, because they read
both values from the plan. This should be fixed after FLINK-41263 is merged.
was:
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.
> 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
> Priority: Major
> Fix For: 2.5.0
>
>
> BatchExecAdaptiveJoin computes the join's managed memory from the compiled
> plan's config. It builds the actual join operator at runtime, though, and
> that operator reads `table.exec.resource.external-buffer-memory` from the
> config of the session that executes the plan.
> If the two sessions use different values, the buffer no longer fits the
> budget. There are two problems associated with this:
> 1. A large enough value fails the job with "Memory size is too small".
> 2. A smaller difference just takes memory from the sorters without any error.
> BatchExecHashJoin and BatchExecSortMergeJoin aren't affected, because they
> read both values from the plan. This should be fixed after FLINK-41263 is
> merged.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
