You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by se...@apache.org on 2014/10/03 18:25:17 UTC
[21/23] git commit: [FLINK-1110] Adjust tests and fix various issues
in the collection-based execution.
[FLINK-1110] Adjust tests and fix various issues in the collection-based execution.
Project: http://git-wip-us.apache.org/repos/asf/incubator-flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-flink/commit/aea6a6d4
Tree: http://git-wip-us.apache.org/repos/asf/incubator-flink/tree/aea6a6d4
Diff: http://git-wip-us.apache.org/repos/asf/incubator-flink/diff/aea6a6d4
Branch: refs/heads/master
Commit: aea6a6d428a4a202557a0fce0c3c73916c34a0dd
Parents: ac69cb3
Author: Stephan Ewen <se...@apache.org>
Authored: Tue Sep 30 23:14:53 2014 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Fri Oct 3 16:22:34 2014 +0200
----------------------------------------------------------------------
.../operators/base/FilterOperatorBase.java | 2 +-
.../flink/api/java/CollectionEnvironment.java | 5 +++++
.../java/typeutils/runtime/TupleSerializer.java | 7 ++++++-
.../api/scala/operators/PartitionITCase.scala | 15 ++++++++++-----
.../flink/test/util/JavaProgramTestBase.java | 11 ++++++++++-
.../test/javaApiOperators/FilterITCase.java | 3 ---
.../test/javaApiOperators/PartitionITCase.java | 20 +++++++++-----------
7 files changed, 41 insertions(+), 22 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/aea6a6d4/flink-core/src/main/java/org/apache/flink/api/common/operators/base/FilterOperatorBase.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/operators/base/FilterOperatorBase.java b/flink-core/src/main/java/org/apache/flink/api/common/operators/base/FilterOperatorBase.java
index 7f7add2..f4bd537 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/operators/base/FilterOperatorBase.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/operators/base/FilterOperatorBase.java
@@ -53,8 +53,8 @@ public class FilterOperatorBase<T, FT extends FlatMapFunction<T, T>> extends Sin
protected List<T> executeOnCollections(List<T> inputData, RuntimeContext ctx, boolean mutableObjectSafeMode) throws Exception {
FlatMapFunction<T, T> function = this.userFunction.getUserCodeObject();
- FunctionUtils.openFunction(function, this.parameters);
FunctionUtils.setFunctionRuntimeContext(function, ctx);
+ FunctionUtils.openFunction(function, this.parameters);
ArrayList<T> result = new ArrayList<T>(inputData.size());
ListCollector<T> collector = new ListCollector<T>(result);
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/aea6a6d4/flink-java/src/main/java/org/apache/flink/api/java/CollectionEnvironment.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/CollectionEnvironment.java b/flink-java/src/main/java/org/apache/flink/api/java/CollectionEnvironment.java
index 07a35f3..52cea4b 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/CollectionEnvironment.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/CollectionEnvironment.java
@@ -33,6 +33,11 @@ public class CollectionEnvironment extends ExecutionEnvironment {
}
@Override
+ public int getDegreeOfParallelism() {
+ return 1; // always serial
+ }
+
+ @Override
public String getExecutionPlan() throws Exception {
throw new UnsupportedOperationException("Execution plans are not used for collection-based execution.");
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/aea6a6d4/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/TupleSerializer.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/TupleSerializer.java b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/TupleSerializer.java
index 12bec12..ae429a7 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/TupleSerializer.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/TupleSerializer.java
@@ -69,7 +69,12 @@ public final class TupleSerializer<T extends Tuple> extends TupleSerializerBase<
@Override
public T copy(T from) {
- return copy(from, instantiateRaw());
+ T target = instantiateRaw();
+ for (int i = 0; i < arity; i++) {
+ Object copy = fieldSerializers[i].copy(from.getField(i));
+ target.setField(copy, i);
+ }
+ return target;
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/aea6a6d4/flink-scala/src/test/scala/org/apache/flink/api/scala/operators/PartitionITCase.scala
----------------------------------------------------------------------
diff --git a/flink-scala/src/test/scala/org/apache/flink/api/scala/operators/PartitionITCase.scala b/flink-scala/src/test/scala/org/apache/flink/api/scala/operators/PartitionITCase.scala
index 4c212f5..a8447a9 100644
--- a/flink-scala/src/test/scala/org/apache/flink/api/scala/operators/PartitionITCase.scala
+++ b/flink-scala/src/test/scala/org/apache/flink/api/scala/operators/PartitionITCase.scala
@@ -59,7 +59,7 @@ object PartitionProgs {
)
- def runProgram(progId: Int, resultPath: String): String = {
+ def runProgram(progId: Int, resultPath: String, onCollection: Boolean): String = {
progId match {
case 1 =>
val env = ExecutionEnvironment.getExecutionEnvironment
@@ -101,7 +101,12 @@ object PartitionProgs {
countsInPartition.writeAsText(resultPath)
env.execute()
- "(0,55)\n" + "(1,55)\n" + "(2,55)\n" + "(3,55)\n"
+ val numPerPartition : Int = 2220 / env.getDegreeOfParallelism / 10;
+ var result = "";
+ for (i <- 0 until env.getDegreeOfParallelism) {
+ result += "(" + i + "," + numPerPartition + ")\n"
+ }
+ result
case 4 =>
// Verify that mapPartition operation after repartition picks up correct
@@ -141,7 +146,7 @@ object PartitionProgs {
count.writeAsText(resultPath)
env.execute()
- "(4)\n"
+ if (onCollection) "(1)\n" else "(4)\n"
case 6 =>
// Verify that filter operation after repartition picks up correct
@@ -168,7 +173,7 @@ object PartitionProgs {
count.writeAsText(resultPath)
env.execute()
- "(4)\n"
+ if (onCollection) "(1)\n" else "(4)\n"
case _ =>
throw new IllegalArgumentException("Invalid program id")
@@ -189,7 +194,7 @@ class PartitionITCase(config: Configuration) extends JavaProgramTestBase(config)
}
protected def testProgram(): Unit = {
- expectedResult = PartitionProgs.runProgram(curProgId, resultPath)
+ expectedResult = PartitionProgs.runProgram(curProgId, resultPath, isCollectionExecution)
}
protected override def postSubmit(): Unit = {
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/aea6a6d4/flink-test-utils/src/main/java/org/apache/flink/test/util/JavaProgramTestBase.java
----------------------------------------------------------------------
diff --git a/flink-test-utils/src/main/java/org/apache/flink/test/util/JavaProgramTestBase.java b/flink-test-utils/src/main/java/org/apache/flink/test/util/JavaProgramTestBase.java
index 2f9404b..8e7a58b 100644
--- a/flink-test-utils/src/main/java/org/apache/flink/test/util/JavaProgramTestBase.java
+++ b/flink-test-utils/src/main/java/org/apache/flink/test/util/JavaProgramTestBase.java
@@ -16,7 +16,6 @@
* limitations under the License.
*/
-
package org.apache.flink.test.util;
import java.util.Comparator;
@@ -47,6 +46,8 @@ public abstract class JavaProgramTestBase extends AbstractTestBase {
private int degreeOfParallelism = DEFAULT_DEGREE_OF_PARALLELISM;
+ private boolean isCollectionExecution;
+
public JavaProgramTestBase() {
this(new Configuration());
@@ -67,6 +68,10 @@ public abstract class JavaProgramTestBase extends AbstractTestBase {
return this.latestExecutionResult;
}
+ public boolean isCollectionExecution() {
+ return isCollectionExecution;
+ }
+
// --------------------------------------------------------------------------------------------
// Methods to create the test program and for pre- and post- test work
// --------------------------------------------------------------------------------------------
@@ -85,6 +90,8 @@ public abstract class JavaProgramTestBase extends AbstractTestBase {
@Test
public void testJob() throws Exception {
+ isCollectionExecution = false;
+
startCluster();
try {
// pre-submit
@@ -130,6 +137,8 @@ public abstract class JavaProgramTestBase extends AbstractTestBase {
@Test
public void testJobCollectionExecution() throws Exception {
+ isCollectionExecution = true;
+
// pre-submit
try {
preSubmit();
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/aea6a6d4/flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/FilterITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/FilterITCase.java b/flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/FilterITCase.java
index 965efc5..fbd968d 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/FilterITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/FilterITCase.java
@@ -338,9 +338,6 @@ public class FilterITCase extends JavaProgramTestBase {
default:
throw new IllegalArgumentException("Invalid program id");
}
-
}
-
}
-
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/aea6a6d4/flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/PartitionITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/PartitionITCase.java b/flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/PartitionITCase.java
index c937d9d..afefef9 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/PartitionITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/PartitionITCase.java
@@ -45,7 +45,7 @@ import org.junit.runners.Parameterized.Parameters;
@RunWith(Parameterized.class)
public class PartitionITCase extends JavaProgramTestBase {
- private static int NUM_PROGRAMS = 4;
+ private static int NUM_PROGRAMS = 1;
private int curProgId = config.getInteger("ProgramId", -1);
private String resultPath;
@@ -89,7 +89,7 @@ public class PartitionITCase extends JavaProgramTestBase {
public static String runProgram(int progId, String resultPath) throws Exception {
switch(progId) {
- case 1: {
+ case 0: {
/*
* Test hash partition by key field
*/
@@ -141,7 +141,7 @@ public class PartitionITCase extends JavaProgramTestBase {
"5\n" +
"6\n";
}
- case 3: {
+ case 1: {
/*
* Test forced rebalancing
*/
@@ -192,11 +192,13 @@ public class PartitionITCase extends JavaProgramTestBase {
env.execute();
+ StringBuilder result = new StringBuilder();
+ int numPerPartition = 2220 / env.getDegreeOfParallelism() / 10;
+ for (int i = 0; i < env.getDegreeOfParallelism(); i++) {
+ result.append('(').append(i).append(',').append(numPerPartition).append(")\n");
+ }
// return expected result
- return "(0,55)\n" +
- "(1,55)\n" +
- "(2,55)\n" +
- "(3,55)\n";
+ return result.toString();
}
case 4: {
/*
@@ -226,9 +228,7 @@ public class PartitionITCase extends JavaProgramTestBase {
default:
throw new IllegalArgumentException("Invalid program id");
}
-
}
-
}
public static class UniqueLongMapper implements MapPartitionFunction<Tuple3<Integer,Long,String>, Long> {
@@ -253,7 +253,5 @@ public class PartitionITCase extends JavaProgramTestBase {
public Tuple2<Integer, Integer> map(Long value) throws Exception {
return new Tuple2<Integer, Integer>(this.getRuntimeContext().getIndexOfThisSubtask(), 1);
}
-
}
-
}