You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ju...@apache.org on 2025/02/25 03:37:24 UTC
(flink) branch release-2.0 updated (804ce5722e6 -> 44c84620a59)
This is an automated email from the ASF dual-hosted git repository.
junrui pushed a change to branch release-2.0
in repository https://gitbox.apache.org/repos/asf/flink.git
from 804ce5722e6 [FLINK-36352][doc] Remove Scala DataStream API related doc
new 64ed65eabbf [hotfix][runtime] Optimize the calculation logic of AllToAllBlockingResultInfo#getNumBytesProduced
new 90135666265 [hotfix][runtime] Optimize logs in DefaultVertexParallelismAndInputInfosDecider and AdaptiveSkewedJoinOptimizationStrategy
new 7b18d97d42f [hotfix][table-planner] Fix skewed join optimization not working as expected in forced mode
new 44c84620a59 [FLINK-37350][table-planner] Fix memory leak in skewed join optimization strategy
The 4 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../VertexInputInfoComputationUtils.java | 2 +-
.../adaptivebatch/AdaptiveBatchScheduler.java | 21 +++---
.../adaptivebatch/AllToAllBlockingResultInfo.java | 31 +++++++--
...faultVertexParallelismAndInputInfosDecider.java | 6 ++
...VertexParallelismAndInputInfosDeciderUtils.java | 52 +++++++++++++++
.../api/graph/util/ImmutableStreamNode.java | 5 ++
.../AllToAllBlockingResultInfoTest.java | 4 +-
.../exec/processor/AdaptiveJoinProcessor.java | 25 ++++++--
.../plan/batch/sql/adaptive/AdaptiveJoinTest.xml | 2 +-
.../batch/sql/adaptive/AdaptiveJoinITCase.scala | 2 +-
.../sql/adaptive/AdaptiveSkewedJoinITCase.scala | 4 +-
.../AdaptiveBroadcastJoinOptimizationStrategy.java | 11 ++--
.../AdaptiveSkewedJoinOptimizationStrategy.java | 75 ++--------------------
...seAdaptiveJoinOperatorOptimizationStrategy.java | 69 ++++++++++++++++++++
14 files changed, 206 insertions(+), 103 deletions(-)
(flink) 04/04: [FLINK-37350][table-planner] Fix memory leak in skewed join optimization strategy
Posted by ju...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
junrui pushed a commit to branch release-2.0
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 44c84620a592e58674ffe32d94e8a5c2087a3846
Author: noorall <86...@qq.com>
AuthorDate: Wed Feb 19 13:36:51 2025 +0800
[FLINK-37350][table-planner] Fix memory leak in skewed join optimization strategy
---
.../table/runtime/strategy/AdaptiveSkewedJoinOptimizationStrategy.java | 1 +
1 file changed, 1 insertion(+)
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/strategy/AdaptiveSkewedJoinOptimizationStrategy.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/strategy/AdaptiveSkewedJoinOptimizationStrategy.java
index ceea6502e2d..e7569d67dfc 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/strategy/AdaptiveSkewedJoinOptimizationStrategy.java
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/strategy/AdaptiveSkewedJoinOptimizationStrategy.java
@@ -99,6 +99,7 @@ public class AdaptiveSkewedJoinOptimizationStrategy
List<ImmutableStreamEdge> upstreamStreamEdges,
AdaptiveJoin adaptiveJoin) {
if (!canPerformOptimization(context, adaptiveJoinNode)) {
+ freeNodeStatistic(adaptiveJoinNode.getId());
return;
}
for (ImmutableStreamEdge edge : upstreamStreamEdges) {
(flink) 02/04: [hotfix][runtime] Optimize logs in DefaultVertexParallelismAndInputInfosDecider and AdaptiveSkewedJoinOptimizationStrategy
Posted by ju...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
junrui pushed a commit to branch release-2.0
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 901356662651dd6e3bac5c2dc1483697a8cc2906
Author: noorall <86...@qq.com>
AuthorDate: Thu Feb 20 15:21:06 2025 +0800
[hotfix][runtime] Optimize logs in DefaultVertexParallelismAndInputInfosDecider and AdaptiveSkewedJoinOptimizationStrategy
---
.../VertexInputInfoComputationUtils.java | 2 +-
...faultVertexParallelismAndInputInfosDecider.java | 6 +++
...VertexParallelismAndInputInfosDeciderUtils.java | 52 ++++++++++++++++++++++
.../api/graph/util/ImmutableStreamNode.java | 5 +++
.../AdaptiveSkewedJoinOptimizationStrategy.java | 8 ++--
5 files changed, 68 insertions(+), 5 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/VertexInputInfoComputationUtils.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/VertexInputInfoComputationUtils.java
index 4f9075c53ad..52da1b0f73f 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/VertexInputInfoComputationUtils.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/VertexInputInfoComputationUtils.java
@@ -171,7 +171,7 @@ public class VertexInputInfoComputationUtils {
* @param isSingleSubpartitionContainsAllData whether single subpartition contains all data
* @return the computed {@link JobVertexInputInfo}
*/
- static JobVertexInputInfo computeVertexInputInfoForAllToAll(
+ public static JobVertexInputInfo computeVertexInputInfoForAllToAll(
int sourceCount,
int targetCount,
Function<Integer, Integer> numOfSubpartitionsRetriever,
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptivebatch/DefaultVertexParallelismAndInputInfosDecider.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptivebatch/DefaultVertexParallelismAndInputInfosDecider.java
index 38ac16e467d..5f1c683d7cd 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptivebatch/DefaultVertexParallelismAndInputInfosDecider.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptivebatch/DefaultVertexParallelismAndInputInfosDecider.java
@@ -41,6 +41,7 @@ import java.util.Map;
import static org.apache.flink.runtime.scheduler.adaptivebatch.util.VertexParallelismAndInputInfosDeciderUtils.calculateDataVolumePerTaskForInputsGroup;
import static org.apache.flink.runtime.scheduler.adaptivebatch.util.VertexParallelismAndInputInfosDeciderUtils.checkAndGetParallelism;
import static org.apache.flink.runtime.scheduler.adaptivebatch.util.VertexParallelismAndInputInfosDeciderUtils.getNonBroadcastInputInfos;
+import static org.apache.flink.runtime.scheduler.adaptivebatch.util.VertexParallelismAndInputInfosDeciderUtils.logBalancedDataDistributionOptimizationResult;
import static org.apache.flink.util.Preconditions.checkArgument;
import static org.apache.flink.util.Preconditions.checkNotNull;
import static org.apache.flink.util.Preconditions.checkState;
@@ -244,6 +245,11 @@ public class DefaultVertexParallelismAndInputInfosDecider
dataVolumePerTask, pointwiseInputs, consumedResults)));
}
+ for (BlockingInputInfo inputInfo : consumedResults) {
+ logBalancedDataDistributionOptimizationResult(
+ LOG, jobVertexId, inputInfo, vertexInputInfos.get(inputInfo.getResultId()));
+ }
+
return new ParallelismAndInputInfos(
checkAndGetParallelism(vertexInputInfos.values()), vertexInputInfos);
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptivebatch/util/VertexParallelismAndInputInfosDeciderUtils.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptivebatch/util/VertexParallelismAndInputInfosDeciderUtils.java
index 8e3487ad117..1b48a9f2aed 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptivebatch/util/VertexParallelismAndInputInfosDeciderUtils.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptivebatch/util/VertexParallelismAndInputInfosDeciderUtils.java
@@ -21,6 +21,7 @@ package org.apache.flink.runtime.scheduler.adaptivebatch.util;
import org.apache.flink.runtime.executiongraph.ExecutionVertexInputInfo;
import org.apache.flink.runtime.executiongraph.IndexRange;
import org.apache.flink.runtime.executiongraph.JobVertexInputInfo;
+import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.scheduler.adaptivebatch.BisectionSearchUtils;
import org.apache.flink.runtime.scheduler.adaptivebatch.BlockingInputInfo;
@@ -43,6 +44,8 @@ import java.util.stream.IntStream;
import java.util.stream.LongStream;
import static org.apache.flink.runtime.executiongraph.IndexRangeUtil.mergeIndexRanges;
+import static org.apache.flink.runtime.executiongraph.VertexInputInfoComputationUtils.computeVertexInputInfoForAllToAll;
+import static org.apache.flink.runtime.executiongraph.VertexInputInfoComputationUtils.computeVertexInputInfoForPointwise;
import static org.apache.flink.util.Preconditions.checkArgument;
import static org.apache.flink.util.Preconditions.checkState;
@@ -658,4 +661,53 @@ public class VertexParallelismAndInputInfosDeciderUtils {
long globalDataVolumePerTask, long inputsGroupBytes, long totalDataBytes) {
return (long) ((double) inputsGroupBytes / totalDataBytes * globalDataVolumePerTask);
}
+
+ /**
+ * Logs the data distribution optimization info when a balanced data distribution algorithm is
+ * effectively optimized compared to the num-based data distribution algorithm.
+ *
+ * @param logger The logger instance used for logging output.
+ * @param jobVertexId The id for the job vertex.
+ * @param inputInfo The original input info
+ * @param optimizedJobVertexInputInfo The optimized job vertex input info.
+ */
+ public static void logBalancedDataDistributionOptimizationResult(
+ Logger logger,
+ JobVertexID jobVertexId,
+ BlockingInputInfo inputInfo,
+ JobVertexInputInfo optimizedJobVertexInputInfo) {
+ List<ExecutionVertexInputInfo> optimizedExecutionVertexInputInfos =
+ optimizedJobVertexInputInfo.getExecutionVertexInputInfos();
+ int parallelism = optimizedExecutionVertexInputInfos.size();
+ List<ExecutionVertexInputInfo> nonOptimizedExecutionVertexInputInfos =
+ computeNumBasedJobVertexInputInfo(parallelism, inputInfo)
+ .getExecutionVertexInputInfos();
+ // When the execution vertex input infos of the two are inconsistent, we consider that
+ // data balanced distribution optimization has performed.
+ if (!optimizedExecutionVertexInputInfos.equals(nonOptimizedExecutionVertexInputInfos)) {
+ logger.info(
+ "Optimized the balanced data distribution for vertex {}, which reads from result {} with type number {}",
+ jobVertexId,
+ inputInfo.getResultId(),
+ inputInfo.getInputTypeNumber());
+ }
+ }
+
+ private static JobVertexInputInfo computeNumBasedJobVertexInputInfo(
+ int parallelism, BlockingInputInfo inputInfo) {
+ int sourceParallelism = inputInfo.getNumPartitions();
+
+ if (inputInfo.isPointwise()) {
+ return computeVertexInputInfoForPointwise(
+ sourceParallelism, parallelism, inputInfo::getNumSubpartitions, true);
+ } else {
+ return computeVertexInputInfoForAllToAll(
+ sourceParallelism,
+ parallelism,
+ inputInfo::getNumSubpartitions,
+ true,
+ inputInfo.isBroadcast(),
+ inputInfo.isSingleSubpartitionContainsAllData());
+ }
+ }
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/api/graph/util/ImmutableStreamNode.java b/flink-runtime/src/main/java/org/apache/flink/streaming/api/graph/util/ImmutableStreamNode.java
index e5dae3fbc1a..39706bad70f 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/api/graph/util/ImmutableStreamNode.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/api/graph/util/ImmutableStreamNode.java
@@ -84,4 +84,9 @@ public class ImmutableStreamNode {
.filter(Objects::nonNull)
.toArray(TypeSerializer<?>[]::new);
}
+
+ @Override
+ public String toString() {
+ return streamNode.toString();
+ }
}
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/strategy/AdaptiveSkewedJoinOptimizationStrategy.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/strategy/AdaptiveSkewedJoinOptimizationStrategy.java
index d8373a98e44..3d9dfbbbcb8 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/strategy/AdaptiveSkewedJoinOptimizationStrategy.java
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/strategy/AdaptiveSkewedJoinOptimizationStrategy.java
@@ -205,7 +205,7 @@ public class AdaptiveSkewedJoinOptimizationStrategy
checkState(
leftInputSize != null,
"Left input bytes of adaptive join [%s] is unknown, which is unexpected.",
- adaptiveJoinNode.getId());
+ adaptiveJoinNode.toString());
long[] rightInputSize =
aggregatedProducedBytesByTypeNumberAndNodeId
.get(adaptiveJoinNode.getId())
@@ -213,7 +213,7 @@ public class AdaptiveSkewedJoinOptimizationStrategy
checkState(
rightInputSize != null,
"Right input bytes of adaptive join [%s] is unknown, which is unexpected.",
- adaptiveJoinNode.getId());
+ adaptiveJoinNode.toString());
long leftSkewedThreshold =
computeSkewThreshold(median(leftInputSize), skewedFactor, skewedThresholdInBytes);
@@ -254,7 +254,7 @@ public class AdaptiveSkewedJoinOptimizationStrategy
LOG.info(
"Apply skewed join optimization {} for left input of node {}.",
isModificationSucceed ? "succeeded" : "failed",
- adaptiveJoinNode.getId());
+ adaptiveJoinNode);
}
if (isRightOptimizable) {
boolean isModificationSucceed =
@@ -263,7 +263,7 @@ public class AdaptiveSkewedJoinOptimizationStrategy
LOG.info(
"Apply skewed join optimization {} for right input of node {}.",
isModificationSucceed ? "succeeded" : "failed",
- adaptiveJoinNode.getId());
+ adaptiveJoinNode);
}
}
(flink) 01/04: [hotfix][runtime] Optimize the calculation logic of AllToAllBlockingResultInfo#getNumBytesProduced
Posted by ju...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
junrui pushed a commit to branch release-2.0
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 64ed65eabbfede9439ab09f351b90fce52ac67ab
Author: noorall <86...@qq.com>
AuthorDate: Fri Feb 21 18:22:50 2025 +0800
[hotfix][runtime] Optimize the calculation logic of AllToAllBlockingResultInfo#getNumBytesProduced
---
.../adaptivebatch/AdaptiveBatchScheduler.java | 21 +++++++--------
.../adaptivebatch/AllToAllBlockingResultInfo.java | 31 +++++++++++++++++-----
.../AllToAllBlockingResultInfoTest.java | 4 +--
3 files changed, 37 insertions(+), 19 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptivebatch/AdaptiveBatchScheduler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptivebatch/AdaptiveBatchScheduler.java
index 231ee420a3e..91762e062b6 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptivebatch/AdaptiveBatchScheduler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptivebatch/AdaptiveBatchScheduler.java
@@ -505,17 +505,16 @@ public class AdaptiveBatchScheduler extends DefaultScheduler implements JobGraph
}
/**
- * Aggregates subpartition bytes if all conditions are met. This method checks whether the
- * result info instance is of type {@link AllToAllBlockingResultInfo}, whether all consumer
- * vertices are created, whether all consumer vertices are initialized, and whether input bytes
- * information for all consumer vertices has been initialized . If these conditions are
- * satisfied, the fine-grained statistic info will not be required by consumer vertices, and
- * then we could aggregate the subpartition bytes.
+ * Notifies that fine-grained subpartition bytes may not be needed for the specified result
+ * info. This method checks if the given {@link BlockingResultInfo} instance is of type {@link
+ * AllToAllBlockingResultInfo}, if all consumer vertices are created and initialized, and if the
+ * input bytes information for all consumer vertices is initialized. If all these conditions are
+ * met, the fine-grained statistics for subpartition bytes will not be needed.
*
- * @param resultInfo the BlockingResultInfo instance to potentially aggregate subpartition bytes
- * for.
+ * @param resultInfo the {@link BlockingResultInfo} instance, which holds the fine-grained
+ * subpartition bytes.
*/
- private void maybeAggregateSubpartitionBytes(BlockingResultInfo resultInfo) {
+ private void notifyFineGrainedSubpartitionBytesMayNotNeeded(BlockingResultInfo resultInfo) {
IntermediateResult intermediateResult =
getExecutionGraph().getAllIntermediateResults().get(resultInfo.getResultId());
@@ -534,7 +533,7 @@ public class AdaptiveBatchScheduler extends DefaultScheduler implements JobGraph
taskVertex ->
taskVertex.getInputBytes()
!= NUM_BYTES_UNKNOWN))) {
- ((AllToAllBlockingResultInfo) resultInfo).aggregateSubpartitionBytes();
+ ((AllToAllBlockingResultInfo) resultInfo).onFineGrainedSubpartitionBytesNotNeeded();
}
}
@@ -824,7 +823,7 @@ public class AdaptiveBatchScheduler extends DefaultScheduler implements JobGraph
blockingResultInfo.getNumBytesProduced(
partitionIndexRange, subpartitionIndexRange);
}
- maybeAggregateSubpartitionBytes(blockingResultInfo);
+ notifyFineGrainedSubpartitionBytesMayNotNeeded(blockingResultInfo);
}
ev.setInputBytes(inputBytes);
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptivebatch/AllToAllBlockingResultInfo.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptivebatch/AllToAllBlockingResultInfo.java
index e8923bb4847..4d54646cdb0 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptivebatch/AllToAllBlockingResultInfo.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptivebatch/AllToAllBlockingResultInfo.java
@@ -19,6 +19,7 @@
package org.apache.flink.runtime.scheduler.adaptivebatch;
import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.runtime.executiongraph.IndexRange;
import org.apache.flink.runtime.executiongraph.ResultPartitionBytes;
import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
@@ -31,6 +32,7 @@ import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.stream.Collectors;
+import java.util.stream.IntStream;
import static org.apache.flink.util.Preconditions.checkState;
@@ -128,7 +130,8 @@ public class AllToAllBlockingResultInfo extends AbstractBlockingResultInfo {
}
/**
- * Aggregates the subpartition bytes to reduce space usage.
+ * This method should be called when fine-grained information is no longer needed. It will
+ * aggregate and clears the fine-grained subpartition bytes to reduce space usage.
*
* <p>Once all partitions are finished and all consumer jobVertices are initialized, we can
* convert the subpartition bytes to aggregated value to reduce the space usage, because the
@@ -136,9 +139,11 @@ public class AllToAllBlockingResultInfo extends AbstractBlockingResultInfo {
* tasks of ALL_TO_ALL edges(Hashing or Rebalancing, we do not consider rare cases such as
* custom partitions here).
*/
- protected void aggregateSubpartitionBytes() {
+ protected void onFineGrainedSubpartitionBytesNotNeeded() {
if (subpartitionBytesByPartitionIndex.size() == numOfPartitions) {
- this.aggregatedSubpartitionBytes = getAggregatedSubpartitionBytesInternal();
+ if (this.aggregatedSubpartitionBytes == null) {
+ this.aggregatedSubpartitionBytes = getAggregatedSubpartitionBytesInternal();
+ }
this.subpartitionBytesByPartitionIndex.clear();
}
}
@@ -179,15 +184,29 @@ public class AllToAllBlockingResultInfo extends AbstractBlockingResultInfo {
}
}
+ @Override
+ public long getNumBytesProduced(
+ IndexRange partitionIndexRange, IndexRange subpartitionIndexRange) {
+ if (partitionIndexRange.getStartIndex() == 0
+ && partitionIndexRange.getEndIndex() == getNumPartitions() - 1) {
+ return IntStream.rangeClosed(
+ subpartitionIndexRange.getStartIndex(),
+ subpartitionIndexRange.getEndIndex())
+ .mapToLong(i -> getAggregatedSubpartitionBytes().get(i))
+ .sum();
+ } else {
+ return super.getNumBytesProduced(partitionIndexRange, subpartitionIndexRange);
+ }
+ }
+
public List<Long> getAggregatedSubpartitionBytes() {
checkState(
aggregatedSubpartitionBytes != null
|| subpartitionBytesByPartitionIndex.size() == numOfPartitions,
"Not all partition infos are ready");
if (aggregatedSubpartitionBytes == null) {
- return getAggregatedSubpartitionBytesInternal();
- } else {
- return Collections.unmodifiableList(aggregatedSubpartitionBytes);
+ aggregatedSubpartitionBytes = getAggregatedSubpartitionBytesInternal();
}
+ return Collections.unmodifiableList(aggregatedSubpartitionBytes);
}
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptivebatch/AllToAllBlockingResultInfoTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptivebatch/AllToAllBlockingResultInfoTest.java
index 6930940fd7c..a97350c3337 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptivebatch/AllToAllBlockingResultInfoTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptivebatch/AllToAllBlockingResultInfoTest.java
@@ -70,7 +70,7 @@ class AllToAllBlockingResultInfoTest {
AllToAllBlockingResultInfo resultInfo =
new AllToAllBlockingResultInfo(new IntermediateDataSetID(), 2, 2, false, false);
resultInfo.recordPartitionInfo(0, new ResultPartitionBytes(new long[] {32L, 64L}));
- resultInfo.aggregateSubpartitionBytes();
+ resultInfo.onFineGrainedSubpartitionBytesNotNeeded();
assertThatThrownBy(resultInfo::getNumBytesProduced)
.isInstanceOf(IllegalStateException.class);
@@ -103,7 +103,7 @@ class AllToAllBlockingResultInfoTest {
assertThat(resultInfo.getAggregatedSubpartitionBytes()).containsExactly(192L, 384L);
// The raw info should not be clear
assertThat(resultInfo.getNumOfRecordedPartitions()).isGreaterThan(0);
- resultInfo.aggregateSubpartitionBytes();
+ resultInfo.onFineGrainedSubpartitionBytesNotNeeded();
// The raw info should be clear
assertThat(resultInfo.getNumOfRecordedPartitions()).isZero();
(flink) 03/04: [hotfix][table-planner] Fix skewed join optimization not working as expected in forced mode
Posted by ju...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
junrui pushed a commit to branch release-2.0
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 7b18d97d42f9fdd6d223f9695f59d1df38225d6b
Author: noorall <86...@qq.com>
AuthorDate: Fri Feb 21 15:10:28 2025 +0800
[hotfix][table-planner] Fix skewed join optimization not working as expected in forced mode
---
.../exec/processor/AdaptiveJoinProcessor.java | 25 ++++++--
.../plan/batch/sql/adaptive/AdaptiveJoinTest.xml | 2 +-
.../batch/sql/adaptive/AdaptiveJoinITCase.scala | 2 +-
.../sql/adaptive/AdaptiveSkewedJoinITCase.scala | 4 +-
.../AdaptiveBroadcastJoinOptimizationStrategy.java | 11 ++--
.../AdaptiveSkewedJoinOptimizationStrategy.java | 66 ---------------------
...seAdaptiveJoinOperatorOptimizationStrategy.java | 69 ++++++++++++++++++++++
7 files changed, 100 insertions(+), 79 deletions(-)
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/processor/AdaptiveJoinProcessor.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/processor/AdaptiveJoinProcessor.java
index 7187eabf589..d5a67b19073 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/processor/AdaptiveJoinProcessor.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/processor/AdaptiveJoinProcessor.java
@@ -61,12 +61,14 @@ public class AdaptiveJoinProcessor implements ExecNodeGraphProcessor {
@Override
protected void visitNode(ExecNode<?> node) {
visitInputs(node);
- // AdaptiveJoin conversion should be avoided when there is a
- // KEEP_INPUT_AS_IS constraint downstream. And we don't need to check all
- // downstream nodes of the join, because the KEEP_INPUT_AS_IS constraint
- // will be bound to BatchExecExchange, which will be the direct downstream
- // node of the join.
- if (shouldKeepInputAsIs(node.getInputProperties())) {
+ // To avoid data correctness issues, we will refrain from transforming join
+ // nodes with a strict KEEP_INPUT_AS_IS constraint downstream. Meanwhile,
+ // deeper evaluations will be conducted in the corresponding
+ // StreamGraphOptimizationStrategy.
+ // And we don't need to check all downstream nodes of the join, because the
+ // KEEP_INPUT_AS_IS constraint will be bound to BatchExecExchange, which
+ // will be the direct downstream node of the join.
+ if (shouldStrictKeepInputAsIs(node.getInputProperties())) {
return;
}
for (int i = 0; i < node.getInputEdges().size(); ++i) {
@@ -116,6 +118,17 @@ public class AdaptiveJoinProcessor implements ExecNodeGraphProcessor {
return newNode;
}
+ private boolean shouldStrictKeepInputAsIs(List<InputProperty> inputProperties) {
+ return inputProperties.stream()
+ .anyMatch(
+ inputProperty ->
+ (inputProperty.getRequiredDistribution().getType()
+ == KEEP_INPUT_AS_IS)
+ && ((InputProperty.KeepInputAsIsDistribution)
+ inputProperty.getRequiredDistribution())
+ .isStrict());
+ }
+
private boolean shouldKeepInputAsIs(List<InputProperty> inputProperties) {
return inputProperties.stream()
.anyMatch(
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/adaptive/AdaptiveJoinTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/adaptive/AdaptiveJoinTest.xml
index 4931121cbc8..105c8e7dafe 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/adaptive/AdaptiveJoinTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/adaptive/AdaptiveJoinTest.xml
@@ -49,7 +49,7 @@ Calc(select=[EXPR$0])
:- Exchange(distribution=[keep_input_as_is[hash[a1]]])
: +- Calc(select=[a1, b1])
: +- Exchange(distribution=[keep_input_as_is[hash[a1]]])
- : +- HashJoin(joinType=[InnerJoin], where=[(a1 = a2)], select=[a1, b1, a2], build=[right])
+ : +- AdaptiveJoin(originalJoin=[ShuffleHashJoin], joinType=[InnerJoin], where=[(a1 = a2)], select=[a1, b1, a2], build=[right])
: :- Exchange(distribution=[hash[a1]])
: : +- Calc(select=[a1, b1])
: : +- TableSourceScan(table=[[default_catalog, default_database, T1]], fields=[a1, b1, c1, d1])
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/adaptive/AdaptiveJoinITCase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/adaptive/AdaptiveJoinITCase.scala
index 5b0c3fc6661..61441957d25 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/adaptive/AdaptiveJoinITCase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/adaptive/AdaptiveJoinITCase.scala
@@ -71,7 +71,7 @@ trait AdaptiveJoinITCase extends AdaptiveBatchTestBase {
"""
|WITH
| r AS (SELECT * FROM T1, T2, T3 WHERE a1 = a2 and a1 = a3)
- |SELECT sum(b1) FROM r group by a1
+ |SELECT sum(b1), count(c1) FROM r group by a1
|""".stripMargin
checkResult(sql)
}
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/adaptive/AdaptiveSkewedJoinITCase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/adaptive/AdaptiveSkewedJoinITCase.scala
index 6ff56242e26..c59d6596380 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/adaptive/AdaptiveSkewedJoinITCase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/adaptive/AdaptiveSkewedJoinITCase.scala
@@ -17,7 +17,7 @@
*/
package org.apache.flink.table.planner.runtime.batch.sql.adaptive
-import org.apache.flink.configuration.{BatchExecutionOptions, MemorySize}
+import org.apache.flink.configuration.{BatchExecutionOptions, MemorySize, PipelineOptions}
import org.apache.flink.table.api.config.OptimizerConfigOptions
import org.junit.jupiter.api.{BeforeEach, Test}
@@ -110,6 +110,8 @@ class AdaptiveSkewedJoinITCase extends AdaptiveJoinITCase {
OptimizerConfigOptions.TABLE_OPTIMIZER_ADAPTIVE_SKEWED_JOIN_OPTIMIZATION_STRATEGY,
OptimizerConfigOptions.AdaptiveSkewedJoinOptimizationStrategy.FORCED
)
+ tEnv.getConfig.set(PipelineOptions.OPERATOR_CHAINING, Boolean.box(false))
checkResult(sql, expected)
+ tEnv.getConfig.set(PipelineOptions.OPERATOR_CHAINING, Boolean.box(true))
}
}
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/strategy/AdaptiveBroadcastJoinOptimizationStrategy.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/strategy/AdaptiveBroadcastJoinOptimizationStrategy.java
index 09b020d0e7a..9b806f8acf7 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/strategy/AdaptiveBroadcastJoinOptimizationStrategy.java
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/strategy/AdaptiveBroadcastJoinOptimizationStrategy.java
@@ -79,8 +79,7 @@ public class AdaptiveBroadcastJoinOptimizationStrategy
ImmutableStreamNode adaptiveJoinNode,
List<ImmutableStreamEdge> upstreamStreamEdges,
AdaptiveJoin adaptiveJoin) {
- if (!canPerformOptimization(
- adaptiveJoinNode, context.getStreamGraph().getConfiguration())) {
+ if (!canPerformOptimization(adaptiveJoinNode, context)) {
return;
}
for (ImmutableStreamEdge upstreamEdge : upstreamStreamEdges) {
@@ -187,8 +186,12 @@ public class AdaptiveBroadcastJoinOptimizationStrategy
}
private boolean canPerformOptimization(
- ImmutableStreamNode adaptiveJoinNode, ReadableConfig config) {
- return !isBroadcastJoinDisabled(config) && !isBroadcastJoin(adaptiveJoinNode);
+ ImmutableStreamNode adaptiveJoinNode, StreamGraphContext context) {
+ if (isBroadcastJoinDisabled(context.getStreamGraph().getConfiguration())
+ || isBroadcastJoin(adaptiveJoinNode)) {
+ return false;
+ }
+ return canPerformOptimizationAutomatic(context, adaptiveJoinNode);
}
private void aggregatedInputBytesByTypeNumber(
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/strategy/AdaptiveSkewedJoinOptimizationStrategy.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/strategy/AdaptiveSkewedJoinOptimizationStrategy.java
index 3d9dfbbbcb8..ceea6502e2d 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/strategy/AdaptiveSkewedJoinOptimizationStrategy.java
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/strategy/AdaptiveSkewedJoinOptimizationStrategy.java
@@ -27,19 +27,11 @@ import org.apache.flink.streaming.api.graph.StreamGraphContext;
import org.apache.flink.streaming.api.graph.util.ImmutableStreamEdge;
import org.apache.flink.streaming.api.graph.util.ImmutableStreamNode;
import org.apache.flink.streaming.api.graph.util.StreamEdgeUpdateRequestInfo;
-import org.apache.flink.streaming.runtime.partitioner.BroadcastPartitioner;
-import org.apache.flink.streaming.runtime.partitioner.CustomPartitionerWrapper;
import org.apache.flink.streaming.runtime.partitioner.ForwardForConsecutiveHashPartitioner;
-import org.apache.flink.streaming.runtime.partitioner.GlobalPartitioner;
-import org.apache.flink.streaming.runtime.partitioner.KeyGroupStreamPartitioner;
-import org.apache.flink.streaming.runtime.partitioner.RebalancePartitioner;
-import org.apache.flink.streaming.runtime.partitioner.ShufflePartitioner;
import org.apache.flink.streaming.runtime.partitioner.StreamPartitioner;
import org.apache.flink.table.api.config.OptimizerConfigOptions;
import org.apache.flink.table.runtime.operators.join.FlinkJoinType;
import org.apache.flink.table.runtime.operators.join.adaptive.AdaptiveJoin;
-import org.apache.flink.table.runtime.partitioner.BinaryHashPartitioner;
-import org.apache.flink.table.runtime.partitioner.RowDataCustomStreamPartitioner;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -48,7 +40,6 @@ import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.Set;
import static org.apache.flink.runtime.scheduler.adaptivebatch.util.VertexParallelismAndInputInfosDeciderUtils.computeSkewThreshold;
import static org.apache.flink.runtime.scheduler.adaptivebatch.util.VertexParallelismAndInputInfosDeciderUtils.median;
@@ -66,25 +57,6 @@ public class AdaptiveSkewedJoinOptimizationStrategy
private static final int LEFT_INPUT_TYPE_NUMBER = 1;
private static final int RIGHT_INPUT_TYPE_NUMBER = 2;
- /** Set of partitioners that can automatically correct key group. */
- private static final Set<Class<?>> partitionersCanCorrectKeyGroupAutomatic =
- Set.of(
- BinaryHashPartitioner.class,
- BroadcastPartitioner.class,
- CustomPartitionerWrapper.class,
- GlobalPartitioner.class,
- KeyGroupStreamPartitioner.class,
- RebalancePartitioner.class,
- RowDataCustomStreamPartitioner.class,
- ShufflePartitioner.class);
-
- /**
- * Set of partitioners that can force key group correction but may introduce additional shuffle
- * overhead.
- */
- private static final Set<Class<?>> partitionersCanCorrectKeyGroupForced =
- Set.of(ForwardForConsecutiveHashPartitioner.class);
-
private Map<Integer, Map<Integer, long[]>> aggregatedProducedBytesByTypeNumberAndNodeId;
private OptimizerConfigOptions.AdaptiveSkewedJoinOptimizationStrategy
@@ -328,42 +300,4 @@ public class AdaptiveSkewedJoinOptimizationStrategy
}
return false;
}
-
- private static boolean canPerformOptimizationAutomatic(
- StreamGraphContext context, ImmutableStreamNode adaptiveJoinNode) {
- return adaptiveJoinNode.getOutEdges().stream()
- .allMatch(
- edge -> {
- Class<?> classOfOutputPartitioner =
- checkNotNull(
- context.getOutputPartitioner(
- edge.getEdgeId(),
- edge.getSourceId(),
- edge.getTargetId()))
- .getClass();
- return !edge.isIntraInputKeyCorrelated()
- || partitionersCanCorrectKeyGroupAutomatic.contains(
- classOfOutputPartitioner);
- });
- }
-
- private static boolean canPerformOptimizationForced(
- StreamGraphContext context, ImmutableStreamNode adaptiveJoinNode) {
- return adaptiveJoinNode.getOutEdges().stream()
- .allMatch(
- edge -> {
- Class<?> classOfOutputPartitioner =
- checkNotNull(
- context.getOutputPartitioner(
- edge.getEdgeId(),
- edge.getSourceId(),
- edge.getTargetId()))
- .getClass();
- return !edge.isIntraInputKeyCorrelated()
- || partitionersCanCorrectKeyGroupAutomatic.contains(
- classOfOutputPartitioner)
- || partitionersCanCorrectKeyGroupForced.contains(
- classOfOutputPartitioner);
- });
- }
}
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/strategy/BaseAdaptiveJoinOperatorOptimizationStrategy.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/strategy/BaseAdaptiveJoinOperatorOptimizationStrategy.java
index 847ce5935dd..bec4b1d9e65 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/strategy/BaseAdaptiveJoinOperatorOptimizationStrategy.java
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/strategy/BaseAdaptiveJoinOperatorOptimizationStrategy.java
@@ -24,17 +24,48 @@ import org.apache.flink.streaming.api.graph.StreamGraphContext;
import org.apache.flink.streaming.api.graph.util.ImmutableStreamEdge;
import org.apache.flink.streaming.api.graph.util.ImmutableStreamGraph;
import org.apache.flink.streaming.api.graph.util.ImmutableStreamNode;
+import org.apache.flink.streaming.runtime.partitioner.BroadcastPartitioner;
+import org.apache.flink.streaming.runtime.partitioner.CustomPartitionerWrapper;
+import org.apache.flink.streaming.runtime.partitioner.ForwardForConsecutiveHashPartitioner;
+import org.apache.flink.streaming.runtime.partitioner.GlobalPartitioner;
+import org.apache.flink.streaming.runtime.partitioner.KeyGroupStreamPartitioner;
+import org.apache.flink.streaming.runtime.partitioner.RebalancePartitioner;
+import org.apache.flink.streaming.runtime.partitioner.ShufflePartitioner;
import org.apache.flink.table.runtime.operators.join.adaptive.AdaptiveJoin;
+import org.apache.flink.table.runtime.partitioner.BinaryHashPartitioner;
+import org.apache.flink.table.runtime.partitioner.RowDataCustomStreamPartitioner;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Set;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
/** The base stream graph optimization strategy class for adaptive join operator. */
public abstract class BaseAdaptiveJoinOperatorOptimizationStrategy
implements StreamGraphOptimizationStrategy {
+ /** Set of partitioners that can automatically correct key group. */
+ private static final Set<Class<?>> PARTITIONERS_CAN_CORRECT_KEY_GROUP_AUTOMATIC =
+ Set.of(
+ BinaryHashPartitioner.class,
+ BroadcastPartitioner.class,
+ CustomPartitionerWrapper.class,
+ GlobalPartitioner.class,
+ KeyGroupStreamPartitioner.class,
+ RebalancePartitioner.class,
+ RowDataCustomStreamPartitioner.class,
+ ShufflePartitioner.class);
+
+ /**
+ * Set of partitioners that can force key group correction but may introduce additional shuffle
+ * overhead.
+ */
+ private static final Set<Class<?>> PARTITIONERS_CAN_CORRECT_KEY_GROUP_FORCED =
+ Set.of(ForwardForConsecutiveHashPartitioner.class);
+
protected void visitDownstreamAdaptiveJoinNode(
OperatorsFinished operatorsFinished, StreamGraphContext context) {
ImmutableStreamGraph streamGraph = context.getStreamGraph();
@@ -68,4 +99,42 @@ public abstract class BaseAdaptiveJoinOperatorOptimizationStrategy
ImmutableStreamNode adaptiveJoinNode,
List<ImmutableStreamEdge> upstreamStreamEdges,
AdaptiveJoin adaptiveJoin);
+
+ static boolean canPerformOptimizationAutomatic(
+ StreamGraphContext context, ImmutableStreamNode adaptiveJoinNode) {
+ return adaptiveJoinNode.getOutEdges().stream()
+ .allMatch(
+ edge -> {
+ Class<?> classOfOutputPartitioner =
+ checkNotNull(
+ context.getOutputPartitioner(
+ edge.getEdgeId(),
+ edge.getSourceId(),
+ edge.getTargetId()))
+ .getClass();
+ return !edge.isIntraInputKeyCorrelated()
+ || PARTITIONERS_CAN_CORRECT_KEY_GROUP_AUTOMATIC.contains(
+ classOfOutputPartitioner);
+ });
+ }
+
+ static boolean canPerformOptimizationForced(
+ StreamGraphContext context, ImmutableStreamNode adaptiveJoinNode) {
+ return adaptiveJoinNode.getOutEdges().stream()
+ .allMatch(
+ edge -> {
+ Class<?> classOfOutputPartitioner =
+ checkNotNull(
+ context.getOutputPartitioner(
+ edge.getEdgeId(),
+ edge.getSourceId(),
+ edge.getTargetId()))
+ .getClass();
+ return !edge.isIntraInputKeyCorrelated()
+ || PARTITIONERS_CAN_CORRECT_KEY_GROUP_AUTOMATIC.contains(
+ classOfOutputPartitioner)
+ || PARTITIONERS_CAN_CORRECT_KEY_GROUP_FORCED.contains(
+ classOfOutputPartitioner);
+ });
+ }
}