You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by zh...@apache.org on 2024/03/22 13:17:44 UTC
(flink) branch release-1.19 updated (6eeae5fe6c9 -> 5ec4bf2f181)
This is an automated email from the ASF dual-hosted git repository.
zhuzh pushed a change to branch release-1.19
in repository https://gitbox.apache.org/repos/asf/flink.git
from 6eeae5fe6c9 Update japicmp configuration for 1.19.0
new 94c32618ac7 [hotfix][tests] Migrate TransformationTest to Junit5
new 5ec4bf2f181 [FLINK-32513][core] Add predecessor caching
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:
.../org/apache/flink/api/dag/Transformation.java | 27 +++-
.../apache/flink/api/dag/TransformationTest.java | 77 +++++-----
.../AbstractBroadcastStateTransformation.java | 12 +-
.../AbstractMultipleInputTransformation.java | 12 +-
.../api/transformations/CacheTransformation.java | 2 +-
.../transformations/CoFeedbackTransformation.java | 2 +-
.../transformations/FeedbackTransformation.java | 2 +-
.../transformations/LegacySinkTransformation.java | 2 +-
.../LegacySourceTransformation.java | 2 +-
.../transformations/OneInputTransformation.java | 2 +-
.../transformations/PartitionTransformation.java | 2 +-
.../api/transformations/ReduceTransformation.java | 2 +-
.../transformations/SideOutputTransformation.java | 2 +-
.../api/transformations/SinkTransformation.java | 2 +-
.../api/transformations/SourceTransformation.java | 2 +-
.../SourceTransformationWrapper.java | 2 +-
.../TimestampsAndWatermarksTransformation.java | 2 +-
.../transformations/TwoInputTransformation.java | 18 ++-
.../api/transformations/UnionTransformation.java | 16 +-
.../api/graph/StreamGraphGeneratorTest.java | 2 +-
.../GetTransitivePredecessorsTest.java | 162 +++++++++++++++++++++
.../TableOperatorWrapperGeneratorTest.java | 2 +-
22 files changed, 280 insertions(+), 74 deletions(-)
create mode 100644 flink-streaming-java/src/test/java/org/apache/flink/streaming/api/transformations/GetTransitivePredecessorsTest.java
(flink) 02/02: [FLINK-32513][core] Add predecessor caching
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zhuzh pushed a commit to branch release-1.19
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 5ec4bf2f18168001b5cbb9012f331d3405228516
Author: Jeyhun Karimov <je...@gmail.com>
AuthorDate: Sun Mar 10 15:56:02 2024 +0100
[FLINK-32513][core] Add predecessor caching
This closes #24475.
---
.../org/apache/flink/api/dag/Transformation.java | 27 +++-
.../apache/flink/api/dag/TransformationTest.java | 2 +-
.../AbstractBroadcastStateTransformation.java | 12 +-
.../AbstractMultipleInputTransformation.java | 12 +-
.../api/transformations/CacheTransformation.java | 2 +-
.../transformations/CoFeedbackTransformation.java | 2 +-
.../transformations/FeedbackTransformation.java | 2 +-
.../transformations/LegacySinkTransformation.java | 2 +-
.../LegacySourceTransformation.java | 2 +-
.../transformations/OneInputTransformation.java | 2 +-
.../transformations/PartitionTransformation.java | 2 +-
.../api/transformations/ReduceTransformation.java | 2 +-
.../transformations/SideOutputTransformation.java | 2 +-
.../api/transformations/SinkTransformation.java | 2 +-
.../api/transformations/SourceTransformation.java | 2 +-
.../SourceTransformationWrapper.java | 2 +-
.../TimestampsAndWatermarksTransformation.java | 2 +-
.../transformations/TwoInputTransformation.java | 18 ++-
.../api/transformations/UnionTransformation.java | 16 +-
.../api/graph/StreamGraphGeneratorTest.java | 2 +-
.../GetTransitivePredecessorsTest.java | 162 +++++++++++++++++++++
.../TableOperatorWrapperGeneratorTest.java | 2 +-
22 files changed, 237 insertions(+), 42 deletions(-)
diff --git a/flink-core/src/main/java/org/apache/flink/api/dag/Transformation.java b/flink-core/src/main/java/org/apache/flink/api/dag/Transformation.java
index 6256f9624f6..f2033a1fd2b 100644
--- a/flink-core/src/main/java/org/apache/flink/api/dag/Transformation.java
+++ b/flink-core/src/main/java/org/apache/flink/api/dag/Transformation.java
@@ -162,6 +162,13 @@ public abstract class Transformation<T> {
private final Map<ManagedMemoryUseCase, Integer> managedMemoryOperatorScopeUseCaseWeights =
new HashMap<>();
+ /**
+ * This map is a cache that stores transitive predecessors and used in {@code
+ * getTransitivePredecessors()}.
+ */
+ private final Map<Transformation<T>, List<Transformation<?>>> predecessorsCache =
+ new HashMap<>();
+
/** Slot scope use cases that this transformation needs managed memory for. */
private final Set<ManagedMemoryUseCase> managedMemorySlotScopeUseCases = new HashSet<>();
@@ -231,6 +238,12 @@ public abstract class Transformation<T> {
return name;
}
+ /** Returns the predecessorsCache of this {@code Transformation}. */
+ @VisibleForTesting
+ Map<Transformation<T>, List<Transformation<?>>> getPredecessorsCache() {
+ return predecessorsCache;
+ }
+
/** Changes the description of this {@code Transformation}. */
public void setDescription(String description) {
this.description = Preconditions.checkNotNull(description);
@@ -579,7 +592,19 @@ public abstract class Transformation<T> {
*
* @return The list of transitive predecessors.
*/
- public abstract List<Transformation<?>> getTransitivePredecessors();
+ protected abstract List<Transformation<?>> getTransitivePredecessorsInternal();
+
+ /**
+ * Returns all transitive predecessor {@code Transformation}s of this {@code Transformation}.
+ * This is, for example, used when determining whether a feedback edge of an iteration actually
+ * has the iteration head as a predecessor. This method is just a wrapper on top of {@code
+ * getTransitivePredecessorsInternal} method with public access. It uses caching internally.
+ *
+ * @return The list of transitive predecessors.
+ */
+ public final List<Transformation<?>> getTransitivePredecessors() {
+ return predecessorsCache.computeIfAbsent(this, key -> getTransitivePredecessorsInternal());
+ }
/**
* Returns the {@link Transformation transformations} that are the immediate predecessors of the
diff --git a/flink-core/src/test/java/org/apache/flink/api/dag/TransformationTest.java b/flink-core/src/test/java/org/apache/flink/api/dag/TransformationTest.java
index ced452cd14f..c200fac33fc 100644
--- a/flink-core/src/test/java/org/apache/flink/api/dag/TransformationTest.java
+++ b/flink-core/src/test/java/org/apache/flink/api/dag/TransformationTest.java
@@ -142,7 +142,7 @@ class TransformationTest {
}
@Override
- public List<Transformation<?>> getTransitivePredecessors() {
+ protected List<Transformation<?>> getTransitivePredecessorsInternal() {
return Collections.emptyList();
}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/AbstractBroadcastStateTransformation.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/AbstractBroadcastStateTransformation.java
index c7869e61389..1ce6c7c9bfd 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/AbstractBroadcastStateTransformation.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/AbstractBroadcastStateTransformation.java
@@ -26,6 +26,8 @@ import org.apache.flink.streaming.api.operators.ChainingStrategy;
import java.util.ArrayList;
import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -103,12 +105,10 @@ public class AbstractBroadcastStateTransformation<IN1, IN2, OUT>
}
@Override
- public List<Transformation<?>> getTransitivePredecessors() {
- final List<Transformation<?>> predecessors = new ArrayList<>();
- predecessors.add(this);
- predecessors.add(regularInput);
- predecessors.add(broadcastInput);
- return predecessors;
+ protected List<Transformation<?>> getTransitivePredecessorsInternal() {
+ return Stream.of(this, regularInput, broadcastInput)
+ .distinct()
+ .collect(Collectors.toList());
}
@Override
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/AbstractMultipleInputTransformation.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/AbstractMultipleInputTransformation.java
index 65bbc53188f..a733cebbaaf 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/AbstractMultipleInputTransformation.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/AbstractMultipleInputTransformation.java
@@ -76,10 +76,14 @@ public abstract class AbstractMultipleInputTransformation<OUT> extends PhysicalT
}
@Override
- public List<Transformation<?>> getTransitivePredecessors() {
- return inputs.stream()
- .flatMap(input -> input.getTransitivePredecessors().stream())
- .collect(Collectors.toList());
+ protected List<Transformation<?>> getTransitivePredecessorsInternal() {
+ List<Transformation<?>> predecessors =
+ getInputs().stream()
+ .flatMap(input -> input.getTransitivePredecessors().stream())
+ .distinct()
+ .collect(Collectors.toList());
+ predecessors.add(this);
+ return predecessors;
}
@Override
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/CacheTransformation.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/CacheTransformation.java
index 4632c9b020d..e8bd22adcc8 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/CacheTransformation.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/CacheTransformation.java
@@ -57,7 +57,7 @@ public class CacheTransformation<T> extends Transformation<T> {
}
@Override
- public List<Transformation<?>> getTransitivePredecessors() {
+ protected List<Transformation<?>> getTransitivePredecessorsInternal() {
List<Transformation<?>> result = Lists.newArrayList();
result.add(this);
if (isCached) {
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/CoFeedbackTransformation.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/CoFeedbackTransformation.java
index 11f8a912608..4d366ba2a8f 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/CoFeedbackTransformation.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/CoFeedbackTransformation.java
@@ -106,7 +106,7 @@ public class CoFeedbackTransformation<F> extends Transformation<F> {
}
@Override
- public List<Transformation<?>> getTransitivePredecessors() {
+ protected List<Transformation<?>> getTransitivePredecessorsInternal() {
return Collections.singletonList(this);
}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/FeedbackTransformation.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/FeedbackTransformation.java
index ab8f7424741..ef4ec2766bf 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/FeedbackTransformation.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/FeedbackTransformation.java
@@ -102,7 +102,7 @@ public class FeedbackTransformation<T> extends Transformation<T> {
}
@Override
- public List<Transformation<?>> getTransitivePredecessors() {
+ protected List<Transformation<?>> getTransitivePredecessorsInternal() {
List<Transformation<?>> result = Lists.newArrayList();
result.add(this);
result.addAll(input.getTransitivePredecessors());
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/LegacySinkTransformation.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/LegacySinkTransformation.java
index a0bd6417881..fcd753c13b4 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/LegacySinkTransformation.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/LegacySinkTransformation.java
@@ -136,7 +136,7 @@ public class LegacySinkTransformation<T> extends PhysicalTransformation<T> {
}
@Override
- public List<Transformation<?>> getTransitivePredecessors() {
+ protected List<Transformation<?>> getTransitivePredecessorsInternal() {
List<Transformation<?>> result = Lists.newArrayList();
result.add(this);
result.addAll(input.getTransitivePredecessors());
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/LegacySourceTransformation.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/LegacySourceTransformation.java
index 4718ff18386..9443a3a14ea 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/LegacySourceTransformation.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/LegacySourceTransformation.java
@@ -92,7 +92,7 @@ public class LegacySourceTransformation<T> extends PhysicalTransformation<T>
}
@Override
- public List<Transformation<?>> getTransitivePredecessors() {
+ protected List<Transformation<?>> getTransitivePredecessorsInternal() {
return Collections.singletonList(this);
}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/OneInputTransformation.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/OneInputTransformation.java
index 0a3957130fc..7222fad97da 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/OneInputTransformation.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/OneInputTransformation.java
@@ -168,7 +168,7 @@ public class OneInputTransformation<IN, OUT> extends PhysicalTransformation<OUT>
}
@Override
- public List<Transformation<?>> getTransitivePredecessors() {
+ protected List<Transformation<?>> getTransitivePredecessorsInternal() {
List<Transformation<?>> result = Lists.newArrayList();
result.add(this);
result.addAll(input.getTransitivePredecessors());
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/PartitionTransformation.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/PartitionTransformation.java
index 92d3f264731..966e565c78f 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/PartitionTransformation.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/PartitionTransformation.java
@@ -89,7 +89,7 @@ public class PartitionTransformation<T> extends Transformation<T> {
}
@Override
- public List<Transformation<?>> getTransitivePredecessors() {
+ protected List<Transformation<?>> getTransitivePredecessorsInternal() {
List<Transformation<?>> result = Lists.newArrayList();
result.add(this);
result.addAll(input.getTransitivePredecessors());
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/ReduceTransformation.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/ReduceTransformation.java
index da159f3af11..82c4e1c4694 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/ReduceTransformation.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/ReduceTransformation.java
@@ -89,7 +89,7 @@ public final class ReduceTransformation<IN, K> extends PhysicalTransformation<IN
}
@Override
- public List<Transformation<?>> getTransitivePredecessors() {
+ protected List<Transformation<?>> getTransitivePredecessorsInternal() {
List<Transformation<?>> result = Lists.newArrayList();
result.add(this);
result.addAll(input.getTransitivePredecessors());
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/SideOutputTransformation.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/SideOutputTransformation.java
index 3177bfa0eb1..e054d37be54 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/SideOutputTransformation.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/SideOutputTransformation.java
@@ -53,7 +53,7 @@ public class SideOutputTransformation<T> extends Transformation<T> {
}
@Override
- public List<Transformation<?>> getTransitivePredecessors() {
+ protected List<Transformation<?>> getTransitivePredecessorsInternal() {
List<Transformation<?>> result = Lists.newArrayList();
result.add(this);
result.addAll(input.getTransitivePredecessors());
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/SinkTransformation.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/SinkTransformation.java
index bdaa8d2dede..a4b4310588e 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/SinkTransformation.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/SinkTransformation.java
@@ -73,7 +73,7 @@ public class SinkTransformation<InputT, OutputT> extends PhysicalTransformation<
}
@Override
- public List<Transformation<?>> getTransitivePredecessors() {
+ protected List<Transformation<?>> getTransitivePredecessorsInternal() {
final List<Transformation<?>> result = Lists.newArrayList();
result.add(this);
result.addAll(input.getTransitivePredecessors());
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/SourceTransformation.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/SourceTransformation.java
index 86c07ca5511..0260da43b82 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/SourceTransformation.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/SourceTransformation.java
@@ -92,7 +92,7 @@ public class SourceTransformation<OUT, SplitT extends SourceSplit, EnumChkT>
}
@Override
- public List<Transformation<?>> getTransitivePredecessors() {
+ protected List<Transformation<?>> getTransitivePredecessorsInternal() {
return Collections.singletonList(this);
}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/SourceTransformationWrapper.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/SourceTransformationWrapper.java
index d536000fde2..6aed1a3da76 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/SourceTransformationWrapper.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/SourceTransformationWrapper.java
@@ -58,7 +58,7 @@ public class SourceTransformationWrapper<T> extends Transformation<T> {
}
@Override
- public List<Transformation<?>> getTransitivePredecessors() {
+ protected List<Transformation<?>> getTransitivePredecessorsInternal() {
List<Transformation<?>> result = Lists.newArrayList();
result.add(this);
result.addAll(input.getTransitivePredecessors());
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/TimestampsAndWatermarksTransformation.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/TimestampsAndWatermarksTransformation.java
index c8bde911a04..26541f74673 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/TimestampsAndWatermarksTransformation.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/TimestampsAndWatermarksTransformation.java
@@ -77,7 +77,7 @@ public class TimestampsAndWatermarksTransformation<IN> extends PhysicalTransform
}
@Override
- public List<Transformation<?>> getTransitivePredecessors() {
+ protected List<Transformation<?>> getTransitivePredecessorsInternal() {
List<Transformation<?>> transformations = Lists.newArrayList();
transformations.add(this);
transformations.addAll(input.getTransitivePredecessors());
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/TwoInputTransformation.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/TwoInputTransformation.java
index 4c576170741..48005bf95c4 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/TwoInputTransformation.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/TwoInputTransformation.java
@@ -28,10 +28,10 @@ import org.apache.flink.streaming.api.operators.SimpleOperatorFactory;
import org.apache.flink.streaming.api.operators.StreamOperatorFactory;
import org.apache.flink.streaming.api.operators.TwoInputStreamOperator;
-import org.apache.flink.shaded.guava31.com.google.common.collect.Lists;
-
import java.util.ArrayList;
import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
/**
* This Transformation represents the application of a {@link TwoInputStreamOperator} to two input
@@ -217,12 +217,14 @@ public class TwoInputTransformation<IN1, IN2, OUT> extends PhysicalTransformatio
}
@Override
- public List<Transformation<?>> getTransitivePredecessors() {
- List<Transformation<?>> result = Lists.newArrayList();
- result.add(this);
- result.addAll(input1.getTransitivePredecessors());
- result.addAll(input2.getTransitivePredecessors());
- return result;
+ protected List<Transformation<?>> getTransitivePredecessorsInternal() {
+ List<Transformation<?>> predecessors =
+ Stream.of(input1, input2)
+ .flatMap(input -> input.getTransitivePredecessors().stream())
+ .distinct()
+ .collect(Collectors.toList());
+ predecessors.add(this);
+ return predecessors;
}
@Override
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/UnionTransformation.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/UnionTransformation.java
index dd29206a9ac..59ce6adceb7 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/UnionTransformation.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/UnionTransformation.java
@@ -25,6 +25,7 @@ import org.apache.flink.shaded.guava31.com.google.common.collect.Lists;
import java.util.ArrayList;
import java.util.List;
+import java.util.stream.Collectors;
/**
* This transformation represents a union of several input {@link Transformation Transformations}.
@@ -63,12 +64,13 @@ public class UnionTransformation<T> extends Transformation<T> {
}
@Override
- public List<Transformation<?>> getTransitivePredecessors() {
- List<Transformation<?>> result = Lists.newArrayList();
- result.add(this);
- for (Transformation<T> input : inputs) {
- result.addAll(input.getTransitivePredecessors());
- }
- return result;
+ protected List<Transformation<?>> getTransitivePredecessorsInternal() {
+ List<Transformation<?>> predecessors =
+ inputs.stream()
+ .flatMap(input -> input.getTransitivePredecessors().stream())
+ .distinct()
+ .collect(Collectors.toList());
+ predecessors.add(this);
+ return predecessors;
}
}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamGraphGeneratorTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamGraphGeneratorTest.java
index db5ac6f7851..2cf832b2ef1 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamGraphGeneratorTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamGraphGeneratorTest.java
@@ -1051,7 +1051,7 @@ public class StreamGraphGeneratorTest extends TestLogger {
}
@Override
- public List<Transformation<?>> getTransitivePredecessors() {
+ protected List<Transformation<?>> getTransitivePredecessorsInternal() {
return Collections.emptyList();
}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/transformations/GetTransitivePredecessorsTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/transformations/GetTransitivePredecessorsTest.java
new file mode 100644
index 00000000000..c658e7bb3fa
--- /dev/null
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/transformations/GetTransitivePredecessorsTest.java
@@ -0,0 +1,162 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.api.transformations;
+
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.dag.Transformation;
+import org.apache.flink.api.java.typeutils.GenericTypeInfo;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.api.operators.SimpleOperatorFactory;
+import org.apache.flink.streaming.api.operators.TwoInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@code getTransitivePredecessors} method of {@link Transformation}. */
+class GetTransitivePredecessorsTest {
+
+ private TestTransformation<Integer> commonNode;
+ private Transformation<Integer> midNode;
+
+ @BeforeEach
+ void setup() {
+ commonNode = new TestTransformation<>("commonNode", new MockIntegerTypeInfo(), 1);
+ midNode =
+ new OneInputTransformation<>(
+ commonNode,
+ "midNode",
+ new DummyOneInputOperator(),
+ new MockIntegerTypeInfo(),
+ 1);
+ }
+
+ @Test
+ void testTwoInputTransformation() {
+ Transformation<Integer> topNode =
+ new TwoInputTransformation<>(
+ commonNode,
+ midNode,
+ "topNode",
+ new DummyTwoInputOperator<>(),
+ midNode.getOutputType(),
+ 1);
+ List<Transformation<?>> predecessors = topNode.getTransitivePredecessors();
+ assertThat(predecessors.size()).isEqualTo(3);
+ assertThat(commonNode.getNumGetTransitivePredecessor()).isEqualTo(1);
+ }
+
+ @Test
+ void testUnionTransformation() {
+ Transformation<Integer> topNode =
+ new UnionTransformation<>(Arrays.asList(commonNode, midNode));
+ List<Transformation<?>> predecessors = topNode.getTransitivePredecessors();
+ assertThat(predecessors.size()).isEqualTo(3);
+ assertThat(commonNode.getNumGetTransitivePredecessor()).isEqualTo(1);
+ }
+
+ @Test
+ void testBroadcastStateTransformation() {
+ Transformation<Integer> topNode =
+ new AbstractBroadcastStateTransformation<>(
+ "topNode", commonNode, midNode, null, midNode.getOutputType(), 1);
+ List<Transformation<?>> predecessors = topNode.getTransitivePredecessors();
+ assertThat(predecessors.size()).isEqualTo(3);
+ assertThat(commonNode.getNumGetTransitivePredecessor()).isEqualTo(0);
+ }
+
+ @Test
+ void testAbstractMultipleInputTransformation() {
+ Transformation<Integer> topNode =
+ new AbstractMultipleInputTransformation<Integer>(
+ "topNode",
+ SimpleOperatorFactory.of(new DummyTwoInputOperator<>()),
+ midNode.getOutputType(),
+ 1) {
+ @Override
+ public List<Transformation<?>> getInputs() {
+ return Arrays.asList(commonNode, midNode);
+ }
+ };
+ List<Transformation<?>> predecessors = topNode.getTransitivePredecessors();
+ assertThat(predecessors.size()).isEqualTo(3);
+ assertThat(commonNode.getNumGetTransitivePredecessor()).isEqualTo(1);
+ }
+
+ /** A test implementation of {@link Transformation}. */
+ private static class TestTransformation<T> extends Transformation<T> {
+ private int numGetTransitivePredecessor = 0;
+
+ public TestTransformation(String name, TypeInformation<T> outputType, int parallelism) {
+ super(name, outputType, parallelism);
+ }
+
+ @Override
+ protected List<Transformation<?>> getTransitivePredecessorsInternal() {
+ ++numGetTransitivePredecessor;
+ return Collections.singletonList(this);
+ }
+
+ @Override
+ public List<Transformation<?>> getInputs() {
+ return Collections.emptyList();
+ }
+
+ public int getNumGetTransitivePredecessor() {
+ return numGetTransitivePredecessor;
+ }
+ }
+
+ /** A test implementation of {@link OneInputTransformation}. */
+ private static class DummyOneInputOperator extends AbstractStreamOperator<Integer>
+ implements OneInputStreamOperator<Integer, Integer> {
+
+ @Override
+ public void processElement(StreamRecord<Integer> element) throws Exception {}
+ }
+
+ /** A test implementation of {@link TwoInputStreamOperator}. */
+ private static class DummyTwoInputOperator<T> extends AbstractStreamOperator<T>
+ implements TwoInputStreamOperator<T, T, T> {
+
+ @Override
+ public void processElement1(StreamRecord<T> element) throws Exception {
+ output.collect(element);
+ }
+
+ @Override
+ public void processElement2(StreamRecord<T> element) throws Exception {
+ output.collect(element);
+ }
+ }
+
+ /** A test implementation of {@link TypeInformation}. */
+ private static class MockIntegerTypeInfo extends GenericTypeInfo<Integer> {
+ public MockIntegerTypeInfo() {
+ super(Integer.class);
+ }
+ }
+}
diff --git a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/multipleinput/TableOperatorWrapperGeneratorTest.java b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/multipleinput/TableOperatorWrapperGeneratorTest.java
index 9d4041bbf2a..6ee22d8f9a4 100644
--- a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/multipleinput/TableOperatorWrapperGeneratorTest.java
+++ b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/multipleinput/TableOperatorWrapperGeneratorTest.java
@@ -484,7 +484,7 @@ public class TableOperatorWrapperGeneratorTest extends MultipleInputTestBase {
}
@Override
- public List<Transformation<?>> getTransitivePredecessors() {
+ protected List<Transformation<?>> getTransitivePredecessorsInternal() {
return Collections.emptyList();
}
(flink) 01/02: [hotfix][tests] Migrate TransformationTest to Junit5
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zhuzh pushed a commit to branch release-1.19
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 94c32618ac7211fe19a53b4124dc26288fdf944a
Author: Jeyhun Karimov <je...@gmail.com>
AuthorDate: Sat Mar 16 10:04:00 2024 +0100
[hotfix][tests] Migrate TransformationTest to Junit5
---
.../apache/flink/api/dag/TransformationTest.java | 75 +++++++++++++---------
1 file changed, 43 insertions(+), 32 deletions(-)
diff --git a/flink-core/src/test/java/org/apache/flink/api/dag/TransformationTest.java b/flink-core/src/test/java/org/apache/flink/api/dag/TransformationTest.java
index c14ab0d8e3d..ced452cd14f 100644
--- a/flink-core/src/test/java/org/apache/flink/api/dag/TransformationTest.java
+++ b/flink-core/src/test/java/org/apache/flink/api/dag/TransformationTest.java
@@ -22,10 +22,9 @@ import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.core.memory.ManagedMemoryUseCase;
import org.apache.flink.core.testutils.CheckedThread;
import org.apache.flink.core.testutils.OneShotLatch;
-import org.apache.flink.util.TestLogger;
-import org.junit.Before;
-import org.junit.Test;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
import java.util.ArrayList;
import java.util.Collections;
@@ -33,23 +32,21 @@ import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;
-import static org.hamcrest.Matchers.contains;
-import static org.hamcrest.Matchers.is;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertThat;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for {@link Transformation}. */
-public class TransformationTest extends TestLogger {
+class TransformationTest {
private Transformation<Void> transformation;
- @Before
+ @BeforeEach
public void setUp() {
transformation = new TestTransformation<>("t", null, 1);
}
@Test
- public void testGetNewNodeIdIsThreadSafe() throws Exception {
+ void testGetNewNodeIdIsThreadSafe() throws Exception {
final int numThreads = 10;
final int numIdsPerThread = 100;
@@ -84,43 +81,57 @@ public class TransformationTest extends TestLogger {
final Set<Integer> deduplicatedIds =
idLists.stream().flatMap(List::stream).collect(Collectors.toSet());
- assertEquals(numThreads * numIdsPerThread, deduplicatedIds.size());
+ assertThat(numThreads * numIdsPerThread).isEqualTo(deduplicatedIds.size());
}
@Test
- public void testDeclareManagedMemoryUseCase() {
+ void testDeclareManagedMemoryUseCase() {
transformation.declareManagedMemoryUseCaseAtOperatorScope(
ManagedMemoryUseCase.OPERATOR, 123);
transformation.declareManagedMemoryUseCaseAtSlotScope(ManagedMemoryUseCase.STATE_BACKEND);
assertThat(
- transformation
- .getManagedMemoryOperatorScopeUseCaseWeights()
- .get(ManagedMemoryUseCase.OPERATOR),
- is(123));
- assertThat(
- transformation.getManagedMemorySlotScopeUseCases(),
- contains(ManagedMemoryUseCase.STATE_BACKEND));
+ transformation
+ .getManagedMemoryOperatorScopeUseCaseWeights()
+ .get(ManagedMemoryUseCase.OPERATOR))
+ .isEqualTo(123);
+ assertThat(transformation.getManagedMemorySlotScopeUseCases())
+ .contains(ManagedMemoryUseCase.STATE_BACKEND);
}
- @Test(expected = IllegalArgumentException.class)
- public void testDeclareManagedMemoryOperatorScopeUseCaseFailWrongScope() {
- transformation.declareManagedMemoryUseCaseAtOperatorScope(ManagedMemoryUseCase.PYTHON, 123);
+ @Test
+ void testDeclareManagedMemoryOperatorScopeUseCaseFailWrongScope() {
+ assertThatThrownBy(
+ () ->
+ transformation.declareManagedMemoryUseCaseAtOperatorScope(
+ ManagedMemoryUseCase.PYTHON, 123))
+ .isInstanceOf(IllegalArgumentException.class);
}
- @Test(expected = IllegalArgumentException.class)
- public void testDeclareManagedMemoryOperatorScopeUseCaseFailZeroWeight() {
- transformation.declareManagedMemoryUseCaseAtOperatorScope(ManagedMemoryUseCase.OPERATOR, 0);
+ @Test
+ void testDeclareManagedMemoryOperatorScopeUseCaseFailZeroWeight() {
+ assertThatThrownBy(
+ () ->
+ transformation.declareManagedMemoryUseCaseAtOperatorScope(
+ ManagedMemoryUseCase.OPERATOR, 0))
+ .isInstanceOf(IllegalArgumentException.class);
}
- @Test(expected = IllegalArgumentException.class)
- public void testDeclareManagedMemoryOperatorScopeUseCaseFailNegativeWeight() {
- transformation.declareManagedMemoryUseCaseAtOperatorScope(
- ManagedMemoryUseCase.OPERATOR, -1);
+ @Test
+ void testDeclareManagedMemoryOperatorScopeUseCaseFailNegativeWeight() {
+ assertThatThrownBy(
+ () ->
+ transformation.declareManagedMemoryUseCaseAtOperatorScope(
+ ManagedMemoryUseCase.OPERATOR, -1))
+ .isInstanceOf(IllegalArgumentException.class);
}
- @Test(expected = IllegalArgumentException.class)
- public void testDeclareManagedMemorySlotScopeUseCaseFailWrongScope() {
- transformation.declareManagedMemoryUseCaseAtSlotScope(ManagedMemoryUseCase.OPERATOR);
+ @Test
+ void testDeclareManagedMemorySlotScopeUseCaseFailWrongScope() {
+ assertThatThrownBy(
+ () ->
+ transformation.declareManagedMemoryUseCaseAtSlotScope(
+ ManagedMemoryUseCase.OPERATOR))
+ .isInstanceOf(IllegalArgumentException.class);
}
/** A test implementation of {@link Transformation}. */