Junrui Li created FLINK-35766:
---------------------------------
Summary: When the job contains many YieldingOperatorFactory
instances, compiling the JobGraph hangs
Key: FLINK-35766
URL: https://issues.apache.org/jira/browse/FLINK-35766
Project: Flink
Issue Type: Bug
Components: Runtime / Coordination
Reporter: Junrui Li
When a job contains YieldingOperatorFactory instances, the time complexity of
compiling the JobGraph is very high (with a complexity of O(N!)). This leads to
the job compilation hanging on creating chains when there are many
YieldingOperatorFactory instances (e.g., more than 30).
This is a very rare bug, but we have users who use SQL that contains many
LookupJoins that use YieldingOperatorFactory in the production environment. A
simple reproducible case is as follows:
{code:java}
@Test
void test() {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.createLocalEnvironment(1);
env.fromSource(
new NumberSequenceSource(0, 10),
WatermarkStrategy.noWatermarks(), "input")
.map((x) -> x)
// add 32 YieldingOperatorFactory
.transform(
"test", BasicTypeInfo.LONG_TYPE_INFO, new
YieldingTestOperatorFactory<>())
.transform(
"test", BasicTypeInfo.LONG_TYPE_INFO, new
YieldingTestOperatorFactory<>())
.transform(
"test", BasicTypeInfo.LONG_TYPE_INFO, new
YieldingTestOperatorFactory<>())
.transform(
"test", BasicTypeInfo.LONG_TYPE_INFO, new
YieldingTestOperatorFactory<>())
.transform(
"test", BasicTypeInfo.LONG_TYPE_INFO, new
YieldingTestOperatorFactory<>())
.transform(
"test", BasicTypeInfo.LONG_TYPE_INFO, new
YieldingTestOperatorFactory<>())
.transform(
"test", BasicTypeInfo.LONG_TYPE_INFO, new
YieldingTestOperatorFactory<>())
.transform(
"test", BasicTypeInfo.LONG_TYPE_INFO, new
YieldingTestOperatorFactory<>())
.transform(
"test", BasicTypeInfo.LONG_TYPE_INFO, new
YieldingTestOperatorFactory<>())
.transform(
"test", BasicTypeInfo.LONG_TYPE_INFO, new
YieldingTestOperatorFactory<>())
.transform(
"test", BasicTypeInfo.LONG_TYPE_INFO, new
YieldingTestOperatorFactory<>())
.transform(
"test", BasicTypeInfo.LONG_TYPE_INFO, new
YieldingTestOperatorFactory<>())
.transform(
"test", BasicTypeInfo.LONG_TYPE_INFO, new
YieldingTestOperatorFactory<>())
.transform(
"test", BasicTypeInfo.LONG_TYPE_INFO, new
YieldingTestOperatorFactory<>())
.transform(
"test", BasicTypeInfo.LONG_TYPE_INFO, new
YieldingTestOperatorFactory<>())
.transform(
"test", BasicTypeInfo.LONG_TYPE_INFO, new
YieldingTestOperatorFactory<>())
.transform(
"test", BasicTypeInfo.LONG_TYPE_INFO, new
YieldingTestOperatorFactory<>())
.transform(
"test", BasicTypeInfo.LONG_TYPE_INFO, new
YieldingTestOperatorFactory<>())
.transform(
"test", BasicTypeInfo.LONG_TYPE_INFO, new
YieldingTestOperatorFactory<>())
.transform(
"test", BasicTypeInfo.LONG_TYPE_INFO, new
YieldingTestOperatorFactory<>())
.transform(
"test", BasicTypeInfo.LONG_TYPE_INFO, new
YieldingTestOperatorFactory<>())
.transform(
"test", BasicTypeInfo.LONG_TYPE_INFO, new
YieldingTestOperatorFactory<>())
.transform(
"test", BasicTypeInfo.LONG_TYPE_INFO, new
YieldingTestOperatorFactory<>())
.transform(
"test", BasicTypeInfo.LONG_TYPE_INFO, new
YieldingTestOperatorFactory<>())
.transform(
"test", BasicTypeInfo.LONG_TYPE_INFO, new
YieldingTestOperatorFactory<>())
.transform(
"test", BasicTypeInfo.LONG_TYPE_INFO, new
YieldingTestOperatorFactory<>())
.transform(
"test", BasicTypeInfo.LONG_TYPE_INFO, new
YieldingTestOperatorFactory<>())
.transform(
"test", BasicTypeInfo.LONG_TYPE_INFO, new
YieldingTestOperatorFactory<>())
.transform(
"test", BasicTypeInfo.LONG_TYPE_INFO, new
YieldingTestOperatorFactory<>())
.transform(
"test", BasicTypeInfo.LONG_TYPE_INFO, new
YieldingTestOperatorFactory<>())
.transform(
"test", BasicTypeInfo.LONG_TYPE_INFO, new
YieldingTestOperatorFactory<>())
.addSink(new DiscardingSink<>());
env.getStreamGraph().getJobGraph();
} {code}
The reason is that there is no caching when determining edge chainable, leading
to repeated backward traversal each time a YiedlingOperatorFactor is
encountered onwards (see code:
[https://github.com/apache/flink/blob/90fc679df073754b93eb5c220373daad7dca0a32/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java#L1602]).
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
