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/21 02:24:52 UTC
(flink) branch release-2.0 updated (94c0ae9a4e0 -> b9734cf768a)
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 94c0ae9a4e0 [hotfix] Fix post-release japicmp update script for macOS
new 83dc6f40138 [FLINK-37333][runtime] Use ForwardForUnspecifiedPartitioner when adaptive broadcast join takes effect.
new b9734cf768a [hotfix][runtime] Add log for adaptive broadcast join strategy.
The 2 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:
.../adaptive/AdaptiveJoinOperatorGenerator.java | 14 +++++++++++++-
.../AdaptiveBroadcastJoinOptimizationStrategy.java | 20 ++++++++++++++++++--
.../strategy/PostProcessAdaptiveJoinStrategy.java | 10 ++++++++++
3 files changed, 41 insertions(+), 3 deletions(-)
(flink) 01/02: [FLINK-37333][runtime] Use ForwardForUnspecifiedPartitioner when adaptive broadcast join takes effect.
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 83dc6f4013854ccc5089daf7d9bc0472504e9bde
Author: sunxia <xi...@gmail.com>
AuthorDate: Mon Feb 17 15:11:18 2025 +0800
[FLINK-37333][runtime] Use ForwardForUnspecifiedPartitioner when adaptive broadcast join takes effect.
---
.../runtime/strategy/AdaptiveBroadcastJoinOptimizationStrategy.java | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
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 0f90a71762f..f073ec13426 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
@@ -27,7 +27,7 @@ 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.ForwardPartitioner;
+import org.apache.flink.streaming.runtime.partitioner.ForwardForUnspecifiedPartitioner;
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;
@@ -208,7 +208,7 @@ public class AdaptiveBroadcastJoinOptimizationStrategy
filterEdges(inEdges, leftIsBuild ? 1 : 2), new BroadcastPartitioner<>());
List<StreamEdgeUpdateRequestInfo> modifiedProbeSideEdges =
generateStreamEdgeUpdateRequestInfos(
- filterEdges(inEdges, leftIsBuild ? 2 : 1), new ForwardPartitioner<>());
+ filterEdges(inEdges, leftIsBuild ? 2 : 1), new ForwardForUnspecifiedPartitioner<>());
modifiedBuildSideEdges.addAll(modifiedProbeSideEdges);
return context.modifyStreamEdge(modifiedBuildSideEdges);
(flink) 02/02: [hotfix][runtime] Add log for adaptive broadcast join 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 b9734cf768a33053919026a2151bbb518eafece3
Author: sunxia <xi...@gmail.com>
AuthorDate: Mon Feb 17 18:08:22 2025 +0800
[hotfix][runtime] Add log for adaptive broadcast join strategy.
---
.../adaptive/AdaptiveJoinOperatorGenerator.java | 14 +++++++++++++-
.../AdaptiveBroadcastJoinOptimizationStrategy.java | 18 +++++++++++++++++-
.../strategy/PostProcessAdaptiveJoinStrategy.java | 10 ++++++++++
3 files changed, 40 insertions(+), 2 deletions(-)
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/adaptive/AdaptiveJoinOperatorGenerator.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/adaptive/AdaptiveJoinOperatorGenerator.java
index 6dd8b79278d..78981a5e8ce 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/adaptive/AdaptiveJoinOperatorGenerator.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/adaptive/AdaptiveJoinOperatorGenerator.java
@@ -27,6 +27,9 @@ import org.apache.flink.table.runtime.operators.join.FlinkJoinType;
import org.apache.flink.table.runtime.operators.join.adaptive.AdaptiveJoin;
import org.apache.flink.table.types.logical.RowType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import static org.apache.flink.util.Preconditions.checkState;
/**
@@ -34,7 +37,7 @@ import static org.apache.flink.util.Preconditions.checkState;
* shuffle hash join or shuffle merge join operator based on actual conditions.
*/
public class AdaptiveJoinOperatorGenerator implements AdaptiveJoin {
-
+ private static final Logger LOG = LoggerFactory.getLogger(AdaptiveJoinOperatorGenerator.class);
private final int[] leftKeys;
private final int[] rightKeys;
@@ -65,6 +68,8 @@ public class AdaptiveJoinOperatorGenerator implements AdaptiveJoin {
private boolean leftIsBuild;
+ private boolean originalLeftIsBuild;
+
private boolean isBroadcastJoin;
public AdaptiveJoinOperatorGenerator(
@@ -105,6 +110,7 @@ public class AdaptiveJoinOperatorGenerator implements AdaptiveJoin {
+ "SortMergeJoin, not including %s.",
originalJoin.toString()));
this.leftIsBuild = leftIsBuild;
+ this.originalLeftIsBuild = leftIsBuild;
this.originalJoin = originalJoin;
}
@@ -164,6 +170,12 @@ public class AdaptiveJoinOperatorGenerator implements AdaptiveJoin {
return false;
}
+ if (leftIsBuild != originalLeftIsBuild) {
+ LOG.info(
+ "The build side of the adaptive join has been updated. Compile phase build side: {}, Runtime build side: {}.",
+ originalLeftIsBuild ? "left" : "right",
+ leftIsBuild ? "left" : "right");
+ }
return !leftIsBuild;
}
}
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 f073ec13426..09b020d0e7a 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
@@ -162,7 +162,22 @@ public class AdaptiveBroadcastJoinOptimizationStrategy
broadcastThreshold,
leftInputSize,
rightInputSize);
+ } else {
+ LOG.info(
+ "Modification to stream edges for the join node [{}] failed. Keep the join node as is.",
+ adaptiveJoinNode.getId());
}
+ } else {
+ LOG.debug(
+ "The size of the specified side of the input data for the join node [{}] "
+ + "is too large to be converted into a broadcast hash join. "
+ + "The Join type: {}, Broadcast threshold: {} bytes, Left input size: "
+ + "{} bytes, Right input size: {} bytes.",
+ adaptiveJoinNode.getId(),
+ joinType,
+ broadcastThreshold,
+ leftInputSize,
+ rightInputSize);
}
adaptiveJoin.markAsBroadcastJoin(
isBroadcast, isBroadcast ? leftIsBuild : leftSmallerThanRight);
@@ -208,7 +223,8 @@ public class AdaptiveBroadcastJoinOptimizationStrategy
filterEdges(inEdges, leftIsBuild ? 1 : 2), new BroadcastPartitioner<>());
List<StreamEdgeUpdateRequestInfo> modifiedProbeSideEdges =
generateStreamEdgeUpdateRequestInfos(
- filterEdges(inEdges, leftIsBuild ? 2 : 1), new ForwardForUnspecifiedPartitioner<>());
+ filterEdges(inEdges, leftIsBuild ? 2 : 1),
+ new ForwardForUnspecifiedPartitioner<>());
modifiedBuildSideEdges.addAll(modifiedProbeSideEdges);
return context.modifyStreamEdge(modifiedBuildSideEdges);
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/strategy/PostProcessAdaptiveJoinStrategy.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/strategy/PostProcessAdaptiveJoinStrategy.java
index 6e898bc9718..02f45cb9c1c 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/strategy/PostProcessAdaptiveJoinStrategy.java
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/strategy/PostProcessAdaptiveJoinStrategy.java
@@ -29,6 +29,9 @@ import org.apache.flink.streaming.api.graph.util.StreamNodeUpdateRequestInfo;
import org.apache.flink.table.runtime.operators.join.adaptive.AdaptiveJoin;
import org.apache.flink.util.Preconditions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import java.util.ArrayList;
import java.util.List;
@@ -40,6 +43,9 @@ import java.util.List;
*/
public class PostProcessAdaptiveJoinStrategy extends BaseAdaptiveJoinOperatorOptimizationStrategy {
+ private static final Logger LOG =
+ LoggerFactory.getLogger(PostProcessAdaptiveJoinStrategy.class);
+
@Override
public boolean onOperatorsFinished(
OperatorsFinished operatorsFinished, StreamGraphContext context) {
@@ -68,6 +74,10 @@ public class PostProcessAdaptiveJoinStrategy extends BaseAdaptiveJoinOperatorOpt
"Unexpected error occurs while reordering the inputs "
+ "of the adaptive join node, potentially leading to data inaccuracies. "
+ "Exceptions will be thrown.");
+ } else {
+ LOG.info(
+ "Reordered the inputs of the adaptive join node {}.",
+ adaptiveJoinNode.getId());
}
}