You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by al...@apache.org on 2014/10/02 09:57:29 UTC
git commit: Move DeltaIteration Sanity Check to Base Operators
Repository: incubator-flink
Updated Branches:
refs/heads/master 02c084564 -> b689f3fca
Move DeltaIteration Sanity Check to Base Operators
Is now in JoinOperator and CoGroupOperator. We don't need
the special Scala API sanity check anymore now since they
use java operators that now correctly check in the base
class.
Project: http://git-wip-us.apache.org/repos/asf/incubator-flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-flink/commit/b689f3fc
Tree: http://git-wip-us.apache.org/repos/asf/incubator-flink/tree/b689f3fc
Diff: http://git-wip-us.apache.org/repos/asf/incubator-flink/diff/b689f3fc
Branch: refs/heads/master
Commit: b689f3fcacf066c862330e955c11ecf5465ffc0a
Parents: 02c0845
Author: Aljoscha Krettek <al...@gmail.com>
Authored: Thu Oct 2 08:55:30 2014 +0200
Committer: Aljoscha Krettek <al...@gmail.com>
Committed: Thu Oct 2 08:56:16 2014 +0200
----------------------------------------------------------------------
.../api/java/operators/CoGroupOperator.java | 36 ++++++++---------
.../flink/api/java/operators/JoinOperator.java | 42 ++++++++++----------
.../apache/flink/api/scala/coGroupDataSet.scala | 2 -
.../apache/flink/api/scala/joinDataSet.scala | 42 --------------------
.../scala/DeltaIterationSanityCheckTest.scala | 23 +++++------
5 files changed, 48 insertions(+), 97 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/b689f3fc/flink-java/src/main/java/org/apache/flink/api/java/operators/CoGroupOperator.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/operators/CoGroupOperator.java b/flink-java/src/main/java/org/apache/flink/api/java/operators/CoGroupOperator.java
index 5063af7..9ecb9a2 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/operators/CoGroupOperator.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/operators/CoGroupOperator.java
@@ -72,6 +72,24 @@ public class CoGroupOperator<I1, I2, OUT> extends TwoInputUdfOperator<I1, I2, OU
throw new NullPointerException();
}
+ // sanity check solution set key mismatches
+ if (input1 instanceof SolutionSetPlaceHolder) {
+ if (keys1 instanceof FieldPositionKeys) {
+ int[] positions = ((FieldPositionKeys<?>) keys1).computeLogicalKeyPositions();
+ ((SolutionSetPlaceHolder<?>) input1).checkJoinKeyFields(positions);
+ } else {
+ throw new InvalidProgramException("Currently, the solution set may only be CoGrouped with using tuple field positions.");
+ }
+ }
+ if (input2 instanceof SolutionSetPlaceHolder) {
+ if (keys2 instanceof FieldPositionKeys) {
+ int[] positions = ((FieldPositionKeys<?>) keys2).computeLogicalKeyPositions();
+ ((SolutionSetPlaceHolder<?>) input2).checkJoinKeyFields(positions);
+ } else {
+ throw new InvalidProgramException("Currently, the solution set may only be CoGrouped with using tuple field positions.");
+ }
+ }
+
this.keys1 = keys1;
this.keys2 = keys2;
@@ -466,24 +484,6 @@ public class CoGroupOperator<I1, I2, OUT> extends TwoInputUdfOperator<I1, I2, OU
if (!keys1.areCompatibale(keys2)) {
throw new InvalidProgramException("The pair of co-group keys are not compatible with each other.");
}
-
- // sanity check solution set key mismatches
- if (input1 instanceof SolutionSetPlaceHolder) {
- if (keys1 instanceof FieldPositionKeys) {
- int[] positions = ((FieldPositionKeys<?>) keys1).computeLogicalKeyPositions();
- ((SolutionSetPlaceHolder<?>) input1).checkJoinKeyFields(positions);
- } else {
- throw new InvalidProgramException("Currently, the solution set may only be CoGrouped with using tuple field positions.");
- }
- }
- if (input2 instanceof SolutionSetPlaceHolder) {
- if (keys2 instanceof FieldPositionKeys) {
- int[] positions = ((FieldPositionKeys<?>) keys2).computeLogicalKeyPositions();
- ((SolutionSetPlaceHolder<?>) input2).checkJoinKeyFields(positions);
- } else {
- throw new InvalidProgramException("Currently, the solution set may only be CoGrouped with using tuple field positions.");
- }
- }
return new CoGroupOperatorWithoutFunction(keys2);
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/b689f3fc/flink-java/src/main/java/org/apache/flink/api/java/operators/JoinOperator.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/operators/JoinOperator.java b/flink-java/src/main/java/org/apache/flink/api/java/operators/JoinOperator.java
index 704bf1b..bc35c14 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/operators/JoinOperator.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/operators/JoinOperator.java
@@ -121,7 +121,25 @@ public abstract class JoinOperator<I1, I2, OUT> extends TwoInputUdfOperator<I1,
if (keys1 == null || keys2 == null) {
throw new NullPointerException();
}
-
+
+ // sanity check solution set key mismatches
+ if (input1 instanceof SolutionSetPlaceHolder) {
+ if (keys1 instanceof FieldPositionKeys) {
+ int[] positions = ((FieldPositionKeys<?>) keys1).computeLogicalKeyPositions();
+ ((SolutionSetPlaceHolder<?>) input1).checkJoinKeyFields(positions);
+ } else {
+ throw new InvalidProgramException("Currently, the solution set may only be joined with using tuple field positions.");
+ }
+ }
+ if (input2 instanceof SolutionSetPlaceHolder) {
+ if (keys2 instanceof FieldPositionKeys) {
+ int[] positions = ((FieldPositionKeys<?>) keys2).computeLogicalKeyPositions();
+ ((SolutionSetPlaceHolder<?>) input2).checkJoinKeyFields(positions);
+ } else {
+ throw new InvalidProgramException("Currently, the solution set may only be joined with using tuple field positions.");
+ }
+ }
+
this.keys1 = keys1;
this.keys2 = keys2;
this.joinHint = hint;
@@ -872,27 +890,7 @@ public abstract class JoinOperator<I1, I2, OUT> extends TwoInputUdfOperator<I1,
if (!keys1.areCompatibale(keys2)) {
throw new InvalidProgramException("The pair of join keys are not compatible with each other.");
}
-
-
- // sanity check solution set key mismatches
- if (input1 instanceof SolutionSetPlaceHolder) {
- if (keys1 instanceof FieldPositionKeys) {
- int[] positions = ((FieldPositionKeys<?>) keys1).computeLogicalKeyPositions();
- ((SolutionSetPlaceHolder<?>) input1).checkJoinKeyFields(positions);
- } else {
- throw new InvalidProgramException("Currently, the solution set may only be joined with using tuple field positions.");
- }
- }
- if (input2 instanceof SolutionSetPlaceHolder) {
- if (keys2 instanceof FieldPositionKeys) {
- int[] positions = ((FieldPositionKeys<?>) keys2).computeLogicalKeyPositions();
- ((SolutionSetPlaceHolder<?>) input2).checkJoinKeyFields(positions);
- } else {
- throw new InvalidProgramException("Currently, the solution set may only be joined with using tuple field positions.");
- }
- }
-
-
+
return new DefaultJoin<I1, I2>(input1, input2, keys1, keys2, joinHint);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/b689f3fc/flink-scala/src/main/scala/org/apache/flink/api/scala/coGroupDataSet.scala
----------------------------------------------------------------------
diff --git a/flink-scala/src/main/scala/org/apache/flink/api/scala/coGroupDataSet.scala b/flink-scala/src/main/scala/org/apache/flink/api/scala/coGroupDataSet.scala
index 5c1d333..cc3d836 100644
--- a/flink-scala/src/main/scala/org/apache/flink/api/scala/coGroupDataSet.scala
+++ b/flink-scala/src/main/scala/org/apache/flink/api/scala/coGroupDataSet.scala
@@ -192,8 +192,6 @@ class UnfinishedCoGroupOperation[L: ClassTag, R: ClassTag](
val coGroupOperator = new CoGroupOperator[L, R, (Array[L], Array[R])](
leftInput.javaSet, rightInput.javaSet, leftKey, rightKey, coGrouper, returnType)
- DeltaIterationSanityCheck(leftInput, rightInput, leftKey, rightKey)
-
new CoGroupDataSet(coGroupOperator, leftInput, rightInput, leftKey, rightKey)
}
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/b689f3fc/flink-scala/src/main/scala/org/apache/flink/api/scala/joinDataSet.scala
----------------------------------------------------------------------
diff --git a/flink-scala/src/main/scala/org/apache/flink/api/scala/joinDataSet.scala b/flink-scala/src/main/scala/org/apache/flink/api/scala/joinDataSet.scala
index 3dca756..d333a66 100644
--- a/flink-scala/src/main/scala/org/apache/flink/api/scala/joinDataSet.scala
+++ b/flink-scala/src/main/scala/org/apache/flink/api/scala/joinDataSet.scala
@@ -203,49 +203,7 @@ class UnfinishedJoinOperation[L, R](
val joinOperator = new EquiJoin[L, R, (L, R)](
leftSet.javaSet, rightSet.javaSet, leftKey, rightKey, joiner, returnType, joinHint)
- DeltaIterationSanityCheck(leftSet, rightSet, leftKey, rightKey)
-
new JoinDataSet(joinOperator, leftSet, rightSet, leftKey, rightKey)
}
}
-
-/**
- * This checks whether joining/coGrouping with the DeltaIteration SolutionSet uses the
- * same key given when creating the DeltaIteration.
- */
-private[flink] object DeltaIterationSanityCheck {
- def apply[L, R](
- leftSet: DataSet[L],
- rightSet: DataSet[R],
- leftKey: Keys[L],
- rightKey: Keys[R]) = {
- // sanity check solution set key mismatches
- leftSet.javaSet match {
- case solutionSet: DeltaIteration.SolutionSetPlaceHolder[_] =>
- leftKey match {
- case keyFields: Keys.FieldPositionKeys[_] =>
- val positions: Array[Int] = keyFields.computeLogicalKeyPositions
- solutionSet.checkJoinKeyFields(positions)
- case _ =>
- throw new InvalidProgramException("Currently, the solution set may only be joined " +
- "with " +
- "using tuple field positions.")
- }
- case _ =>
- }
- rightSet.javaSet match {
- case solutionSet: DeltaIteration.SolutionSetPlaceHolder[_] =>
- rightKey match {
- case keyFields: Keys.FieldPositionKeys[_] =>
- val positions: Array[Int] = keyFields.computeLogicalKeyPositions
- solutionSet.checkJoinKeyFields(positions)
- case _ =>
- throw new InvalidProgramException("Currently, the solution set may only be joined " +
- "with " +
- "using tuple field positions.")
- }
- case _ =>
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/b689f3fc/flink-scala/src/test/scala/org/apache/flink/api/scala/DeltaIterationSanityCheckTest.scala
----------------------------------------------------------------------
diff --git a/flink-scala/src/test/scala/org/apache/flink/api/scala/DeltaIterationSanityCheckTest.scala b/flink-scala/src/test/scala/org/apache/flink/api/scala/DeltaIterationSanityCheckTest.scala
index f809cb4..729fda4 100644
--- a/flink-scala/src/test/scala/org/apache/flink/api/scala/DeltaIterationSanityCheckTest.scala
+++ b/flink-scala/src/test/scala/org/apache/flink/api/scala/DeltaIterationSanityCheckTest.scala
@@ -21,9 +21,6 @@ package org.apache.flink.api.scala
import org.junit.Test
import org.apache.flink.api.common.InvalidProgramException
-import org.apache.flink.api.scala._
-import org.scalatest.junit.AssertionsForJUnit
-
// Verify that the sanity checking in delta iterations works. We just
// have a dummy job that is not meant to be executed. Only verify that
// the join/coGroup inside the iteration is checked.
@@ -40,7 +37,7 @@ class DeltaIterationSanityCheckTest extends Serializable {
(result, ws)
}
- val output = iteration.print()
+ iteration.print()
}
@Test
@@ -54,7 +51,7 @@ class DeltaIterationSanityCheckTest extends Serializable {
(result, ws)
}
- val output = iteration.print()
+ iteration.print()
}
@Test(expected = classOf[InvalidProgramException])
@@ -68,7 +65,7 @@ class DeltaIterationSanityCheckTest extends Serializable {
(result, ws)
}
- val output = iteration.print()
+ iteration.print()
}
@Test(expected = classOf[InvalidProgramException])
@@ -82,7 +79,7 @@ class DeltaIterationSanityCheckTest extends Serializable {
(result, ws)
}
- val output = iteration.print() }
+ iteration.print() }
@Test(expected = classOf[InvalidProgramException])
def testIncorrectJoinWithSolution3(): Unit = {
@@ -95,7 +92,7 @@ class DeltaIterationSanityCheckTest extends Serializable {
(result, ws)
}
- val output = iteration.print()
+ iteration.print()
}
@Test
@@ -109,7 +106,7 @@ class DeltaIterationSanityCheckTest extends Serializable {
(result, ws)
}
- val output = iteration.print()
+ iteration.print()
}
@Test
@@ -123,7 +120,7 @@ class DeltaIterationSanityCheckTest extends Serializable {
(result, ws)
}
- val output = iteration.print()
+ iteration.print()
}
@Test(expected = classOf[InvalidProgramException])
@@ -137,7 +134,7 @@ class DeltaIterationSanityCheckTest extends Serializable {
(result, ws)
}
- val output = iteration.print()
+ iteration.print()
}
@Test(expected = classOf[InvalidProgramException])
@@ -151,7 +148,7 @@ class DeltaIterationSanityCheckTest extends Serializable {
(result, ws)
}
- val output = iteration.print() }
+ iteration.print() }
@Test(expected = classOf[InvalidProgramException])
def testIncorrectCoGroupWithSolution3(): Unit = {
@@ -164,6 +161,6 @@ class DeltaIterationSanityCheckTest extends Serializable {
(result, ws)
}
- val output = iteration.print()
+ iteration.print()
}
}