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/07/09 12:21:28 UTC
[01/12] git commit: [FLINK-836] Integration of the
CachedBuildSideMatchDriver into the optimizer
Repository: incubator-flink
Updated Branches:
refs/heads/master 7b6295d98 -> 26bc321c3
[FLINK-836] Integration of the CachedBuildSideMatchDriver into the optimizer
Project: http://git-wip-us.apache.org/repos/asf/incubator-flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-flink/commit/48a703fd
Tree: http://git-wip-us.apache.org/repos/asf/incubator-flink/tree/48a703fd
Diff: http://git-wip-us.apache.org/repos/asf/incubator-flink/diff/48a703fd
Branch: refs/heads/master
Commit: 48a703fdd89d7d9b9daf7828636c38f3ebddcd65
Parents: 7b6295d
Author: Markus Holzemer <ma...@gmx.de>
Authored: Mon May 19 10:54:34 2014 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Tue Jul 8 14:36:02 2014 +0200
----------------------------------------------------------------------
.../compiler/costs/CostEstimator.java | 41 ++++++++++++--------
.../compiler/costs/DefaultCostEstimator.java | 32 +++++++++++++--
.../compiler/dag/OptimizerNode.java | 10 +++++
.../operators/HashJoinBuildFirstProperties.java | 10 ++++-
.../HashJoinBuildSecondProperties.java | 10 ++++-
.../eu/stratosphere/compiler/plan/Channel.java | 4 ++
.../eu/stratosphere/compiler/plan/PlanNode.java | 4 ++
.../pact/runtime/task/DriverStrategy.java | 4 ++
8 files changed, 93 insertions(+), 22 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/48a703fd/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/costs/CostEstimator.java
----------------------------------------------------------------------
diff --git a/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/costs/CostEstimator.java b/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/costs/CostEstimator.java
index 3e1d94a..3f0c094 100644
--- a/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/costs/CostEstimator.java
+++ b/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/costs/CostEstimator.java
@@ -42,13 +42,15 @@ public abstract class CostEstimator {
public abstract void addLocalSortCost(EstimateProvider estimates, long memorySize, Costs costs);
- public abstract void addLocalMergeCost(EstimateProvider estimates1, EstimateProvider estimates2, long memorySize, Costs costs);
+ public abstract void addLocalMergeCost(EstimateProvider estimates1, EstimateProvider estimates2, long memorySize, Costs costs, int costWeight);
- public abstract void addHybridHashCosts(EstimateProvider buildSide, EstimateProvider probeSide, long memorySize, Costs costs);
+ public abstract void addHybridHashCosts(EstimateProvider buildSide, EstimateProvider probeSide, long memorySize, Costs costs, int costWeight);
+
+ public abstract void addCachedHybridHashCosts(EstimateProvider buildSide, EstimateProvider probeSide, long memorySize, Costs costs, int costWeight);
- public abstract void addStreamedNestedLoopsCosts(EstimateProvider outerSide, EstimateProvider innerSide, long bufferSize, Costs costs);
+ public abstract void addStreamedNestedLoopsCosts(EstimateProvider outerSide, EstimateProvider innerSide, long bufferSize, Costs costs, int costWeight);
- public abstract void addBlockNestedLoopsCosts(EstimateProvider outerSide, EstimateProvider innerSide, long blockSize, Costs costs);
+ public abstract void addBlockNestedLoopsCosts(EstimateProvider outerSide, EstimateProvider innerSide, long blockSize, Costs costs, int costWeight);
// ------------------------------------------------------------------------
@@ -135,6 +137,12 @@ public abstract class CostEstimator {
Channel firstInput = null;
Channel secondInput = null;
Costs driverCosts = new Costs();
+ int costWeight = 1;
+
+ // adjust with the cost weight factor
+ if (n.isOnDynamicPath()) {
+ costWeight = n.getCostWeight();
+ }
// get the inputs, if we have some
{
@@ -173,35 +181,36 @@ public abstract class CostEstimator {
break;
case MERGE:
- addLocalMergeCost(firstInput, secondInput, availableMemory, driverCosts);
+ addLocalMergeCost(firstInput, secondInput, availableMemory, driverCosts, costWeight);
break;
case HYBRIDHASH_BUILD_FIRST:
- addHybridHashCosts(firstInput, secondInput, availableMemory, driverCosts);
+ addHybridHashCosts(firstInput, secondInput, availableMemory, driverCosts, costWeight);
break;
case HYBRIDHASH_BUILD_SECOND:
- addHybridHashCosts(secondInput, firstInput, availableMemory, driverCosts);
+ addHybridHashCosts(secondInput, firstInput, availableMemory, driverCosts, costWeight);
+ break;
+ case HYBRIDHASH_BUILD_FIRST_CACHED:
+ addCachedHybridHashCosts(firstInput, secondInput, availableMemory, driverCosts, costWeight);
+ break;
+ case HYBRIDHASH_BUILD_SECOND_CACHED:
+ addCachedHybridHashCosts(secondInput, firstInput, availableMemory, driverCosts, costWeight);
break;
case NESTEDLOOP_BLOCKED_OUTER_FIRST:
- addBlockNestedLoopsCosts(firstInput, secondInput, availableMemory, driverCosts);
+ addBlockNestedLoopsCosts(firstInput, secondInput, availableMemory, driverCosts, costWeight);
break;
case NESTEDLOOP_BLOCKED_OUTER_SECOND:
- addBlockNestedLoopsCosts(secondInput, firstInput, availableMemory, driverCosts);
+ addBlockNestedLoopsCosts(secondInput, firstInput, availableMemory, driverCosts, costWeight);
break;
case NESTEDLOOP_STREAMED_OUTER_FIRST:
- addStreamedNestedLoopsCosts(firstInput, secondInput, availableMemory, driverCosts);
+ addStreamedNestedLoopsCosts(firstInput, secondInput, availableMemory, driverCosts, costWeight);
break;
case NESTEDLOOP_STREAMED_OUTER_SECOND:
- addStreamedNestedLoopsCosts(secondInput, firstInput, availableMemory, driverCosts);
+ addStreamedNestedLoopsCosts(secondInput, firstInput, availableMemory, driverCosts, costWeight);
break;
default:
throw new CompilerException("Unknown local strategy: " + n.getDriverStrategy().name());
}
- // adjust with the cost weight factor
- if (n.isOnDynamicPath()) {
- driverCosts.multiplyWith(n.getCostWeight());
- }
-
totalCosts.addCosts(driverCosts);
n.setCosts(totalCosts);
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/48a703fd/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/costs/DefaultCostEstimator.java
----------------------------------------------------------------------
diff --git a/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/costs/DefaultCostEstimator.java b/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/costs/DefaultCostEstimator.java
index 3c52f6a..425f93c 100644
--- a/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/costs/DefaultCostEstimator.java
+++ b/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/costs/DefaultCostEstimator.java
@@ -141,13 +141,13 @@ public class DefaultCostEstimator extends CostEstimator {
}
@Override
- public void addLocalMergeCost(EstimateProvider input1, EstimateProvider input2, long availableMemory, Costs costs) {
+ public void addLocalMergeCost(EstimateProvider input1, EstimateProvider input2, long availableMemory, Costs costs, int costWeight) {
// costs nothing. the very rarely incurred cost for a spilling block nested loops join in the
// presence of massively re-occurring duplicate keys is ignored, because not accessible.
}
@Override
- public void addHybridHashCosts(EstimateProvider buildSideInput, EstimateProvider probeSideInput, long availableMemory, Costs costs) {
+ public void addHybridHashCosts(EstimateProvider buildSideInput, EstimateProvider probeSideInput, long availableMemory, Costs costs, int costWeight) {
long bs = buildSideInput.getEstimatedOutputSize();
long ps = probeSideInput.getEstimatedOutputSize();
@@ -160,10 +160,32 @@ public class DefaultCostEstimator extends CostEstimator {
}
costs.addHeuristicDiskCost(2 * HEURISTIC_COST_BASE);
costs.addHeuristicCpuCost((long) (HEURISTIC_COST_BASE * HASHING_CPU_FACTOR));
+ costs.multiplyWith(costWeight);
+ }
+
+ /**
+ * Calculates the costs for the cached variant of the hybrid hash join.
+ * We are assuming by default that half of the cached hash table fit into memory.
+ */
+ @Override
+ public void addCachedHybridHashCosts(EstimateProvider buildSideInput, EstimateProvider probeSideInput, long availableMemory, Costs costs, int costWeight) {
+ long bs = buildSideInput.getEstimatedOutputSize();
+ long ps = probeSideInput.getEstimatedOutputSize();
+
+ if (bs > 0 && ps > 0) {
+ long overallSize = 2*bs + ps;
+ costs.addDiskCost(overallSize / 2 + (overallSize / 2) * costWeight);
+ costs.addCpuCost((long) ((2*bs + ps) * HASHING_CPU_FACTOR));
+ } else {
+ costs.setDiskCost(Costs.UNKNOWN);
+ costs.setCpuCost(Costs.UNKNOWN);
+ }
+ costs.addHeuristicDiskCost(2 * HEURISTIC_COST_BASE * costWeight);
+ costs.addHeuristicCpuCost((long) (HEURISTIC_COST_BASE * HASHING_CPU_FACTOR * costWeight));
}
@Override
- public void addStreamedNestedLoopsCosts(EstimateProvider outerSide, EstimateProvider innerSide, long bufferSize, Costs costs) {
+ public void addStreamedNestedLoopsCosts(EstimateProvider outerSide, EstimateProvider innerSide, long bufferSize, Costs costs, int costWeight) {
long is = innerSide.getEstimatedOutputSize();
long oc = outerSide.getEstimatedNumRecords();
@@ -181,10 +203,11 @@ public class DefaultCostEstimator extends CostEstimator {
// hack: assume 100k loops (should be expensive enough)
costs.addHeuristicDiskCost(HEURISTIC_COST_BASE * 100000);
costs.addHeuristicCpuCost((long) (HEURISTIC_COST_BASE * 100000 * MATERIALIZATION_CPU_FACTOR));
+ costs.multiplyWith(costWeight);
}
@Override
- public void addBlockNestedLoopsCosts(EstimateProvider outerSide, EstimateProvider innerSide, long blockSize, Costs costs) {
+ public void addBlockNestedLoopsCosts(EstimateProvider outerSide, EstimateProvider innerSide, long blockSize, Costs costs, int costWeight) {
long is = innerSide.getEstimatedOutputSize();
long os = outerSide.getEstimatedOutputSize();
@@ -200,6 +223,7 @@ public class DefaultCostEstimator extends CostEstimator {
// hack: assume 1k loops (much cheaper than the streamed variant!)
costs.addHeuristicDiskCost(HEURISTIC_COST_BASE * 1000);
costs.addHeuristicCpuCost((long) (HEURISTIC_COST_BASE * 1000 * MATERIALIZATION_CPU_FACTOR));
+ costs.multiplyWith(costWeight);
}
// --------------------------------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/48a703fd/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/dag/OptimizerNode.java
----------------------------------------------------------------------
diff --git a/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/dag/OptimizerNode.java b/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/dag/OptimizerNode.java
index 85a6568..25f905d 100644
--- a/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/dag/OptimizerNode.java
+++ b/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/dag/OptimizerNode.java
@@ -92,6 +92,8 @@ public abstract class OptimizerNode implements Visitable<OptimizerNode>, Estimat
protected boolean onDynamicPath;
+ protected boolean insideIteration;
+
protected List<PlanNode> cachedPlans; // cache candidates, because the may be accessed repeatedly
protected int[][] remappedKeys;
@@ -499,6 +501,14 @@ public abstract class OptimizerNode implements Visitable<OptimizerNode>, Estimat
}
}
+ public boolean isInsideIteration() {
+ return insideIteration;
+ }
+
+ public void setInsideIteration(boolean insideIteration) {
+ this.insideIteration = insideIteration;
+ }
+
/**
* Checks whether this node has branching output. A node's output is branched, if it has more
* than one output connection.
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/48a703fd/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/operators/HashJoinBuildFirstProperties.java
----------------------------------------------------------------------
diff --git a/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/operators/HashJoinBuildFirstProperties.java b/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/operators/HashJoinBuildFirstProperties.java
index 6506266..661b316 100644
--- a/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/operators/HashJoinBuildFirstProperties.java
+++ b/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/operators/HashJoinBuildFirstProperties.java
@@ -54,7 +54,15 @@ public class HashJoinBuildFirstProperties extends AbstractJoinDescriptor {
@Override
public DualInputPlanNode instantiate(Channel in1, Channel in2, TwoInputNode node) {
- return new DualInputPlanNode(node, "Join("+node.getPactContract().getName()+")", in1, in2, DriverStrategy.HYBRIDHASH_BUILD_FIRST, this.keys1, this.keys2);
+ DriverStrategy strategy;
+
+ if(!in1.isOnDynamicPath() && in1.isInsideIteration() && in2.isInsideIteration()) {
+ strategy = DriverStrategy.HYBRIDHASH_BUILD_FIRST_CACHED;
+ }
+ else {
+ strategy = DriverStrategy.HYBRIDHASH_BUILD_FIRST;
+ }
+ return new DualInputPlanNode(node, "Join("+node.getPactContract().getName()+")", in1, in2, strategy, this.keys1, this.keys2);
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/48a703fd/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/operators/HashJoinBuildSecondProperties.java
----------------------------------------------------------------------
diff --git a/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/operators/HashJoinBuildSecondProperties.java b/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/operators/HashJoinBuildSecondProperties.java
index f092f45..e085588 100644
--- a/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/operators/HashJoinBuildSecondProperties.java
+++ b/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/operators/HashJoinBuildSecondProperties.java
@@ -51,7 +51,15 @@ public final class HashJoinBuildSecondProperties extends AbstractJoinDescriptor
@Override
public DualInputPlanNode instantiate(Channel in1, Channel in2, TwoInputNode node) {
- return new DualInputPlanNode(node, "Join("+node.getPactContract().getName()+")", in1, in2, DriverStrategy.HYBRIDHASH_BUILD_SECOND, this.keys1, this.keys2);
+ DriverStrategy strategy;
+
+ if(!in2.isOnDynamicPath() && in1.isInsideIteration() && in2.isInsideIteration()) {
+ strategy = DriverStrategy.HYBRIDHASH_BUILD_SECOND_CACHED;
+ }
+ else {
+ strategy = DriverStrategy.HYBRIDHASH_BUILD_SECOND;
+ }
+ return new DualInputPlanNode(node, "Join("+node.getPactContract().getName()+")", in1, in2, strategy, this.keys1, this.keys2);
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/48a703fd/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/plan/Channel.java
----------------------------------------------------------------------
diff --git a/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/plan/Channel.java b/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/plan/Channel.java
index 6f9418f..d83da82 100644
--- a/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/plan/Channel.java
+++ b/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/plan/Channel.java
@@ -306,6 +306,10 @@ public class Channel implements EstimateProvider, Cloneable, DumpableConnection<
return this.source.isOnDynamicPath();
}
+ public boolean isInsideIteration() {
+ return this.source.isInsideIteration();
+ }
+
public int getCostWeight() {
return this.source.getCostWeight();
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/48a703fd/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/plan/PlanNode.java
----------------------------------------------------------------------
diff --git a/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/plan/PlanNode.java b/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/plan/PlanNode.java
index 539006c..da24b5f 100644
--- a/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/plan/PlanNode.java
+++ b/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/plan/PlanNode.java
@@ -423,6 +423,10 @@ public abstract class PlanNode implements Visitable<PlanNode>, DumpableNode<Plan
return this.template.getCostWeight();
}
+ public boolean isInsideIteration() {
+ return this.template.isInsideIteration();
+ }
+
// --------------------------------------------------------------------------------------------
/**
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/48a703fd/stratosphere-runtime/src/main/java/eu/stratosphere/pact/runtime/task/DriverStrategy.java
----------------------------------------------------------------------
diff --git a/stratosphere-runtime/src/main/java/eu/stratosphere/pact/runtime/task/DriverStrategy.java b/stratosphere-runtime/src/main/java/eu/stratosphere/pact/runtime/task/DriverStrategy.java
index f96a553..1703494 100644
--- a/stratosphere-runtime/src/main/java/eu/stratosphere/pact/runtime/task/DriverStrategy.java
+++ b/stratosphere-runtime/src/main/java/eu/stratosphere/pact/runtime/task/DriverStrategy.java
@@ -67,6 +67,10 @@ public enum DriverStrategy {
HYBRIDHASH_BUILD_FIRST(MatchDriver.class, null, FULL_DAM, MATERIALIZING, true),
// the second input is build side, the first side is probe side of a hybrid hash table
HYBRIDHASH_BUILD_SECOND(MatchDriver.class, null, MATERIALIZING, FULL_DAM, true),
+ // a cached variant of HYBRIDHASH_BUILD_FIRST, that can only be used inside of iterations
+ HYBRIDHASH_BUILD_FIRST_CACHED(BuildFirstCachedMatchDriver.class, null, FULL_DAM, MATERIALIZING, true),
+ // cached variant of HYBRIDHASH_BUILD_SECOND, that can only be used inside of iterations
+ HYBRIDHASH_BUILD_SECOND_CACHED(BuildSecondCachedMatchDriver.class, null, MATERIALIZING, FULL_DAM, true),
// the second input is inner loop, the first input is outer loop and block-wise processed
NESTEDLOOP_BLOCKED_OUTER_FIRST(CrossDriver.class, null, MATERIALIZING, MATERIALIZING, false),
[06/12] git commit: Clean up exception logging in JobManagerITCase.
Posted by se...@apache.org.
Clean up exception logging in JobManagerITCase.
Project: http://git-wip-us.apache.org/repos/asf/incubator-flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-flink/commit/8ea65360
Tree: http://git-wip-us.apache.org/repos/asf/incubator-flink/tree/8ea65360
Diff: http://git-wip-us.apache.org/repos/asf/incubator-flink/diff/8ea65360
Branch: refs/heads/master
Commit: 8ea6536050ac750f934157728445fac376ecaae3
Parents: 53c7a89
Author: Stephan Ewen <se...@apache.org>
Authored: Tue Jul 8 16:46:47 2014 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Tue Jul 8 16:55:13 2014 +0200
----------------------------------------------------------------------
.../nephele/jobmanager/JobManagerITCase.java | 26 ++++++++++----------
1 file changed, 13 insertions(+), 13 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/8ea65360/stratosphere-runtime/src/test/java/eu/stratosphere/nephele/jobmanager/JobManagerITCase.java
----------------------------------------------------------------------
diff --git a/stratosphere-runtime/src/test/java/eu/stratosphere/nephele/jobmanager/JobManagerITCase.java b/stratosphere-runtime/src/test/java/eu/stratosphere/nephele/jobmanager/JobManagerITCase.java
index ffb958a..635fcab 100644
--- a/stratosphere-runtime/src/test/java/eu/stratosphere/nephele/jobmanager/JobManagerITCase.java
+++ b/stratosphere-runtime/src/test/java/eu/stratosphere/nephele/jobmanager/JobManagerITCase.java
@@ -39,7 +39,6 @@ import eu.stratosphere.core.fs.Path;
import eu.stratosphere.nephele.ExecutionMode;
import eu.stratosphere.nephele.client.JobClient;
import eu.stratosphere.nephele.client.JobExecutionException;
-import eu.stratosphere.nephele.execution.RuntimeEnvironment;
import eu.stratosphere.nephele.jobgraph.DistributionPattern;
import eu.stratosphere.nephele.jobgraph.JobGraph;
import eu.stratosphere.nephele.jobgraph.JobGraphDefinitionException;
@@ -315,10 +314,11 @@ public class JobManagerITCase {
jobClient = new JobClient(jg, configuration);
// deactivate logging of expected test exceptions
- Logger rtLogger = Logger.getLogger(Task.class);
- rtLogger.setLevel(Level.OFF);
- Logger envLogger = Logger.getLogger(RuntimeEnvironment.class);
- envLogger.setLevel(Level.DEBUG);
+ Logger taskLogger = Logger.getLogger(Task.class);
+ taskLogger.setLevel(Level.OFF);
+
+ Logger jmLogger = Logger.getLogger(JobManager.class);
+ jmLogger.setLevel(Level.OFF);
try {
jobClient.submitJobAndWait();
@@ -409,8 +409,11 @@ public class JobManagerITCase {
// deactivate logging of expected test exceptions
Logger jcLogger = Logger.getLogger(JobClient.class);
- Level jcLevel = jcLogger.getEffectiveLevel();
jcLogger.setLevel(Level.OFF);
+ Logger tmLogger = Logger.getLogger(TaskManager.class);
+ tmLogger.setLevel(Level.OFF);
+
+
try {
jobClient.submitJobAndWait();
} catch (JobExecutionException e) {
@@ -426,9 +429,6 @@ public class JobManagerITCase {
// Check if the correct error message is encapsulated in the exception
return;
}
- finally {
- jcLogger.setLevel(jcLevel);
- }
fail("Expected exception but did not receive it");
@@ -515,9 +515,12 @@ public class JobManagerITCase {
jobClient = new JobClient(jg, configuration);
// deactivate logging of expected test exceptions
+ // deactivate logging of expected test exceptions
Logger jcLogger = Logger.getLogger(JobClient.class);
- Level jcLevel = jcLogger.getEffectiveLevel();
jcLogger.setLevel(Level.OFF);
+ Logger tmLogger = Logger.getLogger(TaskManager.class);
+ tmLogger.setLevel(Level.OFF);
+
try {
jobClient.submitJobAndWait();
} catch (JobExecutionException e) {
@@ -534,9 +537,6 @@ public class JobManagerITCase {
// Check if the correct error message is encapsulated in the exception
return;
}
- finally {
- jcLogger.setLevel(jcLevel);
- }
fail("Expected exception but did not receive it");
[10/12] git commit: No explicit caching when hash table is cached Add
tests for explicit cache removal when hash table is cached
Posted by se...@apache.org.
No explicit caching when hash table is cached
Add tests for explicit cache removal when hash table is cached
Project: http://git-wip-us.apache.org/repos/asf/incubator-flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-flink/commit/52f9cf63
Tree: http://git-wip-us.apache.org/repos/asf/incubator-flink/tree/52f9cf63
Diff: http://git-wip-us.apache.org/repos/asf/incubator-flink/diff/52f9cf63
Branch: refs/heads/master
Commit: 52f9cf63e66a53592ea952c154e9e28ab4bce3b7
Parents: a9b1daa
Author: Stephan Ewen <se...@apache.org>
Authored: Wed Jul 9 03:10:45 2014 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Wed Jul 9 03:19:24 2014 +0200
----------------------------------------------------------------------
.../eu/stratosphere/compiler/dag/TempMode.java | 11 ++++++++
.../operators/HashJoinBuildFirstProperties.java | 6 +++++
.../HashJoinBuildSecondProperties.java | 9 ++++++-
.../CachedMatchStrategyCompilerTest.java | 28 +++++++++++++-------
4 files changed, 44 insertions(+), 10 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/52f9cf63/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/dag/TempMode.java
----------------------------------------------------------------------
diff --git a/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/dag/TempMode.java b/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/dag/TempMode.java
index 7907af7..a698422 100644
--- a/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/dag/TempMode.java
+++ b/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/dag/TempMode.java
@@ -63,4 +63,15 @@ public enum TempMode {
return this;
}
}
+
+
+ public TempMode makeNonCached() {
+ if (this == CACHED) {
+ return NONE;
+ } else if (this == CACHING_PIPELINE_BREAKER) {
+ return PIPELINE_BREAKER;
+ } else {
+ return this;
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/52f9cf63/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/operators/HashJoinBuildFirstProperties.java
----------------------------------------------------------------------
diff --git a/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/operators/HashJoinBuildFirstProperties.java b/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/operators/HashJoinBuildFirstProperties.java
index 4f694c5..f8a7c38 100644
--- a/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/operators/HashJoinBuildFirstProperties.java
+++ b/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/operators/HashJoinBuildFirstProperties.java
@@ -17,6 +17,7 @@ import java.util.Collections;
import java.util.List;
import eu.stratosphere.api.common.operators.util.FieldList;
+import eu.stratosphere.compiler.CompilerException;
import eu.stratosphere.compiler.dag.TwoInputNode;
import eu.stratosphere.compiler.dataproperties.LocalProperties;
import eu.stratosphere.compiler.dataproperties.RequestedLocalProperties;
@@ -57,6 +58,11 @@ public class HashJoinBuildFirstProperties extends AbstractJoinDescriptor {
DriverStrategy strategy;
if(!in1.isOnDynamicPath() && in2.isOnDynamicPath()) {
+ // sanity check that the first input is cached and remove that cache
+ if (!in1.getTempMode().isCached()) {
+ throw new CompilerException("No cache at point where static and dynamic parts meet.");
+ }
+ in1.setTempMode(in1.getTempMode().makeNonCached());
strategy = DriverStrategy.HYBRIDHASH_BUILD_FIRST_CACHED;
}
else {
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/52f9cf63/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/operators/HashJoinBuildSecondProperties.java
----------------------------------------------------------------------
diff --git a/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/operators/HashJoinBuildSecondProperties.java b/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/operators/HashJoinBuildSecondProperties.java
index 6bea65a..6982ce6 100644
--- a/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/operators/HashJoinBuildSecondProperties.java
+++ b/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/operators/HashJoinBuildSecondProperties.java
@@ -17,6 +17,7 @@ import java.util.Collections;
import java.util.List;
import eu.stratosphere.api.common.operators.util.FieldList;
+import eu.stratosphere.compiler.CompilerException;
import eu.stratosphere.compiler.dag.TwoInputNode;
import eu.stratosphere.compiler.dataproperties.LocalProperties;
import eu.stratosphere.compiler.dataproperties.RequestedLocalProperties;
@@ -53,7 +54,13 @@ public final class HashJoinBuildSecondProperties extends AbstractJoinDescriptor
public DualInputPlanNode instantiate(Channel in1, Channel in2, TwoInputNode node) {
DriverStrategy strategy;
- if(!in2.isOnDynamicPath() && in1.isOnDynamicPath()) {
+ if (!in2.isOnDynamicPath() && in1.isOnDynamicPath()) {
+ // sanity check that the first input is cached and remove that cache
+ if (!in2.getTempMode().isCached()) {
+ throw new CompilerException("No cache at point where static and dynamic parts meet.");
+ }
+
+ in2.setTempMode(in2.getTempMode().makeNonCached());
strategy = DriverStrategy.HYBRIDHASH_BUILD_SECOND_CACHED;
}
else {
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/52f9cf63/stratosphere-compiler/src/test/java/eu/stratosphere/pact/compiler/CachedMatchStrategyCompilerTest.java
----------------------------------------------------------------------
diff --git a/stratosphere-compiler/src/test/java/eu/stratosphere/pact/compiler/CachedMatchStrategyCompilerTest.java b/stratosphere-compiler/src/test/java/eu/stratosphere/pact/compiler/CachedMatchStrategyCompilerTest.java
index b82c0f9..33fc6e4 100644
--- a/stratosphere-compiler/src/test/java/eu/stratosphere/pact/compiler/CachedMatchStrategyCompilerTest.java
+++ b/stratosphere-compiler/src/test/java/eu/stratosphere/pact/compiler/CachedMatchStrategyCompilerTest.java
@@ -28,6 +28,7 @@ import eu.stratosphere.api.java.IterativeDataSet;
import eu.stratosphere.api.java.functions.JoinFunction;
import eu.stratosphere.api.java.tuple.Tuple3;
import eu.stratosphere.compiler.PactCompiler;
+import eu.stratosphere.compiler.dag.TempMode;
import eu.stratosphere.compiler.plan.DualInputPlanNode;
import eu.stratosphere.compiler.plan.OptimizedPlan;
import eu.stratosphere.compiler.plantranslate.NepheleJobGraphGenerator;
@@ -57,6 +58,8 @@ public class CachedMatchStrategyCompilerTest extends CompilerTestBase {
// verify correct join strategy
assertEquals(DriverStrategy.HYBRIDHASH_BUILD_SECOND_CACHED, innerJoin.getDriverStrategy());
+ assertEquals(TempMode.NONE, innerJoin.getInput1().getTempMode());
+ assertEquals(TempMode.NONE, innerJoin.getInput2().getTempMode());
new NepheleJobGraphGenerator().compileJobGraph(oPlan);
}
@@ -83,6 +86,8 @@ public class CachedMatchStrategyCompilerTest extends CompilerTestBase {
// verify correct join strategy
assertEquals(DriverStrategy.HYBRIDHASH_BUILD_FIRST, innerJoin.getDriverStrategy());
+ assertEquals(TempMode.NONE, innerJoin.getInput1().getTempMode());
+ assertEquals(TempMode.CACHED, innerJoin.getInput2().getTempMode());
new NepheleJobGraphGenerator().compileJobGraph(oPlan);
}
@@ -109,7 +114,9 @@ public class CachedMatchStrategyCompilerTest extends CompilerTestBase {
DualInputPlanNode innerJoin = resolver.getNode("DummyJoiner");
// verify correct join strategy
- assertEquals(DriverStrategy.HYBRIDHASH_BUILD_FIRST_CACHED, innerJoin.getDriverStrategy());
+ assertEquals(DriverStrategy.HYBRIDHASH_BUILD_FIRST_CACHED, innerJoin.getDriverStrategy());
+ assertEquals(TempMode.NONE, innerJoin.getInput1().getTempMode());
+ assertEquals(TempMode.NONE, innerJoin.getInput2().getTempMode());
new NepheleJobGraphGenerator().compileJobGraph(oPlan);
}
@@ -135,7 +142,9 @@ public class CachedMatchStrategyCompilerTest extends CompilerTestBase {
DualInputPlanNode innerJoin = resolver.getNode("DummyJoiner");
// verify correct join strategy
- assertEquals(DriverStrategy.HYBRIDHASH_BUILD_SECOND, innerJoin.getDriverStrategy());
+ assertEquals(DriverStrategy.HYBRIDHASH_BUILD_SECOND, innerJoin.getDriverStrategy());
+ assertEquals(TempMode.CACHED, innerJoin.getInput1().getTempMode());
+ assertEquals(TempMode.NONE, innerJoin.getInput2().getTempMode());
new NepheleJobGraphGenerator().compileJobGraph(oPlan);
}
@@ -169,13 +178,16 @@ public class CachedMatchStrategyCompilerTest extends CompilerTestBase {
}
}
+
OptimizedPlan oPlan = compileNoStats(plan);
OptimizerPlanNodeResolver resolver = getOptimizerPlanNodeResolver(oPlan);
DualInputPlanNode innerJoin = resolver.getNode("DummyJoiner");
// verify correct join strategy
- assertEquals(DriverStrategy.HYBRIDHASH_BUILD_SECOND_CACHED, innerJoin.getDriverStrategy());
+ assertEquals(DriverStrategy.HYBRIDHASH_BUILD_SECOND_CACHED, innerJoin.getDriverStrategy());
+ assertEquals(TempMode.NONE, innerJoin.getInput1().getTempMode());
+ assertEquals(TempMode.NONE, innerJoin.getInput2().getTempMode());
new NepheleJobGraphGenerator().compileJobGraph(oPlan);
}
@@ -197,16 +209,14 @@ public class CachedMatchStrategyCompilerTest extends CompilerTestBase {
IterativeDataSet<Tuple3<Long, Long, Long>> iteration = bigInput.iterate(10);
- DataSet<Tuple3<Long, Long, Long>> inner;
+ Configuration joinStrategy = new Configuration();
+ joinStrategy.setString(PactCompiler.HINT_SHIP_STRATEGY, PactCompiler.HINT_SHIP_STRATEGY_REPARTITION_HASH);
if(strategy != "") {
- Configuration joinStrategy = new Configuration();
joinStrategy.setString(PactCompiler.HINT_LOCAL_STRATEGY, strategy);
- inner = iteration.join(smallInput).where(0).equalTo(0).with(new DummyJoiner()).name("DummyJoiner").withParameters(joinStrategy);
- }
- else {
- inner = iteration.join(smallInput).where(0).equalTo(0).with(new DummyJoiner()).name("DummyJoiner");
}
+
+ DataSet<Tuple3<Long, Long, Long>> inner = iteration.join(smallInput).where(0).equalTo(0).with(new DummyJoiner()).name("DummyJoiner").withParameters(joinStrategy);
DataSet<Tuple3<Long, Long, Long>> output = iteration.closeWith(inner);
[02/12] git commit: [FLINK-795] Tests for the integration of the
cached match driver into the optimizer
Posted by se...@apache.org.
[FLINK-795] Tests for the integration of the cached match driver into the optimizer
Project: http://git-wip-us.apache.org/repos/asf/incubator-flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-flink/commit/99c888c7
Tree: http://git-wip-us.apache.org/repos/asf/incubator-flink/tree/99c888c7
Diff: http://git-wip-us.apache.org/repos/asf/incubator-flink/diff/99c888c7
Branch: refs/heads/master
Commit: 99c888c7b775e1c3c910b9b4a45bbae5102bac14
Parents: 48a703f
Author: Markus Holzemer <ma...@gmx.de>
Authored: Tue May 20 10:00:40 2014 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Tue Jul 8 16:54:01 2014 +0200
----------------------------------------------------------------------
.../compiler/dag/OptimizerNode.java | 15 +-
.../compiler/dag/PactConnection.java | 4 -
.../CachedMatchStrategyCompilerTest.java | 256 +++++++++++++++++++
.../pact/compiler/CompilerTestBase.java | 30 +++
.../pact/compiler/UnionReplacementTest.java | 4 +-
5 files changed, 292 insertions(+), 17 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/99c888c7/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/dag/OptimizerNode.java
----------------------------------------------------------------------
diff --git a/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/dag/OptimizerNode.java b/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/dag/OptimizerNode.java
index 25f905d..6f295a5 100644
--- a/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/dag/OptimizerNode.java
+++ b/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/dag/OptimizerNode.java
@@ -673,9 +673,6 @@ public abstract class OptimizerNode implements Visitable<OptimizerNode>, Estimat
/**
* Returns the key columns for the specific input, if all keys are preserved
* by this node. Null, otherwise.
- *
- * @param input
- * @return
*/
protected int[] getConstantKeySet(int input) {
Operator<?> contract = getPactContract();
@@ -699,7 +696,6 @@ public abstract class OptimizerNode implements Visitable<OptimizerNode>, Estimat
/**
* An optional method where nodes can describe which fields will be unique in their output.
- * @return
*/
public List<FieldSet> createUniqueFieldsForNode() {
return null;
@@ -707,8 +703,6 @@ public abstract class OptimizerNode implements Visitable<OptimizerNode>, Estimat
/**
* Gets the FieldSets which are unique in the output of the node.
- *
- * @return
*/
public Set<FieldSet> getUniqueFields() {
return this.uniqueFields == null ? Collections.<FieldSet>emptySet() : this.uniqueFields;
@@ -921,10 +915,7 @@ public abstract class OptimizerNode implements Visitable<OptimizerNode>, Estimat
return this.openBranches;
}
- /**
- * @param toParent
- * @return
- */
+
protected List<UnclosedBranchDescriptor> getBranchesForParent(PactConnection toParent) {
if (this.outgoingConnections.size() == 1) {
// return our own stack of open branches, because nothing is added
@@ -1005,8 +996,8 @@ public abstract class OptimizerNode implements Visitable<OptimizerNode>, Estimat
* a) There is no branch in the sub-plan of this node
* b) Both candidates have the same candidate as the child at the last open branch.
*
- * @param subPlan1
- * @param subPlan2
+ * @param plan1
+ * @param plan2
* @return True if the nodes are branch compatible in the inputs.
*/
protected boolean areBranchCompatible(PlanNode plan1, PlanNode plan2) {
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/99c888c7/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/dag/PactConnection.java
----------------------------------------------------------------------
diff --git a/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/dag/PactConnection.java b/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/dag/PactConnection.java
index 3106c50..7461abc 100644
--- a/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/dag/PactConnection.java
+++ b/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/dag/PactConnection.java
@@ -76,10 +76,6 @@ public class PactConnection implements EstimateProvider, DumpableConnection<Opti
*
* @param source
* The source node.
- * @param target
- * The target node.
- * @param shipStrategy
- * The shipping strategy.
*/
public PactConnection(OptimizerNode source) {
if (source == null) {
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/99c888c7/stratosphere-compiler/src/test/java/eu/stratosphere/pact/compiler/CachedMatchStrategyCompilerTest.java
----------------------------------------------------------------------
diff --git a/stratosphere-compiler/src/test/java/eu/stratosphere/pact/compiler/CachedMatchStrategyCompilerTest.java b/stratosphere-compiler/src/test/java/eu/stratosphere/pact/compiler/CachedMatchStrategyCompilerTest.java
new file mode 100644
index 0000000..b82c0f9
--- /dev/null
+++ b/stratosphere-compiler/src/test/java/eu/stratosphere/pact/compiler/CachedMatchStrategyCompilerTest.java
@@ -0,0 +1,256 @@
+/***********************************************************************************************************************
+*
+* Copyright (C) 2013 by the Stratosphere project (http://stratosphere.eu)
+*
+* Licensed 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 eu.stratosphere.pact.compiler;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+import org.junit.Test;
+
+import eu.stratosphere.api.common.Plan;
+import eu.stratosphere.api.common.operators.base.GenericDataSourceBase;
+import eu.stratosphere.api.java.DataSet;
+import eu.stratosphere.api.java.ExecutionEnvironment;
+import eu.stratosphere.api.java.IterativeDataSet;
+import eu.stratosphere.api.java.functions.JoinFunction;
+import eu.stratosphere.api.java.tuple.Tuple3;
+import eu.stratosphere.compiler.PactCompiler;
+import eu.stratosphere.compiler.plan.DualInputPlanNode;
+import eu.stratosphere.compiler.plan.OptimizedPlan;
+import eu.stratosphere.compiler.plantranslate.NepheleJobGraphGenerator;
+import eu.stratosphere.configuration.Configuration;
+import eu.stratosphere.pact.runtime.task.DriverStrategy;
+
+/**
+* Tests that validate optimizer choice when using hash joins inside of iterations
+*/
+@SuppressWarnings("serial")
+public class CachedMatchStrategyCompilerTest extends CompilerTestBase {
+
+ /**
+ * This tests whether a HYBRIDHASH_BUILD_SECOND is correctly transformed to a HYBRIDHASH_BUILD_SECOND_CACHED
+ * when inside of an iteration an on the static path
+ */
+ @Test
+ public void testRightSide() {
+ try {
+
+ Plan plan = getTestPlanRightStatic(PactCompiler.HINT_LOCAL_STRATEGY_HASH_BUILD_SECOND);
+
+ OptimizedPlan oPlan = compileNoStats(plan);
+
+ OptimizerPlanNodeResolver resolver = getOptimizerPlanNodeResolver(oPlan);
+ DualInputPlanNode innerJoin = resolver.getNode("DummyJoiner");
+
+ // verify correct join strategy
+ assertEquals(DriverStrategy.HYBRIDHASH_BUILD_SECOND_CACHED, innerJoin.getDriverStrategy());
+
+ new NepheleJobGraphGenerator().compileJobGraph(oPlan);
+ }
+ catch (Exception e) {
+ System.err.println(e.getMessage());
+ e.printStackTrace();
+ fail("Test errored: " + e.getMessage());
+ }
+ }
+
+ /**
+ * This test makes sure that only a HYBRIDHASH on the static path is transformed to the cached variant
+ */
+ @Test
+ public void testRightSideCountercheck() {
+ try {
+
+ Plan plan = getTestPlanRightStatic(PactCompiler.HINT_LOCAL_STRATEGY_HASH_BUILD_FIRST);
+
+ OptimizedPlan oPlan = compileNoStats(plan);
+
+ OptimizerPlanNodeResolver resolver = getOptimizerPlanNodeResolver(oPlan);
+ DualInputPlanNode innerJoin = resolver.getNode("DummyJoiner");
+
+ // verify correct join strategy
+ assertEquals(DriverStrategy.HYBRIDHASH_BUILD_FIRST, innerJoin.getDriverStrategy());
+
+ new NepheleJobGraphGenerator().compileJobGraph(oPlan);
+ }
+ catch (Exception e) {
+ System.err.println(e.getMessage());
+ e.printStackTrace();
+ fail("Test errored: " + e.getMessage());
+ }
+ }
+
+ /**
+ * This tests whether a HYBRIDHASH_BUILD_FIRST is correctly transformed to a HYBRIDHASH_BUILD_FIRST_CACHED
+ * when inside of an iteration an on the static path
+ */
+ @Test
+ public void testLeftSide() {
+ try {
+
+ Plan plan = getTestPlanLeftStatic(PactCompiler.HINT_LOCAL_STRATEGY_HASH_BUILD_FIRST);
+
+ OptimizedPlan oPlan = compileNoStats(plan);
+
+ OptimizerPlanNodeResolver resolver = getOptimizerPlanNodeResolver(oPlan);
+ DualInputPlanNode innerJoin = resolver.getNode("DummyJoiner");
+
+ // verify correct join strategy
+ assertEquals(DriverStrategy.HYBRIDHASH_BUILD_FIRST_CACHED, innerJoin.getDriverStrategy());
+
+ new NepheleJobGraphGenerator().compileJobGraph(oPlan);
+ }
+ catch (Exception e) {
+ System.err.println(e.getMessage());
+ e.printStackTrace();
+ fail("Test errored: " + e.getMessage());
+ }
+ }
+
+ /**
+ * This test makes sure that only a HYBRIDHASH on the static path is transformed to the cached variant
+ */
+ @Test
+ public void testLeftSideCountercheck() {
+ try {
+
+ Plan plan = getTestPlanLeftStatic(PactCompiler.HINT_LOCAL_STRATEGY_HASH_BUILD_SECOND);
+
+ OptimizedPlan oPlan = compileNoStats(plan);
+
+ OptimizerPlanNodeResolver resolver = getOptimizerPlanNodeResolver(oPlan);
+ DualInputPlanNode innerJoin = resolver.getNode("DummyJoiner");
+
+ // verify correct join strategy
+ assertEquals(DriverStrategy.HYBRIDHASH_BUILD_SECOND, innerJoin.getDriverStrategy());
+
+ new NepheleJobGraphGenerator().compileJobGraph(oPlan);
+ }
+ catch (Exception e) {
+ System.err.println(e.getMessage());
+ e.printStackTrace();
+ fail("Test errored: " + e.getMessage());
+ }
+ }
+
+ /**
+ * This test simulates a join of a big left side with a small right side inside of an iteration, where the small side is on a static path.
+ * Currently the best execution plan is a HYBRIDHASH_BUILD_SECOND_CACHED, where the small side is hashed and cached.
+ * This test also makes sure that all relevant plans are correctly enumerated by the optimizer.
+ */
+ @Test
+ public void testCorrectChoosing() {
+ try {
+
+ Plan plan = getTestPlanRightStatic("");
+
+ SourceCollectorVisitor sourceCollector = new SourceCollectorVisitor();
+ plan.accept(sourceCollector);
+
+ for(GenericDataSourceBase<?, ?> s : sourceCollector.getSources()) {
+ if(s.getName().equals("bigFile")) {
+ this.setSourceStatistics(s, 10000000, 1000);
+ }
+ else if(s.getName().equals("smallFile")) {
+ this.setSourceStatistics(s, 100, 100);
+ }
+ }
+
+ OptimizedPlan oPlan = compileNoStats(plan);
+
+ OptimizerPlanNodeResolver resolver = getOptimizerPlanNodeResolver(oPlan);
+ DualInputPlanNode innerJoin = resolver.getNode("DummyJoiner");
+
+ // verify correct join strategy
+ assertEquals(DriverStrategy.HYBRIDHASH_BUILD_SECOND_CACHED, innerJoin.getDriverStrategy());
+
+ new NepheleJobGraphGenerator().compileJobGraph(oPlan);
+ }
+ catch (Exception e) {
+ System.err.println(e.getMessage());
+ e.printStackTrace();
+ fail("Test errored: " + e.getMessage());
+ }
+ }
+
+ private Plan getTestPlanRightStatic(String strategy) {
+
+ ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+ env.setDegreeOfParallelism(DEFAULT_PARALLELISM);
+
+ DataSet<Tuple3<Long, Long, Long>> bigInput = env.readCsvFile("file://bigFile").types(Long.class, Long.class, Long.class).name("bigFile");
+
+ DataSet<Tuple3<Long, Long, Long>> smallInput = env.readCsvFile("file://smallFile").types(Long.class, Long.class, Long.class).name("smallFile");
+
+ IterativeDataSet<Tuple3<Long, Long, Long>> iteration = bigInput.iterate(10);
+
+ DataSet<Tuple3<Long, Long, Long>> inner;
+
+ if(strategy != "") {
+ Configuration joinStrategy = new Configuration();
+ joinStrategy.setString(PactCompiler.HINT_LOCAL_STRATEGY, strategy);
+ inner = iteration.join(smallInput).where(0).equalTo(0).with(new DummyJoiner()).name("DummyJoiner").withParameters(joinStrategy);
+ }
+ else {
+ inner = iteration.join(smallInput).where(0).equalTo(0).with(new DummyJoiner()).name("DummyJoiner");
+ }
+
+ DataSet<Tuple3<Long, Long, Long>> output = iteration.closeWith(inner);
+
+ output.print();
+
+ return env.createProgramPlan();
+
+ }
+
+ private Plan getTestPlanLeftStatic(String strategy) {
+
+ ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+ env.setDegreeOfParallelism(DEFAULT_PARALLELISM);
+
+ @SuppressWarnings("unchecked")
+ DataSet<Tuple3<Long, Long, Long>> bigInput = env.fromElements(new Tuple3<Long, Long, Long>(1L, 2L, 3L),
+ new Tuple3<Long, Long, Long>(1L, 2L, 3L),new Tuple3<Long, Long, Long>(1L, 2L, 3L)).name("Big");
+
+ @SuppressWarnings("unchecked")
+ DataSet<Tuple3<Long, Long, Long>> smallInput = env.fromElements(new Tuple3<Long, Long, Long>(1L, 2L, 3L)).name("Small");
+
+ IterativeDataSet<Tuple3<Long, Long, Long>> iteration = bigInput.iterate(10);
+
+ Configuration joinStrategy = new Configuration();
+ joinStrategy.setString(PactCompiler.HINT_LOCAL_STRATEGY, strategy);
+
+ DataSet<Tuple3<Long, Long, Long>> inner = smallInput.join(iteration).where(0).equalTo(0).with(new DummyJoiner()).name("DummyJoiner").withParameters(joinStrategy);
+
+ DataSet<Tuple3<Long, Long, Long>> output = iteration.closeWith(inner);
+
+ output.print();
+
+ return env.createProgramPlan();
+
+ }
+
+ private static class DummyJoiner extends JoinFunction<Tuple3<Long, Long, Long>, Tuple3<Long, Long, Long>, Tuple3<Long, Long, Long>> {
+
+ @Override
+ public Tuple3<Long, Long, Long> join(Tuple3<Long, Long, Long> first,
+ Tuple3<Long, Long, Long> second) throws Exception {
+
+ return first;
+ }
+ }
+}
+
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/99c888c7/stratosphere-compiler/src/test/java/eu/stratosphere/pact/compiler/CompilerTestBase.java
----------------------------------------------------------------------
diff --git a/stratosphere-compiler/src/test/java/eu/stratosphere/pact/compiler/CompilerTestBase.java b/stratosphere-compiler/src/test/java/eu/stratosphere/pact/compiler/CompilerTestBase.java
index f534ad9..24afb2f 100644
--- a/stratosphere-compiler/src/test/java/eu/stratosphere/pact/compiler/CompilerTestBase.java
+++ b/stratosphere-compiler/src/test/java/eu/stratosphere/pact/compiler/CompilerTestBase.java
@@ -299,4 +299,34 @@ public abstract class CompilerTestBase implements java.io.Serializable {
@Override
public void postVisit(Operator<?> visitable) {}
}
+
+ /**
+ * Collects all DataSources of a plan to add statistics
+ *
+ */
+ public static class SourceCollectorVisitor implements Visitor<Operator<?>> {
+
+ protected final List<GenericDataSourceBase<?, ?>> sources = new ArrayList<GenericDataSourceBase<?, ?>>(4);
+
+ @Override
+ public boolean preVisit(Operator<?> visitable) {
+
+ if(visitable instanceof GenericDataSourceBase) {
+ sources.add((GenericDataSourceBase<?, ?>) visitable);
+ }
+ else if(visitable instanceof BulkIteration) {
+ ((BulkIteration) visitable).getNextPartialSolution().accept(this);
+ }
+
+ return true;
+ }
+
+ @Override
+ public void postVisit(Operator<?> visitable) {}
+
+ public List<GenericDataSourceBase<?, ?>> getSources() {
+ return this.sources;
+ }
+
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/99c888c7/stratosphere-compiler/src/test/java/eu/stratosphere/pact/compiler/UnionReplacementTest.java
----------------------------------------------------------------------
diff --git a/stratosphere-compiler/src/test/java/eu/stratosphere/pact/compiler/UnionReplacementTest.java b/stratosphere-compiler/src/test/java/eu/stratosphere/pact/compiler/UnionReplacementTest.java
index 9ef8519..6de9e60 100644
--- a/stratosphere-compiler/src/test/java/eu/stratosphere/pact/compiler/UnionReplacementTest.java
+++ b/stratosphere-compiler/src/test/java/eu/stratosphere/pact/compiler/UnionReplacementTest.java
@@ -16,13 +16,15 @@ package eu.stratosphere.pact.compiler;
import eu.stratosphere.api.common.Plan;
import eu.stratosphere.api.java.DataSet;
import eu.stratosphere.api.java.ExecutionEnvironment;
-import eu.stratosphere.api.java.IterativeDataSet;
import eu.stratosphere.compiler.CompilerException;
import eu.stratosphere.compiler.plan.OptimizedPlan;
import eu.stratosphere.compiler.plantranslate.NepheleJobGraphGenerator;
+
import org.junit.Test;
+
import static org.junit.Assert.fail;
+@SuppressWarnings("serial")
public class UnionReplacementTest extends CompilerTestBase {
@Test
[09/12] git commit: Fix missing strategy constants in JSON dump
generator
Posted by se...@apache.org.
Fix missing strategy constants in JSON dump generator
Project: http://git-wip-us.apache.org/repos/asf/incubator-flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-flink/commit/a9b1daa1
Tree: http://git-wip-us.apache.org/repos/asf/incubator-flink/tree/a9b1daa1
Diff: http://git-wip-us.apache.org/repos/asf/incubator-flink/diff/a9b1daa1
Branch: refs/heads/master
Commit: a9b1daa125f109280027f9df3029bb4c49a0f326
Parents: 03c6160
Author: Stephan Ewen <se...@apache.org>
Authored: Wed Jul 9 03:07:21 2014 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Wed Jul 9 03:10:18 2014 +0200
----------------------------------------------------------------------
.../stratosphere/compiler/plandump/PlanJSONDumpGenerator.java | 7 +++++++
1 file changed, 7 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/a9b1daa1/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/plandump/PlanJSONDumpGenerator.java
----------------------------------------------------------------------
diff --git a/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/plandump/PlanJSONDumpGenerator.java b/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/plandump/PlanJSONDumpGenerator.java
index a1baff1..849da33 100644
--- a/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/plandump/PlanJSONDumpGenerator.java
+++ b/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/plandump/PlanJSONDumpGenerator.java
@@ -436,6 +436,13 @@ public class PlanJSONDumpGenerator {
case HYBRIDHASH_BUILD_SECOND:
locString = "Hybrid Hash (build: " + child2name + ")";
break;
+
+ case HYBRIDHASH_BUILD_FIRST_CACHED:
+ locString = "Hybrid Hash (CACHED) (build: " + child1name + ")";
+ break;
+ case HYBRIDHASH_BUILD_SECOND_CACHED:
+ locString = "Hybrid Hash (CACHED) (build: " + child2name + ")";
+ break;
case NESTEDLOOP_BLOCKED_OUTER_FIRST:
locString = "Nested Loops (Blocked Outer: " + child1name + ")";
[05/12] git commit: [FLINK-701] Rework of TypeExtractor for SAM
support
Posted by se...@apache.org.
[FLINK-701] Rework of TypeExtractor for SAM support
This closes #52
Project: http://git-wip-us.apache.org/repos/asf/incubator-flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-flink/commit/5cec8e8a
Tree: http://git-wip-us.apache.org/repos/asf/incubator-flink/tree/5cec8e8a
Diff: http://git-wip-us.apache.org/repos/asf/incubator-flink/diff/5cec8e8a
Branch: refs/heads/master
Commit: 5cec8e8a7702e3a1515f406489087b55b9e9f902
Parents: 8ea6536
Author: twalthr <in...@twalthr.com>
Authored: Mon Jun 30 12:02:37 2014 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Tue Jul 8 16:55:13 2014 +0200
----------------------------------------------------------------------
.../api/java/typeutils/TypeExtractor.java | 618 ++++++++++---------
.../java/type/extractor/TypeExtractorTest.java | 39 +-
2 files changed, 364 insertions(+), 293 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/5cec8e8a/stratosphere-java/src/main/java/eu/stratosphere/api/java/typeutils/TypeExtractor.java
----------------------------------------------------------------------
diff --git a/stratosphere-java/src/main/java/eu/stratosphere/api/java/typeutils/TypeExtractor.java b/stratosphere-java/src/main/java/eu/stratosphere/api/java/typeutils/TypeExtractor.java
index 7b2a0c7..ada6535 100644
--- a/stratosphere-java/src/main/java/eu/stratosphere/api/java/typeutils/TypeExtractor.java
+++ b/stratosphere-java/src/main/java/eu/stratosphere/api/java/typeutils/TypeExtractor.java
@@ -26,21 +26,20 @@ import java.util.HashSet;
import java.util.List;
import java.util.Set;
-import eu.stratosphere.types.TypeInformation;
-
import org.apache.commons.lang3.Validate;
import org.apache.hadoop.io.Writable;
+import eu.stratosphere.api.common.functions.GenericCoGrouper;
+import eu.stratosphere.api.common.functions.GenericCrosser;
+import eu.stratosphere.api.common.functions.GenericFlatMap;
+import eu.stratosphere.api.common.functions.GenericGroupReduce;
+import eu.stratosphere.api.common.functions.GenericJoiner;
+import eu.stratosphere.api.common.functions.GenericMap;
import eu.stratosphere.api.common.io.InputFormat;
-import eu.stratosphere.api.java.functions.CoGroupFunction;
-import eu.stratosphere.api.java.functions.CrossFunction;
-import eu.stratosphere.api.java.functions.FlatMapFunction;
-import eu.stratosphere.api.java.functions.GroupReduceFunction;
import eu.stratosphere.api.java.functions.InvalidTypesException;
-import eu.stratosphere.api.java.functions.JoinFunction;
import eu.stratosphere.api.java.functions.KeySelector;
-import eu.stratosphere.api.java.functions.MapFunction;
import eu.stratosphere.api.java.tuple.Tuple;
+import eu.stratosphere.types.TypeInformation;
import eu.stratosphere.types.Value;
public class TypeExtractor {
@@ -53,103 +52,107 @@ public class TypeExtractor {
alreadySeen = new HashSet<Class<?>>();
}
+ // --------------------------------------------------------------------------------------------
+ // Function specific methods
+ // --------------------------------------------------------------------------------------------
+
@SuppressWarnings("unchecked")
- public static <IN, OUT> TypeInformation<OUT> getMapReturnTypes(MapFunction<IN, OUT> mapFunction, TypeInformation<IN> inType) {
- validateInputType(MapFunction.class, mapFunction.getClass(), 0, inType);
- if(mapFunction instanceof ResultTypeQueryable) {
- return ((ResultTypeQueryable<OUT>) mapFunction).getProducedType();
+ public static <IN, OUT> TypeInformation<OUT> getMapReturnTypes(GenericMap<IN, OUT> mapInterface, TypeInformation<IN> inType) {
+ validateInputType(GenericMap.class, mapInterface.getClass(), 0, inType);
+ if(mapInterface instanceof ResultTypeQueryable) {
+ return ((ResultTypeQueryable<OUT>) mapInterface).getProducedType();
}
- return new TypeExtractor().privateCreateTypeInfo(MapFunction.class, mapFunction.getClass(), 1, inType, null);
+ return new TypeExtractor().privateCreateTypeInfo(GenericMap.class, mapInterface.getClass(), 1, inType, null);
}
@SuppressWarnings("unchecked")
- public static <IN, OUT> TypeInformation<OUT> getFlatMapReturnTypes(FlatMapFunction<IN, OUT> flatMapFunction, TypeInformation<IN> inType) {
- validateInputType(FlatMapFunction.class, flatMapFunction.getClass(), 0, inType);
- if(flatMapFunction instanceof ResultTypeQueryable) {
- return ((ResultTypeQueryable<OUT>) flatMapFunction).getProducedType();
+ public static <IN, OUT> TypeInformation<OUT> getFlatMapReturnTypes(GenericFlatMap<IN, OUT> flatMapInterface, TypeInformation<IN> inType) {
+ validateInputType(GenericFlatMap.class, flatMapInterface.getClass(), 0, inType);
+ if(flatMapInterface instanceof ResultTypeQueryable) {
+ return ((ResultTypeQueryable<OUT>) flatMapInterface).getProducedType();
}
- return new TypeExtractor().privateCreateTypeInfo(FlatMapFunction.class, flatMapFunction.getClass(), 1, inType, null);
+ return new TypeExtractor().privateCreateTypeInfo(GenericFlatMap.class, flatMapInterface.getClass(), 1, inType, null);
}
@SuppressWarnings("unchecked")
- public static <IN, OUT> TypeInformation<OUT> getGroupReduceReturnTypes(GroupReduceFunction<IN, OUT> groupReduceFunction,
+ public static <IN, OUT> TypeInformation<OUT> getGroupReduceReturnTypes(GenericGroupReduce<IN, OUT> groupReduceInterface,
TypeInformation<IN> inType) {
- validateInputType(GroupReduceFunction.class, groupReduceFunction.getClass(), 0, inType);
- if(groupReduceFunction instanceof ResultTypeQueryable) {
- return ((ResultTypeQueryable<OUT>) groupReduceFunction).getProducedType();
+ validateInputType(GenericGroupReduce.class, groupReduceInterface.getClass(), 0, inType);
+ if(groupReduceInterface instanceof ResultTypeQueryable) {
+ return ((ResultTypeQueryable<OUT>) groupReduceInterface).getProducedType();
}
- return new TypeExtractor().privateCreateTypeInfo(GroupReduceFunction.class, groupReduceFunction.getClass(), 1, inType, null);
+ return new TypeExtractor().privateCreateTypeInfo(GenericGroupReduce.class, groupReduceInterface.getClass(), 1, inType, null);
}
@SuppressWarnings("unchecked")
- public static <IN1, IN2, OUT> TypeInformation<OUT> getJoinReturnTypes(JoinFunction<IN1, IN2, OUT> joinFunction,
+ public static <IN1, IN2, OUT> TypeInformation<OUT> getJoinReturnTypes(GenericJoiner<IN1, IN2, OUT> joinInterface,
TypeInformation<IN1> in1Type, TypeInformation<IN2> in2Type) {
- validateInputType(JoinFunction.class, joinFunction.getClass(), 0, in1Type);
- validateInputType(JoinFunction.class, joinFunction.getClass(), 1, in2Type);
- if(joinFunction instanceof ResultTypeQueryable) {
- return ((ResultTypeQueryable<OUT>) joinFunction).getProducedType();
+ validateInputType(GenericJoiner.class, joinInterface.getClass(), 0, in1Type);
+ validateInputType(GenericJoiner.class, joinInterface.getClass(), 1, in2Type);
+ if(joinInterface instanceof ResultTypeQueryable) {
+ return ((ResultTypeQueryable<OUT>) joinInterface).getProducedType();
}
- return new TypeExtractor().privateCreateTypeInfo(JoinFunction.class, joinFunction.getClass(), 2, in1Type, in2Type);
+ return new TypeExtractor().privateCreateTypeInfo(GenericJoiner.class, joinInterface.getClass(), 2, in1Type, in2Type);
}
@SuppressWarnings("unchecked")
- public static <IN1, IN2, OUT> TypeInformation<OUT> getCoGroupReturnTypes(CoGroupFunction<IN1, IN2, OUT> coGroupFunction,
+ public static <IN1, IN2, OUT> TypeInformation<OUT> getCoGroupReturnTypes(GenericCoGrouper<IN1, IN2, OUT> coGroupInterface,
TypeInformation<IN1> in1Type, TypeInformation<IN2> in2Type) {
- validateInputType(CoGroupFunction.class, coGroupFunction.getClass(), 0, in1Type);
- validateInputType(CoGroupFunction.class, coGroupFunction.getClass(), 1, in2Type);
- if(coGroupFunction instanceof ResultTypeQueryable) {
- return ((ResultTypeQueryable<OUT>) coGroupFunction).getProducedType();
+ validateInputType(GenericCoGrouper.class, coGroupInterface.getClass(), 0, in1Type);
+ validateInputType(GenericCoGrouper.class, coGroupInterface.getClass(), 1, in2Type);
+ if(coGroupInterface instanceof ResultTypeQueryable) {
+ return ((ResultTypeQueryable<OUT>) coGroupInterface).getProducedType();
}
- return new TypeExtractor().privateCreateTypeInfo(CoGroupFunction.class, coGroupFunction.getClass(), 2, in1Type, in2Type);
+ return new TypeExtractor().privateCreateTypeInfo(GenericCoGrouper.class, coGroupInterface.getClass(), 2, in1Type, in2Type);
}
@SuppressWarnings("unchecked")
- public static <IN1, IN2, OUT> TypeInformation<OUT> getCrossReturnTypes(CrossFunction<IN1, IN2, OUT> crossFunction,
+ public static <IN1, IN2, OUT> TypeInformation<OUT> getCrossReturnTypes(GenericCrosser<IN1, IN2, OUT> crossInterface,
TypeInformation<IN1> in1Type, TypeInformation<IN2> in2Type) {
- validateInputType(CrossFunction.class, crossFunction.getClass(), 0, in1Type);
- validateInputType(CrossFunction.class, crossFunction.getClass(), 1, in2Type);
- if(crossFunction instanceof ResultTypeQueryable) {
- return ((ResultTypeQueryable<OUT>) crossFunction).getProducedType();
+ validateInputType(GenericCrosser.class, crossInterface.getClass(), 0, in1Type);
+ validateInputType(GenericCrosser.class, crossInterface.getClass(), 1, in2Type);
+ if(crossInterface instanceof ResultTypeQueryable) {
+ return ((ResultTypeQueryable<OUT>) crossInterface).getProducedType();
}
- return new TypeExtractor().privateCreateTypeInfo(CrossFunction.class, crossFunction.getClass(), 2, in1Type, in2Type);
+ return new TypeExtractor().privateCreateTypeInfo(GenericCrosser.class, crossInterface.getClass(), 2, in1Type, in2Type);
}
@SuppressWarnings("unchecked")
- public static <IN, OUT> TypeInformation<OUT> getKeySelectorTypes(KeySelector<IN, OUT> selector, TypeInformation<IN> inType) {
- validateInputType(KeySelector.class, selector.getClass(), 0, inType);
- if(selector instanceof ResultTypeQueryable) {
- return ((ResultTypeQueryable<OUT>) selector).getProducedType();
+ public static <IN, OUT> TypeInformation<OUT> getKeySelectorTypes(KeySelector<IN, OUT> selectorInterface, TypeInformation<IN> inType) {
+ validateInputType(KeySelector.class, selectorInterface.getClass(), 0, inType);
+ if(selectorInterface instanceof ResultTypeQueryable) {
+ return ((ResultTypeQueryable<OUT>) selectorInterface).getProducedType();
}
- return new TypeExtractor().privateCreateTypeInfo(KeySelector.class, selector.getClass(), 1, inType, null);
+ return new TypeExtractor().privateCreateTypeInfo(KeySelector.class, selectorInterface.getClass(), 1, inType, null);
}
@SuppressWarnings("unchecked")
- public static <IN> TypeInformation<IN> getInputFormatTypes(InputFormat<IN, ?> inputFormat) {
- if(inputFormat instanceof ResultTypeQueryable) {
- return ((ResultTypeQueryable<IN>) inputFormat).getProducedType();
+ public static <IN> TypeInformation<IN> getInputFormatTypes(InputFormat<IN, ?> inputFormatInterface) {
+ if(inputFormatInterface instanceof ResultTypeQueryable) {
+ return ((ResultTypeQueryable<IN>) inputFormatInterface).getProducedType();
}
- return new TypeExtractor().privateCreateTypeInfo(InputFormat.class, inputFormat.getClass(), 0, null, null);
+ return new TypeExtractor().privateCreateTypeInfo(InputFormat.class, inputFormatInterface.getClass(), 0, null, null);
}
// --------------------------------------------------------------------------------------------
- // Generic utility methods
+ // Create type information
// --------------------------------------------------------------------------------------------
public static TypeInformation<?> createTypeInfo(Type t) {
return new TypeExtractor().privateCreateTypeInfo(t);
}
-
+
private TypeInformation<?> privateCreateTypeInfo(Type t) {
ArrayList<Type> typeHierarchy = new ArrayList<Type>();
typeHierarchy.add(t);
return createTypeInfoWithTypeHierarchy(typeHierarchy, t, null, null);
}
-
+
public static <IN1, IN2, OUT> TypeInformation<OUT> createTypeInfo(Class<?> baseClass, Class<?> clazz, int returnParamPos,
TypeInformation<IN1> in1Type, TypeInformation<IN2> in2Type) {
return new TypeExtractor().privateCreateTypeInfo(baseClass, clazz, returnParamPos, in1Type, in2Type);
}
-
+
@SuppressWarnings("unchecked")
private <IN1, IN2, OUT> TypeInformation<OUT> privateCreateTypeInfo(Class<?> baseClass, Class<?> clazz, int returnParamPos,
TypeInformation<IN1> in1Type, TypeInformation<IN2> in2Type) {
@@ -158,11 +161,9 @@ public class TypeExtractor {
TypeInformation<OUT> typeInfo = null;
- // return type is a variable -> try to get the type info from the input of the base class directly
+ // return type is a variable -> try to get the type info from the input directly
if (returnType instanceof TypeVariable<?>) {
- ParameterizedType immediateBaseChild = (ParameterizedType) typeHierarchy.get(typeHierarchy.size() - 1);
- typeInfo = (TypeInformation<OUT>) createTypeInfoWithImmediateBaseChildInput(immediateBaseChild, (TypeVariable<?>) returnType,
- in1Type, in2Type);
+ typeInfo = (TypeInformation<OUT>) createTypeInfoFromInput((TypeVariable<?>) returnType, typeHierarchy, in1Type, in2Type);
if (typeInfo != null) {
return typeInfo;
@@ -173,10 +174,255 @@ public class TypeExtractor {
return (TypeInformation<OUT>) createTypeInfoWithTypeHierarchy(typeHierarchy, returnType, in1Type, in2Type);
}
+ @SuppressWarnings({ "unchecked", "rawtypes" })
+ private <IN1, IN2, OUT> TypeInformation<OUT> createTypeInfoWithTypeHierarchy(ArrayList<Type> typeHierarchy, Type t,
+ TypeInformation<IN1> in1Type, TypeInformation<IN2> in2Type) {
+
+ // check if type is a subclass of tuple
+ if ((t instanceof Class<?> && Tuple.class.isAssignableFrom((Class<?>) t))
+ || (t instanceof ParameterizedType && Tuple.class.isAssignableFrom((Class<?>) ((ParameterizedType) t).getRawType()))) {
+
+ Type curT = t;
+
+ // do not allow usage of Tuple as type
+ if (curT instanceof Class<?> && ((Class<?>) curT).equals(Tuple.class)) {
+ throw new InvalidTypesException(
+ "Usage of class Tuple as a type is not allowed. Use a concrete subclass (e.g. Tuple1, Tuple2, etc.) instead.");
+ }
+
+ // go up the hierarchy until we reach immediate child of Tuple (with or without generics)
+ // collect the types while moving up for a later top-down
+ while (!(curT instanceof ParameterizedType && ((Class<?>) ((ParameterizedType) curT).getRawType()).getSuperclass().equals(
+ Tuple.class))
+ && !(curT instanceof Class<?> && ((Class<?>) curT).getSuperclass().equals(Tuple.class))) {
+ typeHierarchy.add(curT);
+
+ // parameterized type
+ if (curT instanceof ParameterizedType) {
+ curT = ((Class<?>) ((ParameterizedType) curT).getRawType()).getGenericSuperclass();
+ }
+ // class
+ else {
+ curT = ((Class<?>) curT).getGenericSuperclass();
+ }
+ }
+
+ // check if immediate child of Tuple has generics
+ if (curT instanceof Class<?>) {
+ throw new InvalidTypesException("Tuple needs to be parameterized by using generics.");
+ }
+
+ ParameterizedType tupleChild = (ParameterizedType) curT;
+
+ Type[] subtypes = new Type[tupleChild.getActualTypeArguments().length];
+
+ // materialize possible type variables
+ for (int i = 0; i < subtypes.length; i++) {
+ // materialize immediate TypeVariables
+ if (tupleChild.getActualTypeArguments()[i] instanceof TypeVariable<?>) {
+ subtypes[i] = materializeTypeVariable(typeHierarchy, (TypeVariable<?>) tupleChild.getActualTypeArguments()[i]);
+ }
+ // class or parameterized type
+ else {
+ subtypes[i] = tupleChild.getActualTypeArguments()[i];
+ }
+ }
+
+ TypeInformation<?>[] tupleSubTypes = new TypeInformation<?>[subtypes.length];
+ for (int i = 0; i < subtypes.length; i++) {
+ // sub type could not be determined with materializing
+ // try to derive the type info of the TypeVariable from the immediate base child input as a last attempt
+ if (subtypes[i] instanceof TypeVariable<?>) {
+ tupleSubTypes[i] = createTypeInfoFromInput((TypeVariable<?>) subtypes[i], typeHierarchy, in1Type, in2Type);
+
+ // variable could not be determined
+ if (tupleSubTypes[i] == null) {
+ throw new InvalidTypesException("Type of TypeVariable '" + ((TypeVariable<?>) subtypes[i]).getName() + "' in '"
+ + ((TypeVariable<?>) subtypes[i]).getGenericDeclaration()
+ + "' could not be determined. This is most likely a type erasure problem. "
+ + "The type extraction currently supports types with generic variables only in cases where "
+ + "all variables in the return type can be deduced from the input type(s).");
+ }
+ } else {
+ tupleSubTypes[i] = createTypeInfoWithTypeHierarchy(new ArrayList<Type>(typeHierarchy), subtypes[i], in1Type, in2Type);
+ }
+ }
+
+ // TODO: Check that type that extends Tuple does not have additional fields.
+ // Right now, these fields are not be serialized by the TupleSerializer.
+ // We might want to add an ExtendedTupleSerializer for that.
+
+ if (t instanceof Class<?>) {
+ return new TupleTypeInfo(((Class<? extends Tuple>) t), tupleSubTypes);
+ } else if (t instanceof ParameterizedType) {
+ return new TupleTypeInfo(((Class<? extends Tuple>) ((ParameterizedType) t).getRawType()), tupleSubTypes);
+ }
+ }
+ // type depends on another type
+ // e.g. class MyMapper<E> extends MapFunction<String, E>
+ else if (t instanceof TypeVariable) {
+ Type typeVar = materializeTypeVariable(typeHierarchy, (TypeVariable<?>) t);
+
+ if (!(typeVar instanceof TypeVariable)) {
+ return createTypeInfoWithTypeHierarchy(typeHierarchy, typeVar, in1Type, in2Type);
+ }
+ // try to derive the type info of the TypeVariable from the immediate base child input as a last attempt
+ else {
+ TypeInformation<OUT> typeInfo = (TypeInformation<OUT>) createTypeInfoFromInput((TypeVariable<?>) t, typeHierarchy, in1Type, in2Type);
+ if (typeInfo != null) {
+ return typeInfo;
+ } else {
+ throw new InvalidTypesException("Type of TypeVariable '" + ((TypeVariable<?>) t).getName() + "' in '"
+ + ((TypeVariable<?>) t).getGenericDeclaration() + "' could not be determined. This is most likely a type erasure problem. "
+ + "The type extraction currently supports types with generic variables only in cases where "
+ + "all variables in the return type can be deduced from the input type(s).");
+ }
+ }
+ }
+ // arrays with generics
+ else if (t instanceof GenericArrayType) {
+ GenericArrayType genericArray = (GenericArrayType) t;
+
+ Type componentType = genericArray.getGenericComponentType();
+
+ // due to a Java 6 bug, it is possible that the JVM classifies e.g. String[] or int[] as GenericArrayType instead of Class
+ if (componentType instanceof Class) {
+
+ Class<?> componentClass = (Class<?>) componentType;
+ String className;
+ // for int[], double[] etc.
+ if(componentClass.isPrimitive()) {
+ className = encodePrimitiveClass(componentClass);
+ }
+ // for String[], Integer[] etc.
+ else {
+ className = "L" + componentClass.getName() + ";";
+ }
+
+ Class<OUT> classArray = null;
+ try {
+ classArray = (Class<OUT>) Class.forName("[" + className);
+ } catch (ClassNotFoundException e) {
+ throw new InvalidTypesException("Could not convert GenericArrayType to Class.");
+ }
+ return getForClass(classArray);
+ }
+
+ TypeInformation<?> componentInfo = createTypeInfoWithTypeHierarchy(typeHierarchy, genericArray.getGenericComponentType(),
+ in1Type, in2Type);
+ return ObjectArrayTypeInfo.getInfoFor(t, componentInfo);
+ }
+ // objects with generics are treated as raw type
+ else if (t instanceof ParameterizedType) {
+ return privateGetForClass((Class<OUT>) ((ParameterizedType) t).getRawType());
+ }
+ // no tuple, no TypeVariable, no generic type
+ else if (t instanceof Class) {
+ return privateGetForClass((Class<OUT>) t);
+ }
+
+ throw new InvalidTypesException("Type Information could not be created.");
+ }
+
+ private <IN1, IN2> TypeInformation<?> createTypeInfoFromInput(TypeVariable<?> returnTypeVar, ArrayList<Type> returnTypeHierarchy,
+ TypeInformation<IN1> in1TypeInfo, TypeInformation<IN2> in2TypeInfo) {
+
+ Type matReturnTypeVar = materializeTypeVariable(returnTypeHierarchy, returnTypeVar);
+
+ // variable could be resolved
+ if (!(matReturnTypeVar instanceof TypeVariable)) {
+ return createTypeInfoWithTypeHierarchy(returnTypeHierarchy, matReturnTypeVar, in1TypeInfo, in2TypeInfo);
+ }
+ else {
+ returnTypeVar = (TypeVariable<?>) matReturnTypeVar;
+ }
+
+ TypeInformation<?> info = null;
+ if (in1TypeInfo != null) {
+ // find the deepest type variable that describes the type of input 1
+ ParameterizedType baseClass = (ParameterizedType) returnTypeHierarchy.get(returnTypeHierarchy.size() - 1 );
+ Type in1Type = baseClass.getActualTypeArguments()[0];
+ if (in1Type instanceof TypeVariable) {
+ in1Type = materializeTypeVariable(returnTypeHierarchy, (TypeVariable<?>) in1Type);
+ info = findCorrespondingInfo(returnTypeVar, in1Type, in1TypeInfo);
+ }
+ }
+
+ if (info == null && in2TypeInfo != null) {
+ // find the deepest type variable that describes the type of input 2
+ ParameterizedType baseClass = (ParameterizedType) returnTypeHierarchy.get(returnTypeHierarchy.size() - 1 );
+ Type in2Type = baseClass.getActualTypeArguments()[1];
+ if (in2Type instanceof TypeVariable) {
+ in2Type = materializeTypeVariable(returnTypeHierarchy, (TypeVariable<?>) in2Type);
+ info = findCorrespondingInfo(returnTypeVar, in2Type, in2TypeInfo);
+ }
+ }
+
+ if (info != null) {
+ return info;
+ }
+
+ return null;
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Extract type parameters
+ // --------------------------------------------------------------------------------------------
+
public static Type getParameterType(Class<?> baseClass, Class<?> clazz, int pos) {
return getParameterType(baseClass, null, clazz, pos);
}
+ private static Type getParameterType(Class<?> baseClass, ArrayList<Type> typeHierarchy, Class<?> clazz, int pos) {
+ Type[] interfaceTypes = clazz.getGenericInterfaces();
+
+ // search in interfaces for base class
+ for (Type t : interfaceTypes) {
+ Type parameter = getParameterTypeFromGenericType(baseClass, typeHierarchy, t, pos);
+ if (parameter != null) {
+ return parameter;
+ }
+ }
+
+ // search in superclass for base class
+ Type t = clazz.getGenericSuperclass();
+ Type parameter = getParameterTypeFromGenericType(baseClass, typeHierarchy, t, pos);
+ if (parameter != null) {
+ return parameter;
+ }
+
+ throw new IllegalArgumentException(baseClass.getName() + " must be implemented.");
+ }
+
+ private static Type getParameterTypeFromGenericType(Class<?> baseClass, ArrayList<Type> typeHierarchy, Type t, int pos) {
+ // base class
+ if (t instanceof ParameterizedType && baseClass.equals((Class<?>) ((ParameterizedType) t).getRawType())) {
+ if (typeHierarchy != null) {
+ typeHierarchy.add(t);
+ }
+ ParameterizedType baseClassChild = (ParameterizedType) t;
+ return baseClassChild.getActualTypeArguments()[pos];
+ }
+ // interface that extended base class as class or parameterized type
+ else if (t instanceof ParameterizedType && baseClass.isAssignableFrom((Class<?>) ((ParameterizedType) t).getRawType())) {
+ if (typeHierarchy != null) {
+ typeHierarchy.add(t);
+ }
+ return getParameterType(baseClass, typeHierarchy, (Class<?>) ((ParameterizedType) t).getRawType(), pos);
+ }
+ else if (t instanceof Class<?> && baseClass.isAssignableFrom((Class<?>) t)) {
+ if (typeHierarchy != null) {
+ typeHierarchy.add(t);
+ }
+ return getParameterType(baseClass, typeHierarchy, (Class<?>) t, pos);
+ }
+ return null;
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Validate input
+ // --------------------------------------------------------------------------------------------
+
private static void validateInputType(Class<?> baseClass, Class<?> clazz, int inputParamPos, TypeInformation<?> inType) {
ArrayList<Type> typeHierarchy = new ArrayList<Type>();
try {
@@ -286,7 +532,7 @@ public class TypeExtractor {
if (component instanceof TypeVariable<?>) {
component = materializeTypeVariable(typeHierarchy, (TypeVariable<?>) component);
- if (component == null) {
+ if (component instanceof TypeVariable) {
return;
}
}
@@ -311,7 +557,7 @@ public class TypeExtractor {
if (component instanceof TypeVariable<?>) {
component = materializeTypeVariable(typeHierarchy, (TypeVariable<?>) component);
- if (component == null) {
+ if (component instanceof TypeVariable) {
return;
}
}
@@ -331,7 +577,18 @@ public class TypeExtractor {
throw new InvalidTypesException("Value type '" + typeInfo + "' expected but was '" + actual + "'.");
}
}
- // check for custom object
+ // check for POJO
+ else if (typeInfo instanceof PojoTypeInfo) {
+ Class<?> clazz = null;
+ if (!(type instanceof Class<?> && ((PojoTypeInfo<?>) typeInfo).getTypeClass() == (clazz = (Class<?>) type))
+ && !(type instanceof ParameterizedType && (clazz = (Class<?>) ((ParameterizedType) type).getRawType()) == ((PojoTypeInfo<?>) typeInfo)
+ .getTypeClass())) {
+ throw new InvalidTypesException("POJO type '"
+ + ((PojoTypeInfo<?>) typeInfo).getTypeClass().getCanonicalName() + "' expected but was '"
+ + clazz.getCanonicalName() + "'.");
+ }
+ }
+ // check for generic object
else if (typeInfo instanceof GenericTypeInfo<?>) {
Class<?> clazz = null;
if (!(type instanceof Class<?> && ((GenericTypeInfo<?>) typeInfo).getTypeClass() == (clazz = (Class<?>) type))
@@ -344,218 +601,15 @@ public class TypeExtractor {
}
} else {
type = materializeTypeVariable(typeHierarchy, (TypeVariable<?>) type);
- if (type != null) {
+ if (!(type instanceof TypeVariable)) {
validateInfo(typeHierarchy, type, typeInfo);
}
}
}
- private static Type getParameterType(Class<?> baseClass, ArrayList<Type> typeHierarchy, Class<?> clazz, int pos) {
- Type t = clazz.getGenericSuperclass();
-
- // check if type is child of the base class
- if (!(t instanceof Class<?> && baseClass.isAssignableFrom((Class<?>) t))
- && !(t instanceof ParameterizedType && baseClass.isAssignableFrom((Class<?>) ((ParameterizedType) t).getRawType()))) {
- throw new IllegalArgumentException("A generic function base class must be a super class.");
- }
- if (typeHierarchy != null) {
- typeHierarchy.add(t);
- }
-
- Type curT = t;
- // go up the hierarchy until we reach the base class (with or without generics)
- // collect the types while moving up for a later top-down
- while (!(curT instanceof ParameterizedType && ((Class<?>) ((ParameterizedType) curT).getRawType()).equals(baseClass))
- && !(curT instanceof Class<?> && ((Class<?>) curT).equals(baseClass))) {
- if (typeHierarchy != null) {
- typeHierarchy.add(curT);
- }
-
- // parameterized type
- if (curT instanceof ParameterizedType) {
- curT = ((Class<?>) ((ParameterizedType) curT).getRawType()).getGenericSuperclass();
- }
- // class
- else {
- curT = ((Class<?>) curT).getGenericSuperclass();
- }
- }
-
- // check if immediate child of base class has generics
- if (curT instanceof Class<?>) {
- throw new InvalidTypesException("Function needs to be parameterized by using generics.");
- }
-
- if (typeHierarchy != null) {
- typeHierarchy.add(curT);
- }
-
- ParameterizedType baseClassChild = (ParameterizedType) curT;
-
- return baseClassChild.getActualTypeArguments()[pos];
- }
-
- @SuppressWarnings({ "unchecked", "rawtypes" })
- private <IN1, IN2, OUT> TypeInformation<OUT> createTypeInfoWithTypeHierarchy(ArrayList<Type> typeHierarchy, Type t,
- TypeInformation<IN1> in1Type, TypeInformation<IN2> in2Type) {
-
- // check if type is a subclass of tuple
- if ((t instanceof Class<?> && Tuple.class.isAssignableFrom((Class<?>) t))
- || (t instanceof ParameterizedType && Tuple.class.isAssignableFrom((Class<?>) ((ParameterizedType) t).getRawType()))) {
-
- Type curT = t;
-
- // do not allow usage of Tuple as type
- if (curT instanceof Class<?> && ((Class<?>) curT).equals(Tuple.class)) {
- throw new InvalidTypesException(
- "Usage of class Tuple as a type is not allowed. Use a concrete subclass (e.g. Tuple1, Tuple2, etc.) instead.");
- }
-
- // go up the hierarchy until we reach immediate child of Tuple (with or without generics)
- // collect the types while moving up for a later top-down
- while (!(curT instanceof ParameterizedType && ((Class<?>) ((ParameterizedType) curT).getRawType()).getSuperclass().equals(
- Tuple.class))
- && !(curT instanceof Class<?> && ((Class<?>) curT).getSuperclass().equals(Tuple.class))) {
- typeHierarchy.add(curT);
-
- // parameterized type
- if (curT instanceof ParameterizedType) {
- curT = ((Class<?>) ((ParameterizedType) curT).getRawType()).getGenericSuperclass();
- }
- // class
- else {
- curT = ((Class<?>) curT).getGenericSuperclass();
- }
- }
-
- // check if immediate child of Tuple has generics
- if (curT instanceof Class<?>) {
- throw new InvalidTypesException("Tuple needs to be parameterized by using generics.");
- }
-
- ParameterizedType tupleChild = (ParameterizedType) curT;
-
- Type[] subtypes = new Type[tupleChild.getActualTypeArguments().length];
-
- // materialize possible type variables
- for (int i = 0; i < subtypes.length; i++) {
- // materialize immediate TypeVariables
- if (tupleChild.getActualTypeArguments()[i] instanceof TypeVariable<?>) {
- Type varContent = materializeTypeVariable(typeHierarchy, (TypeVariable<?>) tupleChild.getActualTypeArguments()[i]);
- // variable could not be materialized
- if (varContent == null) {
- // add the TypeVariable as subtype for step in next section
- subtypes[i] = tupleChild.getActualTypeArguments()[i];
- } else {
- // add class or parameterized type
- subtypes[i] = varContent;
- }
- }
- // class or parameterized type
- else {
- subtypes[i] = tupleChild.getActualTypeArguments()[i];
- }
- }
-
- TypeInformation<?>[] tupleSubTypes = new TypeInformation<?>[subtypes.length];
- for (int i = 0; i < subtypes.length; i++) {
- // sub type could not be determined with materializing
- // try to derive the type info of the TypeVariable from the immediate base child input as a last attempt
- if (subtypes[i] instanceof TypeVariable<?>) {
- ParameterizedType immediateBaseChild = (ParameterizedType) typeHierarchy.get(typeHierarchy.size() - 1);
- tupleSubTypes[i] = createTypeInfoWithImmediateBaseChildInput(immediateBaseChild, (TypeVariable<?>) subtypes[i],
- in1Type, in2Type);
-
- // variable could not be determined
- if (tupleSubTypes[i] == null) {
- throw new InvalidTypesException("Type of TypeVariable '" + ((TypeVariable<?>) subtypes[i]).getName() + "' in '"
- + ((TypeVariable<?>) subtypes[i]).getGenericDeclaration()
- + "' could not be determined. This is most likely a type erasure problem. "
- + "The type extraction currently supports types with generic variables only in cases where "
- + "all variables in the return type can be deduced from the input type(s).");
- }
- } else {
- tupleSubTypes[i] = createTypeInfoWithTypeHierarchy(new ArrayList<Type>(typeHierarchy), subtypes[i], in1Type, in2Type);
- }
- }
-
- // TODO: Check that type that extends Tuple does not have additional fields.
- // Right now, these fields are not be serialized by the TupleSerializer.
- // We might want to add an ExtendedTupleSerializer for that.
-
- if (t instanceof Class<?>) {
- return new TupleTypeInfo(((Class<? extends Tuple>) t), tupleSubTypes);
- } else if (t instanceof ParameterizedType) {
- return new TupleTypeInfo(((Class<? extends Tuple>) ((ParameterizedType) t).getRawType()), tupleSubTypes);
- }
- }
- // type depends on another type
- // e.g. class MyMapper<E> extends MapFunction<String, E>
- else if (t instanceof TypeVariable) {
- Type typeVar = materializeTypeVariable(typeHierarchy, (TypeVariable<?>) t);
-
- if (typeVar != null) {
- return createTypeInfoWithTypeHierarchy(typeHierarchy, typeVar, in1Type, in2Type);
- }
- // try to derive the type info of the TypeVariable from the immediate base child input as a last attempt
- else {
- ParameterizedType immediateBaseChild = (ParameterizedType) typeHierarchy.get(typeHierarchy.size() - 1);
- TypeInformation<OUT> typeInfo = (TypeInformation<OUT>) createTypeInfoWithImmediateBaseChildInput(immediateBaseChild,
- (TypeVariable<?>) t, in1Type, in2Type);
- if (typeInfo != null) {
- return typeInfo;
- } else {
- throw new InvalidTypesException("Type of TypeVariable '" + ((TypeVariable<?>) t).getName() + "' in '"
- + ((TypeVariable<?>) t).getGenericDeclaration() + "' could not be determined. This is most likely a type erasure problem. "
- + "The type extraction currently supports types with generic variables only in cases where "
- + "all variables in the return type can be deduced from the input type(s).");
- }
- }
- }
- // arrays with generics
- else if (t instanceof GenericArrayType) {
- GenericArrayType genericArray = (GenericArrayType) t;
-
- Type componentType = genericArray.getGenericComponentType();
-
- // due to a Java 6 bug, it is possible that the JVM classifies e.g. String[] or int[] as GenericArrayType instead of Class
- if (componentType instanceof Class) {
-
- Class<?> componentClass = (Class<?>) componentType;
- String className;
- // for int[], double[] etc.
- if(componentClass.isPrimitive()) {
- className = encodePrimitiveClass(componentClass);
- }
- // for String[], Integer[] etc.
- else {
- className = "L" + componentClass.getName() + ";";
- }
-
- Class<OUT> classArray = null;
- try {
- classArray = (Class<OUT>) Class.forName("[" + className);
- } catch (ClassNotFoundException e) {
- throw new InvalidTypesException("Could not convert GenericArrayType to Class.");
- }
- return getForClass(classArray);
- }
-
- TypeInformation<?> componentInfo = createTypeInfoWithTypeHierarchy(typeHierarchy, genericArray.getGenericComponentType(),
- in1Type, in2Type);
- return ObjectArrayTypeInfo.getInfoFor(t, componentInfo);
- }
- // objects with generics are treated as raw type
- else if (t instanceof ParameterizedType) {
- return privateGetForClass((Class<OUT>) ((ParameterizedType) t).getRawType());
- }
- // no tuple, no TypeVariable, no generic type
- else if (t instanceof Class) {
- return privateGetForClass((Class<OUT>) t);
- }
-
- throw new InvalidTypesException("Type Information could not be created.");
- }
+ // --------------------------------------------------------------------------------------------
+ // Utility methods
+ // --------------------------------------------------------------------------------------------
private static String encodePrimitiveClass(Class<?> primitiveClass) {
final String name = primitiveClass.getName();
@@ -586,26 +640,6 @@ public class TypeExtractor {
throw new InvalidTypesException();
}
- private static <IN1, IN2> TypeInformation<?> createTypeInfoWithImmediateBaseChildInput(ParameterizedType baseChild,
- TypeVariable<?> typeVar, TypeInformation<IN1> in1Type, TypeInformation<IN2> in2Type) {
- Type[] baseChildArgs = baseChild.getActualTypeArguments();
-
- TypeInformation<?> info = null;
- if (in1Type != null) {
- info = findCorrespondingInfo(typeVar, baseChildArgs[0], in1Type);
- }
-
- if (info == null && in2Type != null) {
- info = findCorrespondingInfo(typeVar, baseChildArgs[1], in2Type);
- }
-
- if (info != null) {
- return info;
- }
-
- return null;
- }
-
private static TypeInformation<?> findCorrespondingInfo(TypeVariable<?> typeVar, Type type, TypeInformation<?> corrInfo) {
if (type instanceof TypeVariable) {
TypeVariable<?> variable = (TypeVariable<?>) type;
@@ -627,7 +661,6 @@ public class TypeExtractor {
}
private static Type materializeTypeVariable(ArrayList<Type> typeHierarchy, TypeVariable<?> typeVar) {
-
TypeVariable<?> inTypeTypeVar = typeVar;
// iterate thru hierarchy from top to bottom until type variable gets a class assigned
for (int i = typeHierarchy.size() - 1; i >= 0; i--) {
@@ -659,13 +692,14 @@ public class TypeExtractor {
}
}
// can not be materialized, most likely due to type erasure
- return null;
+ // return the type variable of the deepest level
+ return inTypeTypeVar;
}
-
+
public static <X> TypeInformation<X> getForClass(Class<X> clazz) {
return new TypeExtractor().privateGetForClass(clazz);
}
-
+
@SuppressWarnings("unchecked")
private <X> TypeInformation<X> privateGetForClass(Class<X> clazz) {
Validate.notNull(clazz);
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/5cec8e8a/stratosphere-java/src/test/java/eu/stratosphere/api/java/type/extractor/TypeExtractorTest.java
----------------------------------------------------------------------
diff --git a/stratosphere-java/src/test/java/eu/stratosphere/api/java/type/extractor/TypeExtractorTest.java b/stratosphere-java/src/test/java/eu/stratosphere/api/java/type/extractor/TypeExtractorTest.java
index c3923fc..43fd0ce 100644
--- a/stratosphere-java/src/test/java/eu/stratosphere/api/java/type/extractor/TypeExtractorTest.java
+++ b/stratosphere-java/src/test/java/eu/stratosphere/api/java/type/extractor/TypeExtractorTest.java
@@ -23,6 +23,8 @@ import org.apache.hadoop.io.Writable;
import org.junit.Assert;
import org.junit.Test;
+import eu.stratosphere.api.common.functions.GenericMap;
+import eu.stratosphere.api.common.functions.RuntimeContext;
import eu.stratosphere.api.java.functions.CoGroupFunction;
import eu.stratosphere.api.java.functions.CrossFunction;
import eu.stratosphere.api.java.functions.FlatMapFunction;
@@ -40,14 +42,15 @@ import eu.stratosphere.api.java.typeutils.BasicArrayTypeInfo;
import eu.stratosphere.api.java.typeutils.BasicTypeInfo;
import eu.stratosphere.api.java.typeutils.GenericTypeInfo;
import eu.stratosphere.api.java.typeutils.ObjectArrayTypeInfo;
-import eu.stratosphere.api.java.typeutils.PrimitiveArrayTypeInfo;
import eu.stratosphere.api.java.typeutils.PojoTypeInfo;
+import eu.stratosphere.api.java.typeutils.PrimitiveArrayTypeInfo;
import eu.stratosphere.api.java.typeutils.ResultTypeQueryable;
import eu.stratosphere.api.java.typeutils.TupleTypeInfo;
import eu.stratosphere.api.java.typeutils.TypeExtractor;
import eu.stratosphere.api.java.typeutils.TypeInfoParser;
import eu.stratosphere.api.java.typeutils.ValueTypeInfo;
import eu.stratosphere.api.java.typeutils.WritableTypeInfo;
+import eu.stratosphere.configuration.Configuration;
import eu.stratosphere.types.DoubleValue;
import eu.stratosphere.types.IntValue;
import eu.stratosphere.types.StringValue;
@@ -1374,4 +1377,38 @@ public class TypeExtractorTest {
Assert.assertEquals(PrimitiveArrayTypeInfo.BOOLEAN_PRIMITIVE_ARRAY_TYPE_INFO, tti.getTypeAt(7));
Assert.assertEquals(BasicArrayTypeInfo.STRING_ARRAY_TYPE_INFO, tti.getTypeAt(8));
}
+
+ @Test
+ public void testInterface() {
+ GenericMap<String, Boolean> mapInterface = new GenericMap<String, Boolean>() {
+
+ @Override
+ public void setRuntimeContext(RuntimeContext t) {
+
+ }
+
+ @Override
+ public void open(Configuration parameters) throws Exception {
+
+ }
+
+ @Override
+ public RuntimeContext getRuntimeContext() {
+ return null;
+ }
+
+ @Override
+ public void close() throws Exception {
+
+ }
+
+ @Override
+ public Boolean map(String record) throws Exception {
+ return null;
+ }
+ };
+
+ TypeInformation<?> ti = TypeExtractor.getMapReturnTypes(mapInterface, BasicTypeInfo.STRING_TYPE_INFO);
+ Assert.assertEquals(BasicTypeInfo.BOOLEAN_TYPE_INFO, ti);
+ }
}
[03/12] git commit: [FLINK-836] Rework of the cached match driver
Posted by se...@apache.org.
[FLINK-836] Rework of the cached match driver
Project: http://git-wip-us.apache.org/repos/asf/incubator-flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-flink/commit/47a68adc
Tree: http://git-wip-us.apache.org/repos/asf/incubator-flink/tree/47a68adc
Diff: http://git-wip-us.apache.org/repos/asf/incubator-flink/diff/47a68adc
Branch: refs/heads/master
Commit: 47a68adc7fa5da7743cb51fe69f81264e908e630
Parents: 99c888c
Author: Markus Holzemer <ma...@gmx.de>
Authored: Mon Jun 16 09:41:46 2014 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Tue Jul 8 16:54:15 2014 +0200
----------------------------------------------------------------------
.../compiler/dag/OptimizerNode.java | 10 -
.../operators/HashJoinBuildFirstProperties.java | 2 +-
.../HashJoinBuildSecondProperties.java | 2 +-
.../eu/stratosphere/compiler/plan/Channel.java | 4 -
.../eu/stratosphere/compiler/plan/PlanNode.java | 4 -
.../services/memorymanager/MemoryManager.java | 8 +-
.../memorymanager/spi/DefaultMemoryManager.java | 4 +-
.../BuildFirstReOpenableHashMatchIterator.java | 10 +-
.../hash/BuildSecondHashMatchIterator.java | 6 +-
.../BuildSecondReOpenableHashMatchIterator.java | 71 +++
.../runtime/sort/AsynchronousPartialSorter.java | 13 +-
.../pact/runtime/sort/UnilateralSortMerger.java | 2 +-
.../AbstractCachedBuildSideMatchDriver.java | 131 ++---
.../pact/runtime/task/RegularPactTask.java | 10 +
.../sort/AsynchonousPartialSorterITCase.java | 6 +-
.../pact/runtime/task/CachedMatchTaskTest.java | 491 +++++++++++++++++++
.../pact/runtime/test/util/DriverTestBase.java | 33 ++
17 files changed, 683 insertions(+), 124 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/47a68adc/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/dag/OptimizerNode.java
----------------------------------------------------------------------
diff --git a/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/dag/OptimizerNode.java b/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/dag/OptimizerNode.java
index 6f295a5..ec9bd69 100644
--- a/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/dag/OptimizerNode.java
+++ b/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/dag/OptimizerNode.java
@@ -92,8 +92,6 @@ public abstract class OptimizerNode implements Visitable<OptimizerNode>, Estimat
protected boolean onDynamicPath;
- protected boolean insideIteration;
-
protected List<PlanNode> cachedPlans; // cache candidates, because the may be accessed repeatedly
protected int[][] remappedKeys;
@@ -501,14 +499,6 @@ public abstract class OptimizerNode implements Visitable<OptimizerNode>, Estimat
}
}
- public boolean isInsideIteration() {
- return insideIteration;
- }
-
- public void setInsideIteration(boolean insideIteration) {
- this.insideIteration = insideIteration;
- }
-
/**
* Checks whether this node has branching output. A node's output is branched, if it has more
* than one output connection.
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/47a68adc/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/operators/HashJoinBuildFirstProperties.java
----------------------------------------------------------------------
diff --git a/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/operators/HashJoinBuildFirstProperties.java b/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/operators/HashJoinBuildFirstProperties.java
index 661b316..4f694c5 100644
--- a/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/operators/HashJoinBuildFirstProperties.java
+++ b/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/operators/HashJoinBuildFirstProperties.java
@@ -56,7 +56,7 @@ public class HashJoinBuildFirstProperties extends AbstractJoinDescriptor {
public DualInputPlanNode instantiate(Channel in1, Channel in2, TwoInputNode node) {
DriverStrategy strategy;
- if(!in1.isOnDynamicPath() && in1.isInsideIteration() && in2.isInsideIteration()) {
+ if(!in1.isOnDynamicPath() && in2.isOnDynamicPath()) {
strategy = DriverStrategy.HYBRIDHASH_BUILD_FIRST_CACHED;
}
else {
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/47a68adc/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/operators/HashJoinBuildSecondProperties.java
----------------------------------------------------------------------
diff --git a/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/operators/HashJoinBuildSecondProperties.java b/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/operators/HashJoinBuildSecondProperties.java
index e085588..6bea65a 100644
--- a/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/operators/HashJoinBuildSecondProperties.java
+++ b/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/operators/HashJoinBuildSecondProperties.java
@@ -53,7 +53,7 @@ public final class HashJoinBuildSecondProperties extends AbstractJoinDescriptor
public DualInputPlanNode instantiate(Channel in1, Channel in2, TwoInputNode node) {
DriverStrategy strategy;
- if(!in2.isOnDynamicPath() && in1.isInsideIteration() && in2.isInsideIteration()) {
+ if(!in2.isOnDynamicPath() && in1.isOnDynamicPath()) {
strategy = DriverStrategy.HYBRIDHASH_BUILD_SECOND_CACHED;
}
else {
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/47a68adc/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/plan/Channel.java
----------------------------------------------------------------------
diff --git a/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/plan/Channel.java b/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/plan/Channel.java
index d83da82..6f9418f 100644
--- a/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/plan/Channel.java
+++ b/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/plan/Channel.java
@@ -306,10 +306,6 @@ public class Channel implements EstimateProvider, Cloneable, DumpableConnection<
return this.source.isOnDynamicPath();
}
- public boolean isInsideIteration() {
- return this.source.isInsideIteration();
- }
-
public int getCostWeight() {
return this.source.getCostWeight();
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/47a68adc/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/plan/PlanNode.java
----------------------------------------------------------------------
diff --git a/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/plan/PlanNode.java b/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/plan/PlanNode.java
index da24b5f..539006c 100644
--- a/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/plan/PlanNode.java
+++ b/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/plan/PlanNode.java
@@ -423,10 +423,6 @@ public abstract class PlanNode implements Visitable<PlanNode>, DumpableNode<Plan
return this.template.getCostWeight();
}
- public boolean isInsideIteration() {
- return this.template.isInsideIteration();
- }
-
// --------------------------------------------------------------------------------------------
/**
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/47a68adc/stratosphere-runtime/src/main/java/eu/stratosphere/nephele/services/memorymanager/MemoryManager.java
----------------------------------------------------------------------
diff --git a/stratosphere-runtime/src/main/java/eu/stratosphere/nephele/services/memorymanager/MemoryManager.java b/stratosphere-runtime/src/main/java/eu/stratosphere/nephele/services/memorymanager/MemoryManager.java
index 8b20c75..6c5109b 100644
--- a/stratosphere-runtime/src/main/java/eu/stratosphere/nephele/services/memorymanager/MemoryManager.java
+++ b/stratosphere-runtime/src/main/java/eu/stratosphere/nephele/services/memorymanager/MemoryManager.java
@@ -72,7 +72,8 @@ public interface MemoryManager {
/**
* Returns the total size of memory.
- * @return
+ *
+ * @return The total size of memory.
*/
long getMemorySize();
@@ -88,8 +89,9 @@ public interface MemoryManager {
/**
* Computes the memory size of the fraction per slot.
- * @param fraction
- * @return
+ *
+ * @param fraction The fraction of the memory of the task slot.
+ * @return The number of pages corresponding to the memory fraction.
*/
long computeMemorySize(double fraction);
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/47a68adc/stratosphere-runtime/src/main/java/eu/stratosphere/nephele/services/memorymanager/spi/DefaultMemoryManager.java
----------------------------------------------------------------------
diff --git a/stratosphere-runtime/src/main/java/eu/stratosphere/nephele/services/memorymanager/spi/DefaultMemoryManager.java b/stratosphere-runtime/src/main/java/eu/stratosphere/nephele/services/memorymanager/spi/DefaultMemoryManager.java
index d4a2b36..85455d2 100644
--- a/stratosphere-runtime/src/main/java/eu/stratosphere/nephele/services/memorymanager/spi/DefaultMemoryManager.java
+++ b/stratosphere-runtime/src/main/java/eu/stratosphere/nephele/services/memorymanager/spi/DefaultMemoryManager.java
@@ -394,8 +394,8 @@ public class DefaultMemoryManager implements MemoryManager {
}
private final int getRelativeNumPages(double fraction){
- if(fraction < 0){
- throw new IllegalArgumentException("The fraction of memory to allocate must not be negative.");
+ if (fraction <= 0 || fraction > 1) {
+ throw new IllegalArgumentException("The fraction of memory to allocate must within (0, 1].");
}
return (int)(this.totalNumPages * fraction / this.numberOfSlots);
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/47a68adc/stratosphere-runtime/src/main/java/eu/stratosphere/pact/runtime/hash/BuildFirstReOpenableHashMatchIterator.java
----------------------------------------------------------------------
diff --git a/stratosphere-runtime/src/main/java/eu/stratosphere/pact/runtime/hash/BuildFirstReOpenableHashMatchIterator.java b/stratosphere-runtime/src/main/java/eu/stratosphere/pact/runtime/hash/BuildFirstReOpenableHashMatchIterator.java
index 8c2b9ca..7898c41 100644
--- a/stratosphere-runtime/src/main/java/eu/stratosphere/pact/runtime/hash/BuildFirstReOpenableHashMatchIterator.java
+++ b/stratosphere-runtime/src/main/java/eu/stratosphere/pact/runtime/hash/BuildFirstReOpenableHashMatchIterator.java
@@ -37,15 +37,19 @@ public class BuildFirstReOpenableHashMatchIterator<V1, V2, O> extends BuildFirst
TypeSerializer<V1> serializer1, TypeComparator<V1> comparator1,
TypeSerializer<V2> serializer2, TypeComparator<V2> comparator2,
TypePairComparator<V2, V1> pairComparator,
- MemoryManager memManager, IOManager ioManager,
- AbstractInvokable ownerTask, double memoryFraction)
- throws MemoryAllocationException {
+ MemoryManager memManager,
+ IOManager ioManager,
+ AbstractInvokable ownerTask,
+ double memoryFraction)
+ throws MemoryAllocationException
+ {
super(firstInput, secondInput, serializer1, comparator1, serializer2,
comparator2, pairComparator, memManager, ioManager, ownerTask,
memoryFraction);
reopenHashTable = (ReOpenableMutableHashTable<V1, V2>) hashJoin;
}
+ @Override
public <BT, PT> MutableHashTable<BT, PT> getHashJoin(TypeSerializer<BT> buildSideSerializer, TypeComparator<BT> buildSideComparator,
TypeSerializer<PT> probeSideSerializer, TypeComparator<PT> probeSideComparator,
TypePairComparator<PT, BT> pairComparator,
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/47a68adc/stratosphere-runtime/src/main/java/eu/stratosphere/pact/runtime/hash/BuildSecondHashMatchIterator.java
----------------------------------------------------------------------
diff --git a/stratosphere-runtime/src/main/java/eu/stratosphere/pact/runtime/hash/BuildSecondHashMatchIterator.java b/stratosphere-runtime/src/main/java/eu/stratosphere/pact/runtime/hash/BuildSecondHashMatchIterator.java
index 732d256..9f3fd97 100644
--- a/stratosphere-runtime/src/main/java/eu/stratosphere/pact/runtime/hash/BuildSecondHashMatchIterator.java
+++ b/stratosphere-runtime/src/main/java/eu/stratosphere/pact/runtime/hash/BuildSecondHashMatchIterator.java
@@ -34,9 +34,9 @@ import eu.stratosphere.util.MutableObjectIterator;
* An implementation of the {@link eu.stratosphere.pact.runtime.task.util.JoinTaskIterator} that uses a hybrid-hash-join
* internally to match the records with equal key. The build side of the hash is the second input of the match.
*/
-public final class BuildSecondHashMatchIterator<V1, V2, O> implements JoinTaskIterator<V1, V2, O> {
+public class BuildSecondHashMatchIterator<V1, V2, O> implements JoinTaskIterator<V1, V2, O> {
- private final MutableHashTable<V2, V1> hashJoin;
+ protected final MutableHashTable<V2, V1> hashJoin;
private final V2 nextBuildSideObject;
@@ -44,7 +44,7 @@ public final class BuildSecondHashMatchIterator<V1, V2, O> implements JoinTaskIt
private final V1 probeCopy;
- private final TypeSerializer<V1> probeSideSerializer;
+ protected final TypeSerializer<V1> probeSideSerializer;
private final MemoryManager memManager;
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/47a68adc/stratosphere-runtime/src/main/java/eu/stratosphere/pact/runtime/hash/BuildSecondReOpenableHashMatchIterator.java
----------------------------------------------------------------------
diff --git a/stratosphere-runtime/src/main/java/eu/stratosphere/pact/runtime/hash/BuildSecondReOpenableHashMatchIterator.java b/stratosphere-runtime/src/main/java/eu/stratosphere/pact/runtime/hash/BuildSecondReOpenableHashMatchIterator.java
new file mode 100644
index 0000000..597ae73
--- /dev/null
+++ b/stratosphere-runtime/src/main/java/eu/stratosphere/pact/runtime/hash/BuildSecondReOpenableHashMatchIterator.java
@@ -0,0 +1,71 @@
+/***********************************************************************************************************************
+ * Copyright (C) 2010-2013 by the Stratosphere project (http://stratosphere.eu)
+ *
+ * Licensed 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 eu.stratosphere.pact.runtime.hash;
+
+import java.io.IOException;
+import java.util.List;
+
+import eu.stratosphere.api.common.typeutils.TypeComparator;
+import eu.stratosphere.api.common.typeutils.TypePairComparator;
+import eu.stratosphere.api.common.typeutils.TypeSerializer;
+import eu.stratosphere.core.memory.MemorySegment;
+import eu.stratosphere.nephele.services.iomanager.IOManager;
+import eu.stratosphere.nephele.services.memorymanager.MemoryAllocationException;
+import eu.stratosphere.nephele.services.memorymanager.MemoryManager;
+import eu.stratosphere.nephele.template.AbstractInvokable;
+import eu.stratosphere.util.MutableObjectIterator;
+
+public class BuildSecondReOpenableHashMatchIterator<V1, V2, O> extends BuildSecondHashMatchIterator<V1, V2, O> {
+
+
+ private final ReOpenableMutableHashTable<V2, V1> reopenHashTable;
+
+ public BuildSecondReOpenableHashMatchIterator(
+ MutableObjectIterator<V1> firstInput,
+ MutableObjectIterator<V2> secondInput,
+ TypeSerializer<V1> serializer1, TypeComparator<V1> comparator1,
+ TypeSerializer<V2> serializer2, TypeComparator<V2> comparator2,
+ TypePairComparator<V1, V2> pairComparator,
+ MemoryManager memManager,
+ IOManager ioManager,
+ AbstractInvokable ownerTask,
+ double memoryFraction)
+ throws MemoryAllocationException
+ {
+ super(firstInput, secondInput, serializer1, comparator1, serializer2,
+ comparator2, pairComparator, memManager, ioManager, ownerTask, memoryFraction);
+ reopenHashTable = (ReOpenableMutableHashTable<V2, V1>) hashJoin;
+ }
+
+ @Override
+ public <BT, PT> MutableHashTable<BT, PT> getHashJoin(TypeSerializer<BT> buildSideSerializer, TypeComparator<BT> buildSideComparator,
+ TypeSerializer<PT> probeSideSerializer, TypeComparator<PT> probeSideComparator,
+ TypePairComparator<PT, BT> pairComparator,
+ MemoryManager memManager, IOManager ioManager, AbstractInvokable ownerTask, double memoryFraction)
+ throws MemoryAllocationException
+ {
+ final int numPages = memManager.computeNumberOfPages(memoryFraction);
+ final List<MemorySegment> memorySegments = memManager.allocatePages(ownerTask, numPages);
+ return new ReOpenableMutableHashTable<BT, PT>(buildSideSerializer, probeSideSerializer, buildSideComparator, probeSideComparator, pairComparator, memorySegments, ioManager);
+ }
+
+ /**
+ * Set new input for probe side
+ * @throws IOException
+ */
+ public void reopenProbe(MutableObjectIterator<V1> probeInput) throws IOException {
+ reopenHashTable.reopenProbe(probeInput);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/47a68adc/stratosphere-runtime/src/main/java/eu/stratosphere/pact/runtime/sort/AsynchronousPartialSorter.java
----------------------------------------------------------------------
diff --git a/stratosphere-runtime/src/main/java/eu/stratosphere/pact/runtime/sort/AsynchronousPartialSorter.java b/stratosphere-runtime/src/main/java/eu/stratosphere/pact/runtime/sort/AsynchronousPartialSorter.java
index 35377cf..f87f1f8 100644
--- a/stratosphere-runtime/src/main/java/eu/stratosphere/pact/runtime/sort/AsynchronousPartialSorter.java
+++ b/stratosphere-runtime/src/main/java/eu/stratosphere/pact/runtime/sort/AsynchronousPartialSorter.java
@@ -34,8 +34,6 @@ import eu.stratosphere.util.MutableObjectIterator;
*/
public class AsynchronousPartialSorter<E> extends UnilateralSortMerger<E> {
- private static final int MAX_MEM_PER_PARTIAL_SORT = 64 * 1024 * 0124;
-
private BufferQueueIterator bufferIterator;
// ------------------------------------------------------------------------
@@ -62,11 +60,7 @@ public class AsynchronousPartialSorter<E> extends UnilateralSortMerger<E> {
double memoryFraction)
throws IOException, MemoryAllocationException
{
- super(memoryManager, null, input, parentTask, serializerFactory, comparator, memoryFraction,
- memoryManager.computeNumberOfPages(memoryFraction) < 2 * MIN_NUM_SORT_MEM_SEGMENTS ? 1 :
- Math.max((int) Math.ceil(((double) memoryManager.computeMemorySize(memoryFraction)) /
- MAX_MEM_PER_PARTIAL_SORT), 2),
- 2, 0.0f, true);
+ super(memoryManager, null, input, parentTask, serializerFactory, comparator, memoryFraction, 1, 2, 0.0f, true);
}
@@ -101,11 +95,6 @@ public class AsynchronousPartialSorter<E> extends UnilateralSortMerger<E> {
// ------------------------------------------------------------------------
- /**
- * This class implements an iterator over values from a {@link eu.stratosphere.pact.runtime.sort.BufferSortable}.
- * The iterator returns the values of a given
- * interval.
- */
private final class BufferQueueIterator implements MutableObjectIterator<E> {
private final CircularQueues<E> queues;
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/47a68adc/stratosphere-runtime/src/main/java/eu/stratosphere/pact/runtime/sort/UnilateralSortMerger.java
----------------------------------------------------------------------
diff --git a/stratosphere-runtime/src/main/java/eu/stratosphere/pact/runtime/sort/UnilateralSortMerger.java b/stratosphere-runtime/src/main/java/eu/stratosphere/pact/runtime/sort/UnilateralSortMerger.java
index 6905b85..9109e9a 100644
--- a/stratosphere-runtime/src/main/java/eu/stratosphere/pact/runtime/sort/UnilateralSortMerger.java
+++ b/stratosphere-runtime/src/main/java/eu/stratosphere/pact/runtime/sort/UnilateralSortMerger.java
@@ -239,7 +239,7 @@ public class UnilateralSortMerger<E> implements Sorter<E> {
* @param maxNumFileHandles The maximum number of files to be merged at once.
* @param startSpillingFraction The faction of the buffers that have to be filled before the spilling thread
* actually begins spilling data to disk.
- * @param noSpilling When set to true, no memory will be allocated for writing and no spilling thread
+ * @param noSpillingMemory When set to true, no memory will be allocated for writing and no spilling thread
* will be spawned.
*
* @throws IOException Thrown, if an error occurs initializing the resources for external sorting.
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/47a68adc/stratosphere-runtime/src/main/java/eu/stratosphere/pact/runtime/task/AbstractCachedBuildSideMatchDriver.java
----------------------------------------------------------------------
diff --git a/stratosphere-runtime/src/main/java/eu/stratosphere/pact/runtime/task/AbstractCachedBuildSideMatchDriver.java b/stratosphere-runtime/src/main/java/eu/stratosphere/pact/runtime/task/AbstractCachedBuildSideMatchDriver.java
index 1d3c55d..1c8c427 100644
--- a/stratosphere-runtime/src/main/java/eu/stratosphere/pact/runtime/task/AbstractCachedBuildSideMatchDriver.java
+++ b/stratosphere-runtime/src/main/java/eu/stratosphere/pact/runtime/task/AbstractCachedBuildSideMatchDriver.java
@@ -13,26 +13,20 @@
package eu.stratosphere.pact.runtime.task;
-import java.util.List;
-
import eu.stratosphere.api.common.functions.GenericJoiner;
import eu.stratosphere.api.common.typeutils.TypeComparator;
import eu.stratosphere.api.common.typeutils.TypePairComparatorFactory;
import eu.stratosphere.api.common.typeutils.TypeSerializer;
-import eu.stratosphere.core.memory.MemorySegment;
-import eu.stratosphere.pact.runtime.hash.MutableHashTable;
+import eu.stratosphere.pact.runtime.hash.BuildFirstReOpenableHashMatchIterator;
+import eu.stratosphere.pact.runtime.hash.BuildSecondReOpenableHashMatchIterator;
+import eu.stratosphere.pact.runtime.task.util.JoinTaskIterator;
import eu.stratosphere.pact.runtime.task.util.TaskConfig;
-import eu.stratosphere.pact.runtime.util.EmptyMutableObjectIterator;
import eu.stratosphere.util.Collector;
import eu.stratosphere.util.MutableObjectIterator;
public abstract class AbstractCachedBuildSideMatchDriver<IT1, IT2, OT> extends MatchDriver<IT1, IT2, OT> implements ResettablePactDriver<GenericJoiner<IT1, IT2, OT>, OT> {
-
- /**
- * We keep it without generic parameters, because they vary depending on which input is the build side.
- */
- protected volatile MutableHashTable<?, ?> hashJoin;
+ private volatile JoinTaskIterator<IT1, IT2, OT> matchIterator;
private final int buildSideIndex;
@@ -67,23 +61,39 @@ public abstract class AbstractCachedBuildSideMatchDriver<IT1, IT2, OT> extends M
TypePairComparatorFactory<IT1, IT2> pairComparatorFactory =
this.taskContext.getTaskConfig().getPairComparatorFactory(this.taskContext.getUserCodeClassLoader());
- int numMemoryPages = this.taskContext.getMemoryManager().computeNumberOfPages(config.getRelativeMemoryDriver());
- List<MemorySegment> memSegments = this.taskContext.getMemoryManager().allocatePages(
- this.taskContext.getOwningNepheleTask(), numMemoryPages);
+ double availableMemory = config.getRelativeMemoryDriver();
if (buildSideIndex == 0 && probeSideIndex == 1) {
- MutableHashTable<IT1, IT2> hashJoin = new MutableHashTable<IT1, IT2>(serializer1, serializer2, comparator1, comparator2,
- pairComparatorFactory.createComparator21(comparator1, comparator2), memSegments, this.taskContext.getIOManager());
- this.hashJoin = hashJoin;
- hashJoin.open(input1, EmptyMutableObjectIterator.<IT2>get());
+
+ matchIterator =
+ new BuildFirstReOpenableHashMatchIterator<IT1, IT2, OT>(input1, input2,
+ serializer1, comparator1,
+ serializer2, comparator2,
+ pairComparatorFactory.createComparator21(comparator1, comparator2),
+ this.taskContext.getMemoryManager(),
+ this.taskContext.getIOManager(),
+ this.taskContext.getOwningNepheleTask(),
+ availableMemory
+ );
+
} else if (buildSideIndex == 1 && probeSideIndex == 0) {
- MutableHashTable<IT2, IT1> hashJoin = new MutableHashTable<IT2, IT1>(serializer2, serializer1, comparator2, comparator1,
- pairComparatorFactory.createComparator12(comparator1, comparator2), memSegments, this.taskContext.getIOManager());
- this.hashJoin = hashJoin;
- hashJoin.open(input2, EmptyMutableObjectIterator.<IT1>get());
+
+ matchIterator =
+ new BuildSecondReOpenableHashMatchIterator<IT1, IT2, OT>(input1, input2,
+ serializer1, comparator1,
+ serializer2, comparator2,
+ pairComparatorFactory.createComparator12(comparator1, comparator2),
+ this.taskContext.getMemoryManager(),
+ this.taskContext.getIOManager(),
+ this.taskContext.getOwningNepheleTask(),
+ availableMemory
+ );
+
} else {
throw new Exception("Error: Inconcistent setup for repeatable hash join driver.");
}
+
+ this.matchIterator.open();
}
@Override
@@ -98,63 +108,17 @@ public abstract class AbstractCachedBuildSideMatchDriver<IT1, IT2, OT> extends M
final Collector<OT> collector = this.taskContext.getOutputCollector();
if (buildSideIndex == 0) {
- final TypeSerializer<IT1> buildSideSerializer = taskContext.<IT1> getInputSerializer(0).getSerializer();
- final TypeSerializer<IT2> probeSideSerializer = taskContext.<IT2> getInputSerializer(1).getSerializer();
- IT1 buildSideRecordFirst;
- IT1 buildSideRecordOther;
- IT2 probeSideRecord;
- IT2 probeSideRecordCopy;
- final IT1 buildSideRecordFirstReuse = buildSideSerializer.createInstance();
- final IT1 buildSideRecordOtherReuse = buildSideSerializer.createInstance();
- final IT2 probeSideRecordReuse = probeSideSerializer.createInstance();
- final IT2 probeSideRecordCopyReuse = probeSideSerializer.createInstance();
+ final BuildFirstReOpenableHashMatchIterator<IT1, IT2, OT> matchIterator = (BuildFirstReOpenableHashMatchIterator<IT1, IT2, OT>) this.matchIterator;
- @SuppressWarnings("unchecked")
- final MutableHashTable<IT1, IT2> join = (MutableHashTable<IT1, IT2>) this.hashJoin;
+ while (this.running && matchIterator != null && matchIterator.callWithNextKey(matchStub, collector));
- final MutableObjectIterator<IT2> probeSideInput = taskContext.<IT2>getInput(1);
-
- while (this.running && ((probeSideRecord = probeSideInput.next(probeSideRecordReuse)) != null)) {
- final MutableHashTable.HashBucketIterator<IT1, IT2> bucket = join.getMatchesFor(probeSideRecord);
-
- if ((buildSideRecordFirst = bucket.next(buildSideRecordFirstReuse)) != null) {
- while ((buildSideRecordOther = bucket.next(buildSideRecordOtherReuse)) != null) {
- probeSideRecordCopy = probeSideSerializer.copy(probeSideRecord, probeSideRecordCopyReuse);
- matchStub.join(buildSideRecordOther, probeSideRecordCopy, collector);
- }
- matchStub.join(buildSideRecordFirst, probeSideRecord, collector);
- }
- }
} else if (buildSideIndex == 1) {
- final TypeSerializer<IT2> buildSideSerializer = taskContext.<IT2>getInputSerializer(1).getSerializer();
- final TypeSerializer<IT1> probeSideSerializer = taskContext.<IT1>getInputSerializer(0).getSerializer();
- IT2 buildSideRecordFirst;
- IT2 buildSideRecordOther;
- IT1 probeSideRecord;
- IT1 probeSideRecordCopy;
- final IT2 buildSideRecordFirstReuse = buildSideSerializer.createInstance();
- final IT2 buildSideRecordOtherReuse = buildSideSerializer.createInstance();
- final IT1 probeSideRecordReuse = probeSideSerializer.createInstance();
- final IT1 probeSideRecordCopyReuse = probeSideSerializer.createInstance();
-
- @SuppressWarnings("unchecked")
- final MutableHashTable<IT2, IT1> join = (MutableHashTable<IT2, IT1>) this.hashJoin;
+ final BuildSecondReOpenableHashMatchIterator<IT1, IT2, OT> matchIterator = (BuildSecondReOpenableHashMatchIterator<IT1, IT2, OT>) this.matchIterator;
- final MutableObjectIterator<IT1> probeSideInput = taskContext.<IT1>getInput(0);
+ while (this.running && matchIterator != null && matchIterator.callWithNextKey(matchStub, collector));
- while (this.running && ((probeSideRecord = probeSideInput.next(probeSideRecordReuse)) != null)) {
- final MutableHashTable.HashBucketIterator<IT2, IT1> bucket = join.getMatchesFor(probeSideRecord);
-
- if ((buildSideRecordFirst = bucket.next(buildSideRecordFirstReuse)) != null) {
- while ((buildSideRecordOther = bucket.next(buildSideRecordOtherReuse)) != null) {
- probeSideRecordCopy = probeSideSerializer.copy(probeSideRecord, probeSideRecordCopyReuse);
- matchStub.join(probeSideRecordCopy, buildSideRecordOther, collector);
- }
- matchStub.join(probeSideRecord, buildSideRecordFirst, collector);
- }
- }
} else {
throw new Exception();
}
@@ -164,21 +128,34 @@ public abstract class AbstractCachedBuildSideMatchDriver<IT1, IT2, OT> extends M
public void cleanup() throws Exception {}
@Override
- public void reset() throws Exception {}
+ public void reset() throws Exception {
+
+ MutableObjectIterator<IT1> input1 = this.taskContext.getInput(0);
+ MutableObjectIterator<IT2> input2 = this.taskContext.getInput(1);
+
+ if (buildSideIndex == 0 && probeSideIndex == 1) {
+ final BuildFirstReOpenableHashMatchIterator<IT1, IT2, OT> matchIterator = (BuildFirstReOpenableHashMatchIterator<IT1, IT2, OT>) this.matchIterator;
+ matchIterator.reopenProbe(input2);
+ }
+ else {
+ final BuildSecondReOpenableHashMatchIterator<IT1, IT2, OT> matchIterator = (BuildSecondReOpenableHashMatchIterator<IT1, IT2, OT>) this.matchIterator;
+ matchIterator.reopenProbe(input1);
+ }
+ }
@Override
public void teardown() {
- MutableHashTable<?, ?> ht = this.hashJoin;
- if (ht != null) {
- ht.close();
+ this.running = false;
+ if (this.matchIterator != null) {
+ this.matchIterator.close();
}
}
@Override
public void cancel() {
this.running = false;
- if (this.hashJoin != null) {
- this.hashJoin.close();
+ if (this.matchIterator != null) {
+ this.matchIterator.abort();
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/47a68adc/stratosphere-runtime/src/main/java/eu/stratosphere/pact/runtime/task/RegularPactTask.java
----------------------------------------------------------------------
diff --git a/stratosphere-runtime/src/main/java/eu/stratosphere/pact/runtime/task/RegularPactTask.java b/stratosphere-runtime/src/main/java/eu/stratosphere/pact/runtime/task/RegularPactTask.java
index 3140525..3e4a1fd 100644
--- a/stratosphere-runtime/src/main/java/eu/stratosphere/pact/runtime/task/RegularPactTask.java
+++ b/stratosphere-runtime/src/main/java/eu/stratosphere/pact/runtime/task/RegularPactTask.java
@@ -531,6 +531,16 @@ public class RegularPactTask<S extends Function, OT> extends AbstractInvokable i
}
catch (Throwable t) {}
}
+
+ // if resettable driver invoke treardown
+ if (this.driver instanceof ResettablePactDriver) {
+ final ResettablePactDriver<?, ?> resDriver = (ResettablePactDriver<?, ?>) this.driver;
+ try {
+ resDriver.teardown();
+ } catch (Throwable t) {
+ throw new Exception("Error while shutting down an iterative operator: " + t.getMessage(), t);
+ }
+ }
RegularPactTask.cancelChainedTasks(this.chainedTasks);
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/47a68adc/stratosphere-runtime/src/test/java/eu/stratosphere/pact/runtime/sort/AsynchonousPartialSorterITCase.java
----------------------------------------------------------------------
diff --git a/stratosphere-runtime/src/test/java/eu/stratosphere/pact/runtime/sort/AsynchonousPartialSorterITCase.java b/stratosphere-runtime/src/test/java/eu/stratosphere/pact/runtime/sort/AsynchonousPartialSorterITCase.java
index f191075..155bf28 100644
--- a/stratosphere-runtime/src/test/java/eu/stratosphere/pact/runtime/sort/AsynchonousPartialSorterITCase.java
+++ b/stratosphere-runtime/src/test/java/eu/stratosphere/pact/runtime/sort/AsynchonousPartialSorterITCase.java
@@ -128,7 +128,7 @@ public class AsynchonousPartialSorterITCase {
// merge iterator
LOG.debug("Initializing sortmerger...");
Sorter<Record> sorter = new AsynchronousPartialSorter<Record>(this.memoryManager, source,
- this.parentTask, this.serializer, this.comparator, 1.0);
+ this.parentTask, this.serializer, this.comparator, 0.2);
runPartialSorter(sorter, NUM_RECORDS, 2);
}
@@ -151,9 +151,9 @@ public class AsynchonousPartialSorterITCase {
// merge iterator
LOG.debug("Initializing sortmerger...");
Sorter<Record> sorter = new AsynchronousPartialSorter<Record>(this.memoryManager, source,
- this.parentTask, this.serializer, this.comparator, 1.0);
+ this.parentTask, this.serializer, this.comparator, 0.15);
- runPartialSorter(sorter, NUM_RECORDS, 28);
+ runPartialSorter(sorter, NUM_RECORDS, 27);
}
catch (Exception t) {
t.printStackTrace();
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/47a68adc/stratosphere-runtime/src/test/java/eu/stratosphere/pact/runtime/task/CachedMatchTaskTest.java
----------------------------------------------------------------------
diff --git a/stratosphere-runtime/src/test/java/eu/stratosphere/pact/runtime/task/CachedMatchTaskTest.java b/stratosphere-runtime/src/test/java/eu/stratosphere/pact/runtime/task/CachedMatchTaskTest.java
new file mode 100644
index 0000000..ff560df
--- /dev/null
+++ b/stratosphere-runtime/src/test/java/eu/stratosphere/pact/runtime/task/CachedMatchTaskTest.java
@@ -0,0 +1,491 @@
+/***********************************************************************************************************************
+ * Copyright (C) 2010-2013 by the Stratosphere project (http://stratosphere.eu)
+ *
+ * Licensed 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 eu.stratosphere.pact.runtime.task;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import eu.stratosphere.api.common.functions.GenericJoiner;
+import eu.stratosphere.api.java.record.functions.JoinFunction;
+import eu.stratosphere.api.java.typeutils.runtime.record.RecordComparator;
+import eu.stratosphere.api.java.typeutils.runtime.record.RecordPairComparatorFactory;
+import eu.stratosphere.pact.runtime.test.util.DelayingInfinitiveInputIterator;
+import eu.stratosphere.pact.runtime.test.util.DriverTestBase;
+import eu.stratosphere.pact.runtime.test.util.ExpectedTestException;
+import eu.stratosphere.pact.runtime.test.util.NirvanaOutputList;
+import eu.stratosphere.pact.runtime.test.util.TaskCancelThread;
+import eu.stratosphere.pact.runtime.test.util.UniformRecordGenerator;
+import eu.stratosphere.types.IntValue;
+import eu.stratosphere.types.Key;
+import eu.stratosphere.types.Record;
+import eu.stratosphere.util.Collector;
+
+public class CachedMatchTaskTest extends DriverTestBase<GenericJoiner<Record, Record, Record>>
+{
+ private static final long HASH_MEM = 6*1024*1024;
+
+ private static final long SORT_MEM = 3*1024*1024;
+
+ @SuppressWarnings("unchecked")
+ private final RecordComparator comparator1 = new RecordComparator(
+ new int[]{0}, (Class<? extends Key<?>>[])new Class[]{ IntValue.class });
+
+ @SuppressWarnings("unchecked")
+ private final RecordComparator comparator2 = new RecordComparator(
+ new int[]{0}, (Class<? extends Key<?>>[])new Class[]{ IntValue.class });
+
+ private final List<Record> outList = new ArrayList<Record>();
+
+
+ public CachedMatchTaskTest() {
+ super(HASH_MEM, 2, SORT_MEM);
+ }
+
+ @Test
+ public void testHash1MatchTask() {
+ int keyCnt1 = 20;
+ int valCnt1 = 1;
+
+ int keyCnt2 = 10;
+ int valCnt2 = 2;
+
+ addInput(new UniformRecordGenerator(keyCnt1, valCnt1, false));
+ addInput(new UniformRecordGenerator(keyCnt2, valCnt2, false));
+ addInputComparator(this.comparator1);
+ addInputComparator(this.comparator2);
+ getTaskConfig().setDriverPairComparator(RecordPairComparatorFactory.get());
+ setOutput(this.outList);
+ getTaskConfig().setDriverStrategy(DriverStrategy.HYBRIDHASH_BUILD_FIRST_CACHED);
+ getTaskConfig().setRelativeMemoryDriver(1.0f);
+
+ BuildFirstCachedMatchDriver<Record, Record, Record> testTask = new BuildFirstCachedMatchDriver<Record, Record, Record>();
+
+ try {
+ testResettableDriver(testTask, MockMatchStub.class, 3);
+ } catch (Exception e) {
+ e.printStackTrace();
+ Assert.fail("Test caused an exception.");
+ }
+
+ final int expCnt = valCnt1*valCnt2*Math.min(keyCnt1, keyCnt2);
+ Assert.assertEquals("Wrong result set size.", expCnt, this.outList.size());
+ this.outList.clear();
+ }
+
+ @Test
+ public void testHash2MatchTask() {
+ int keyCnt1 = 20;
+ int valCnt1 = 1;
+
+ int keyCnt2 = 20;
+ int valCnt2 = 1;
+
+ addInput(new UniformRecordGenerator(keyCnt1, valCnt1, false));
+ addInput(new UniformRecordGenerator(keyCnt2, valCnt2, false));
+ addInputComparator(this.comparator1);
+ addInputComparator(this.comparator2);
+ getTaskConfig().setDriverPairComparator(RecordPairComparatorFactory.get());
+ setOutput(this.outList);
+ getTaskConfig().setDriverStrategy(DriverStrategy.HYBRIDHASH_BUILD_SECOND_CACHED);
+ getTaskConfig().setRelativeMemoryDriver(1.0f);
+
+ BuildSecondCachedMatchDriver<Record, Record, Record> testTask = new BuildSecondCachedMatchDriver<Record, Record, Record>();
+
+ try {
+ testResettableDriver(testTask, MockMatchStub.class, 3);
+ } catch (Exception e) {
+ e.printStackTrace();
+ Assert.fail("Test caused an exception.");
+ }
+
+ final int expCnt = valCnt1*valCnt2*Math.min(keyCnt1, keyCnt2);
+ Assert.assertEquals("Wrong result set size.", expCnt, this.outList.size());
+ this.outList.clear();
+ }
+
+ @Test
+ public void testHash3MatchTask() {
+ int keyCnt1 = 20;
+ int valCnt1 = 1;
+
+ int keyCnt2 = 20;
+ int valCnt2 = 20;
+
+ addInput(new UniformRecordGenerator(keyCnt1, valCnt1, false));
+ addInput(new UniformRecordGenerator(keyCnt2, valCnt2, false));
+ addInputComparator(this.comparator1);
+ addInputComparator(this.comparator2);
+ getTaskConfig().setDriverPairComparator(RecordPairComparatorFactory.get());
+ setOutput(this.outList);
+ getTaskConfig().setDriverStrategy(DriverStrategy.HYBRIDHASH_BUILD_FIRST_CACHED);
+ getTaskConfig().setRelativeMemoryDriver(1.0f);
+
+ BuildFirstCachedMatchDriver<Record, Record, Record> testTask = new BuildFirstCachedMatchDriver<Record, Record, Record>();
+
+ try {
+ testResettableDriver(testTask, MockMatchStub.class, 3);
+ } catch (Exception e) {
+ e.printStackTrace();
+ Assert.fail("Test caused an exception.");
+ }
+
+ final int expCnt = valCnt1*valCnt2*Math.min(keyCnt1, keyCnt2);
+ Assert.assertEquals("Wrong result set size.", expCnt, this.outList.size());
+ this.outList.clear();
+ }
+
+ @Test
+ public void testHash4MatchTask() {
+ int keyCnt1 = 20;
+ int valCnt1 = 20;
+
+ int keyCnt2 = 20;
+ int valCnt2 = 1;
+
+ addInput(new UniformRecordGenerator(keyCnt1, valCnt1, false));
+ addInput(new UniformRecordGenerator(keyCnt2, valCnt2, false));
+ addInputComparator(this.comparator1);
+ addInputComparator(this.comparator2);
+ getTaskConfig().setDriverPairComparator(RecordPairComparatorFactory.get());
+ setOutput(this.outList);
+ getTaskConfig().setDriverStrategy(DriverStrategy.HYBRIDHASH_BUILD_SECOND_CACHED);
+ getTaskConfig().setRelativeMemoryDriver(1.0f);
+
+ BuildSecondCachedMatchDriver<Record, Record, Record> testTask = new BuildSecondCachedMatchDriver<Record, Record, Record>();
+
+ try {
+ testResettableDriver(testTask, MockMatchStub.class, 3);
+ } catch (Exception e) {
+ e.printStackTrace();
+ Assert.fail("Test caused an exception.");
+ }
+
+ final int expCnt = valCnt1*valCnt2*Math.min(keyCnt1, keyCnt2);
+ Assert.assertEquals("Wrong result set size.", expCnt, this.outList.size());
+ this.outList.clear();
+ }
+
+ @Test
+ public void testHash5MatchTask() {
+ int keyCnt1 = 20;
+ int valCnt1 = 20;
+
+ int keyCnt2 = 20;
+ int valCnt2 = 20;
+
+ addInput(new UniformRecordGenerator(keyCnt1, valCnt1, false));
+ addInput(new UniformRecordGenerator(keyCnt2, valCnt2, false));
+ addInputComparator(this.comparator1);
+ addInputComparator(this.comparator2);
+ getTaskConfig().setDriverPairComparator(RecordPairComparatorFactory.get());
+ setOutput(this.outList);
+ getTaskConfig().setDriverStrategy(DriverStrategy.HYBRIDHASH_BUILD_FIRST_CACHED);
+ getTaskConfig().setRelativeMemoryDriver(1.0f);
+
+ BuildFirstCachedMatchDriver<Record, Record, Record> testTask = new BuildFirstCachedMatchDriver<Record, Record, Record>();
+
+ try {
+ testResettableDriver(testTask, MockMatchStub.class, 3);
+ } catch (Exception e) {
+ e.printStackTrace();
+ Assert.fail("Test caused an exception.");
+ }
+
+ final int expCnt = valCnt1*valCnt2*Math.min(keyCnt1, keyCnt2);
+ Assert.assertEquals("Wrong result set size.", expCnt, this.outList.size());
+ this.outList.clear();
+ }
+
+ @Test
+ public void testFailingHashFirstMatchTask() {
+ int keyCnt1 = 20;
+ int valCnt1 = 20;
+
+ int keyCnt2 = 20;
+ int valCnt2 = 20;
+
+ addInput(new UniformRecordGenerator(keyCnt1, valCnt1, false));
+ addInput(new UniformRecordGenerator(keyCnt2, valCnt2, false));
+ addInputComparator(this.comparator1);
+ addInputComparator(this.comparator2);
+ getTaskConfig().setDriverPairComparator(RecordPairComparatorFactory.get());
+ setOutput(new NirvanaOutputList());
+ getTaskConfig().setDriverStrategy(DriverStrategy.HYBRIDHASH_BUILD_FIRST_CACHED);
+ getTaskConfig().setRelativeMemoryDriver(1.0f);
+
+ BuildFirstCachedMatchDriver<Record, Record, Record> testTask = new BuildFirstCachedMatchDriver<Record, Record, Record>();
+
+ try {
+ testResettableDriver(testTask, MockFailingMatchStub.class, 3);
+ Assert.fail("Function exception was not forwarded.");
+ } catch (ExpectedTestException etex) {
+ // good!
+ } catch (Exception e) {
+ e.printStackTrace();
+ Assert.fail("Test caused an exception.");
+ }
+ }
+
+ @Test
+ public void testFailingHashSecondMatchTask() {
+ int keyCnt1 = 20;
+ int valCnt1 = 20;
+
+ int keyCnt2 = 20;
+ int valCnt2 = 20;
+
+ addInput(new UniformRecordGenerator(keyCnt1, valCnt1, false));
+ addInput(new UniformRecordGenerator(keyCnt2, valCnt2, false));
+ addInputComparator(this.comparator1);
+ addInputComparator(this.comparator2);
+ getTaskConfig().setDriverPairComparator(RecordPairComparatorFactory.get());
+ setOutput(new NirvanaOutputList());
+ getTaskConfig().setDriverStrategy(DriverStrategy.HYBRIDHASH_BUILD_SECOND_CACHED);
+ getTaskConfig().setRelativeMemoryDriver(1.0f);
+
+ BuildSecondCachedMatchDriver<Record, Record, Record> testTask = new BuildSecondCachedMatchDriver<Record, Record, Record>();
+
+ try {
+ testResettableDriver(testTask, MockFailingMatchStub.class, 3);
+ Assert.fail("Function exception was not forwarded.");
+ } catch (ExpectedTestException etex) {
+ // good!
+ } catch (Exception e) {
+ e.printStackTrace();
+ Assert.fail("Test caused an exception.");
+ }
+ }
+
+ @Test
+ public void testCancelHashMatchTaskWhileBuildFirst() {
+ int keyCnt = 20;
+ int valCnt = 20;
+
+ addInput(new DelayingInfinitiveInputIterator(100));
+ addInput(new UniformRecordGenerator(keyCnt, valCnt, false));
+
+ addInputComparator(this.comparator1);
+ addInputComparator(this.comparator2);
+
+ getTaskConfig().setDriverPairComparator(RecordPairComparatorFactory.get());
+
+ setOutput(new NirvanaOutputList());
+
+ getTaskConfig().setDriverStrategy(DriverStrategy.HYBRIDHASH_BUILD_FIRST_CACHED);
+ getTaskConfig().setRelativeMemoryDriver(1.0f);
+
+ final BuildFirstCachedMatchDriver<Record, Record, Record> testTask = new BuildFirstCachedMatchDriver<Record, Record, Record>();
+
+ final AtomicBoolean success = new AtomicBoolean(false);
+
+ Thread taskRunner = new Thread() {
+ @Override
+ public void run() {
+ try {
+ testDriver(testTask, MockFailingMatchStub.class);
+ success.set(true);
+ } catch (Exception ie) {
+ ie.printStackTrace();
+ }
+ }
+ };
+ taskRunner.start();
+
+ TaskCancelThread tct = new TaskCancelThread(1, taskRunner, this);
+ tct.start();
+
+ try {
+ tct.join();
+ taskRunner.join();
+ } catch(InterruptedException ie) {
+ Assert.fail("Joining threads failed");
+ }
+
+ Assert.assertTrue("Test threw an exception even though it was properly canceled.", success.get());
+ }
+
+ @Test
+ public void testHashCancelMatchTaskWhileBuildSecond() {
+ int keyCnt = 20;
+ int valCnt = 20;
+
+ addInput(new UniformRecordGenerator(keyCnt, valCnt, false));
+ addInput(new DelayingInfinitiveInputIterator(100));
+ addInputComparator(this.comparator1);
+ addInputComparator(this.comparator2);
+ getTaskConfig().setDriverPairComparator(RecordPairComparatorFactory.get());
+ setOutput(new NirvanaOutputList());
+ getTaskConfig().setDriverStrategy(DriverStrategy.HYBRIDHASH_BUILD_SECOND_CACHED);
+ getTaskConfig().setRelativeMemoryDriver(1.0f);
+
+ final BuildSecondCachedMatchDriver<Record, Record, Record> testTask = new BuildSecondCachedMatchDriver<Record, Record, Record>();
+
+ final AtomicBoolean success = new AtomicBoolean(false);
+
+ Thread taskRunner = new Thread() {
+ @Override
+ public void run() {
+ try {
+ testDriver(testTask, MockMatchStub.class);
+ success.set(true);
+ } catch (Exception ie) {
+ ie.printStackTrace();
+ }
+ }
+ };
+ taskRunner.start();
+
+ TaskCancelThread tct = new TaskCancelThread(1, taskRunner, this);
+ tct.start();
+
+ try {
+ tct.join();
+ taskRunner.join();
+ } catch(InterruptedException ie) {
+ Assert.fail("Joining threads failed");
+ }
+
+ Assert.assertTrue("Test threw an exception even though it was properly canceled.", success.get());
+ }
+
+ @Test
+ public void testHashFirstCancelMatchTaskWhileMatching() {
+ int keyCnt = 20;
+ int valCnt = 20;
+
+ addInput(new UniformRecordGenerator(keyCnt, valCnt, false));
+ addInput(new UniformRecordGenerator(keyCnt, valCnt, false));
+ addInputComparator(this.comparator1);
+ addInputComparator(this.comparator2);
+ getTaskConfig().setDriverPairComparator(RecordPairComparatorFactory.get());
+ setOutput(new NirvanaOutputList());
+ getTaskConfig().setDriverStrategy(DriverStrategy.HYBRIDHASH_BUILD_FIRST);
+ getTaskConfig().setRelativeMemoryDriver(1.0f);
+
+ final BuildFirstCachedMatchDriver<Record, Record, Record> testTask = new BuildFirstCachedMatchDriver<Record, Record, Record>();
+
+ final AtomicBoolean success = new AtomicBoolean(false);
+
+ Thread taskRunner = new Thread() {
+ @Override
+ public void run() {
+ try {
+ testDriver(testTask, MockMatchStub.class);
+ success.set(true);
+ } catch (Exception ie) {
+ ie.printStackTrace();
+ }
+ }
+ };
+ taskRunner.start();
+
+ TaskCancelThread tct = new TaskCancelThread(1, taskRunner, this);
+ tct.start();
+
+ try {
+ tct.join();
+ taskRunner.join();
+ } catch(InterruptedException ie) {
+ Assert.fail("Joining threads failed");
+ }
+
+ Assert.assertTrue("Test threw an exception even though it was properly canceled.", success.get());
+ }
+
+ @Test
+ public void testHashSecondCancelMatchTaskWhileMatching() {
+ int keyCnt = 20;
+ int valCnt = 20;
+
+ addInput(new UniformRecordGenerator(keyCnt, valCnt, false));
+ addInput(new UniformRecordGenerator(keyCnt, valCnt, false));
+ addInputComparator(this.comparator1);
+ addInputComparator(this.comparator2);
+ getTaskConfig().setDriverPairComparator(RecordPairComparatorFactory.get());
+ setOutput(new NirvanaOutputList());
+ getTaskConfig().setDriverStrategy(DriverStrategy.HYBRIDHASH_BUILD_SECOND);
+ getTaskConfig().setRelativeMemoryDriver(1.0f);
+
+ final BuildSecondCachedMatchDriver<Record, Record, Record> testTask = new BuildSecondCachedMatchDriver<Record, Record, Record>();
+
+
+ final AtomicBoolean success = new AtomicBoolean(false);
+
+ Thread taskRunner = new Thread() {
+ @Override
+ public void run() {
+ try {
+ testDriver(testTask, MockMatchStub.class);
+ success.set(true);
+ } catch (Exception ie) {
+ ie.printStackTrace();
+ }
+ }
+ };
+ taskRunner.start();
+
+ TaskCancelThread tct = new TaskCancelThread(1, taskRunner, this);
+ tct.start();
+
+ try {
+ tct.join();
+ taskRunner.join();
+ } catch(InterruptedException ie) {
+ Assert.fail("Joining threads failed");
+ }
+
+ Assert.assertTrue("Test threw an exception even though it was properly canceled.", success.get());
+ }
+
+ // =================================================================================================
+
+ public static final class MockMatchStub extends JoinFunction {
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public void join(Record record1, Record record2, Collector<Record> out) throws Exception {
+ out.collect(record1);
+ }
+ }
+
+ public static final class MockFailingMatchStub extends JoinFunction {
+ private static final long serialVersionUID = 1L;
+
+ private int cnt = 0;
+
+ @Override
+ public void join(Record record1, Record record2, Collector<Record> out) {
+ if (++this.cnt >= 10) {
+ throw new ExpectedTestException();
+ }
+
+ out.collect(record1);
+ }
+ }
+
+ public static final class MockDelayingMatchStub extends JoinFunction {
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public void join(Record record1, Record record2, Collector<Record> out) {
+ try {
+ Thread.sleep(100);
+ } catch (InterruptedException e) { }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/47a68adc/stratosphere-runtime/src/test/java/eu/stratosphere/pact/runtime/test/util/DriverTestBase.java
----------------------------------------------------------------------
diff --git a/stratosphere-runtime/src/test/java/eu/stratosphere/pact/runtime/test/util/DriverTestBase.java b/stratosphere-runtime/src/test/java/eu/stratosphere/pact/runtime/test/util/DriverTestBase.java
index 531382e..4156204 100644
--- a/stratosphere-runtime/src/test/java/eu/stratosphere/pact/runtime/test/util/DriverTestBase.java
+++ b/stratosphere-runtime/src/test/java/eu/stratosphere/pact/runtime/test/util/DriverTestBase.java
@@ -35,6 +35,7 @@ import eu.stratosphere.api.java.typeutils.runtime.record.RecordSerializerFactory
import eu.stratosphere.pact.runtime.sort.UnilateralSortMerger;
import eu.stratosphere.pact.runtime.task.PactDriver;
import eu.stratosphere.pact.runtime.task.PactTaskContext;
+import eu.stratosphere.pact.runtime.task.ResettablePactDriver;
import eu.stratosphere.pact.runtime.task.util.TaskConfig;
import eu.stratosphere.types.Record;
import eu.stratosphere.util.Collector;
@@ -194,16 +195,48 @@ public class DriverTestBase<S extends Function> implements PactTaskContext<S, Re
catch (Throwable t) {}
}
+ // if resettable driver invoke treardown
+ if (this.driver instanceof ResettablePactDriver) {
+ final ResettablePactDriver<?, ?> resDriver = (ResettablePactDriver<?, ?>) this.driver;
+ try {
+ resDriver.teardown();
+ } catch (Throwable t) {
+ throw new Exception("Error while shutting down an iterative operator: " + t.getMessage(), t);
+ }
+ }
+
// drop exception, if the task was canceled
if (this.running) {
throw ex;
}
+
}
finally {
driver.cleanup();
}
}
+ @SuppressWarnings({"unchecked","rawtypes"})
+ public void testResettableDriver(ResettablePactDriver driver, Class stubClass, int iterations) throws Exception {
+
+ driver.setup(this);
+
+ for(int i = 0; i < iterations; i++) {
+
+ if(i == 0) {
+ driver.initialize();
+ }
+ else {
+ driver.reset();
+ }
+
+ testDriver(driver, stubClass);
+
+ }
+
+ driver.teardown();
+ }
+
public void cancel() throws Exception {
this.running = false;
this.driver.cancel();
[08/12] git commit: Fix cost formulas for cached hash joins Fix
estimation bug in channels Add tests for isolated cost formulas Add tests for
channel class in plan candidates
Posted by se...@apache.org.
Fix cost formulas for cached hash joins
Fix estimation bug in channels
Add tests for isolated cost formulas
Add tests for channel class in plan candidates
Project: http://git-wip-us.apache.org/repos/asf/incubator-flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-flink/commit/03c6160e
Tree: http://git-wip-us.apache.org/repos/asf/incubator-flink/tree/03c6160e
Diff: http://git-wip-us.apache.org/repos/asf/incubator-flink/diff/03c6160e
Branch: refs/heads/master
Commit: 03c6160e47bc2ab613a2734388aa22f83323e6a5
Parents: ce65fb6
Author: Stephan Ewen <se...@apache.org>
Authored: Wed Jul 9 03:03:21 2014 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Wed Jul 9 03:09:48 2014 +0200
----------------------------------------------------------------------
.../compiler/costs/CostEstimator.java | 20 +-
.../compiler/costs/DefaultCostEstimator.java | 38 +-
.../compiler/dag/OptimizerNode.java | 8 +
.../eu/stratosphere/compiler/plan/Channel.java | 6 +-
.../costs/DefaultCostEstimatorTest.java | 425 +++++++++++++++++++
.../stratosphere/compiler/plan/ChannelTest.java | 87 ++++
6 files changed, 559 insertions(+), 25 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/03c6160e/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/costs/CostEstimator.java
----------------------------------------------------------------------
diff --git a/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/costs/CostEstimator.java b/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/costs/CostEstimator.java
index 3f0c094..e45ed69 100644
--- a/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/costs/CostEstimator.java
+++ b/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/costs/CostEstimator.java
@@ -40,13 +40,13 @@ public abstract class CostEstimator {
public abstract void addFileInputCost(long fileSizeInBytes, Costs costs);
- public abstract void addLocalSortCost(EstimateProvider estimates, long memorySize, Costs costs);
+ public abstract void addLocalSortCost(EstimateProvider estimates, Costs costs);
- public abstract void addLocalMergeCost(EstimateProvider estimates1, EstimateProvider estimates2, long memorySize, Costs costs, int costWeight);
+ public abstract void addLocalMergeCost(EstimateProvider estimates1, EstimateProvider estimates2, Costs costs, int costWeight);
- public abstract void addHybridHashCosts(EstimateProvider buildSide, EstimateProvider probeSide, long memorySize, Costs costs, int costWeight);
+ public abstract void addHybridHashCosts(EstimateProvider buildSide, EstimateProvider probeSide, Costs costs, int costWeight);
- public abstract void addCachedHybridHashCosts(EstimateProvider buildSide, EstimateProvider probeSide, long memorySize, Costs costs, int costWeight);
+ public abstract void addCachedHybridHashCosts(EstimateProvider buildSide, EstimateProvider probeSide, Costs costs, int costWeight);
public abstract void addStreamedNestedLoopsCosts(EstimateProvider outerSide, EstimateProvider innerSide, long bufferSize, Costs costs, int costWeight);
@@ -116,7 +116,7 @@ public abstract class CostEstimator {
break;
case SORT:
case COMBININGSORT:
- addLocalSortCost(channel, availableMemory, costs);
+ addLocalSortCost(channel, costs);
break;
default:
throw new CompilerException("Unsupported local strategy for input: " + channel.getLocalStrategy());
@@ -181,19 +181,19 @@ public abstract class CostEstimator {
break;
case MERGE:
- addLocalMergeCost(firstInput, secondInput, availableMemory, driverCosts, costWeight);
+ addLocalMergeCost(firstInput, secondInput, driverCosts, costWeight);
break;
case HYBRIDHASH_BUILD_FIRST:
- addHybridHashCosts(firstInput, secondInput, availableMemory, driverCosts, costWeight);
+ addHybridHashCosts(firstInput, secondInput, driverCosts, costWeight);
break;
case HYBRIDHASH_BUILD_SECOND:
- addHybridHashCosts(secondInput, firstInput, availableMemory, driverCosts, costWeight);
+ addHybridHashCosts(secondInput, firstInput, driverCosts, costWeight);
break;
case HYBRIDHASH_BUILD_FIRST_CACHED:
- addCachedHybridHashCosts(firstInput, secondInput, availableMemory, driverCosts, costWeight);
+ addCachedHybridHashCosts(firstInput, secondInput, driverCosts, costWeight);
break;
case HYBRIDHASH_BUILD_SECOND_CACHED:
- addCachedHybridHashCosts(secondInput, firstInput, availableMemory, driverCosts, costWeight);
+ addCachedHybridHashCosts(secondInput, firstInput, driverCosts, costWeight);
break;
case NESTEDLOOP_BLOCKED_OUTER_FIRST:
addBlockNestedLoopsCosts(firstInput, secondInput, availableMemory, driverCosts, costWeight);
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/03c6160e/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/costs/DefaultCostEstimator.java
----------------------------------------------------------------------
diff --git a/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/costs/DefaultCostEstimator.java b/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/costs/DefaultCostEstimator.java
index 425f93c..93eb66a 100644
--- a/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/costs/DefaultCostEstimator.java
+++ b/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/costs/DefaultCostEstimator.java
@@ -96,6 +96,9 @@ public class DefaultCostEstimator extends CostEstimator {
@Override
public void addBroadcastCost(EstimateProvider estimates, int replicationFactor, Costs costs) {
// if our replication factor is negative, we cannot calculate broadcast costs
+ if (replicationFactor <= 0) {
+ throw new IllegalArgumentException("The replication factor of must be larger than zero.");
+ }
if (replicationFactor > 0) {
// assumption: we need ship the whole data over the network to each node.
@@ -126,7 +129,7 @@ public class DefaultCostEstimator extends CostEstimator {
}
@Override
- public void addLocalSortCost(EstimateProvider estimates, long availableMemory, Costs costs) {
+ public void addLocalSortCost(EstimateProvider estimates, Costs costs) {
final long s = estimates.getEstimatedOutputSize();
// we assume a two phase merge sort, so all in all 2 I/O operations per block
if (s <= 0) {
@@ -141,25 +144,28 @@ public class DefaultCostEstimator extends CostEstimator {
}
@Override
- public void addLocalMergeCost(EstimateProvider input1, EstimateProvider input2, long availableMemory, Costs costs, int costWeight) {
+ public void addLocalMergeCost(EstimateProvider input1, EstimateProvider input2, Costs costs, int costWeight) {
// costs nothing. the very rarely incurred cost for a spilling block nested loops join in the
- // presence of massively re-occurring duplicate keys is ignored, because not accessible.
+ // presence of massively re-occurring duplicate keys is ignored, because cannot be assessed
}
@Override
- public void addHybridHashCosts(EstimateProvider buildSideInput, EstimateProvider probeSideInput, long availableMemory, Costs costs, int costWeight) {
+ public void addHybridHashCosts(EstimateProvider buildSideInput, EstimateProvider probeSideInput, Costs costs, int costWeight) {
long bs = buildSideInput.getEstimatedOutputSize();
long ps = probeSideInput.getEstimatedOutputSize();
if (bs > 0 && ps > 0) {
- costs.addDiskCost(2*bs + ps);
- costs.addCpuCost((long) ((2*bs + ps) * HASHING_CPU_FACTOR));
+ long overall = 2*bs + ps;
+ costs.addDiskCost(overall);
+ costs.addCpuCost((long) (overall * HASHING_CPU_FACTOR));
} else {
costs.setDiskCost(Costs.UNKNOWN);
costs.setCpuCost(Costs.UNKNOWN);
}
costs.addHeuristicDiskCost(2 * HEURISTIC_COST_BASE);
- costs.addHeuristicCpuCost((long) (HEURISTIC_COST_BASE * HASHING_CPU_FACTOR));
+ costs.addHeuristicCpuCost((long) (2 * HEURISTIC_COST_BASE * HASHING_CPU_FACTOR));
+
+ // cost weight applies to everything
costs.multiplyWith(costWeight);
}
@@ -168,20 +174,26 @@ public class DefaultCostEstimator extends CostEstimator {
* We are assuming by default that half of the cached hash table fit into memory.
*/
@Override
- public void addCachedHybridHashCosts(EstimateProvider buildSideInput, EstimateProvider probeSideInput, long availableMemory, Costs costs, int costWeight) {
+ public void addCachedHybridHashCosts(EstimateProvider buildSideInput, EstimateProvider probeSideInput, Costs costs, int costWeight) {
+ if (costWeight < 1) {
+ throw new IllegalArgumentException("The cost weight must be at least one.");
+ }
+
long bs = buildSideInput.getEstimatedOutputSize();
long ps = probeSideInput.getEstimatedOutputSize();
if (bs > 0 && ps > 0) {
- long overallSize = 2*bs + ps;
- costs.addDiskCost(overallSize / 2 + (overallSize / 2) * costWeight);
- costs.addCpuCost((long) ((2*bs + ps) * HASHING_CPU_FACTOR));
+ long overall = 2*bs + costWeight*ps;
+ costs.addDiskCost(overall);
+ costs.addCpuCost((long) (overall * HASHING_CPU_FACTOR));
} else {
costs.setDiskCost(Costs.UNKNOWN);
costs.setCpuCost(Costs.UNKNOWN);
}
- costs.addHeuristicDiskCost(2 * HEURISTIC_COST_BASE * costWeight);
- costs.addHeuristicCpuCost((long) (HEURISTIC_COST_BASE * HASHING_CPU_FACTOR * costWeight));
+
+ // one time the build side plus cost-weight time the probe side
+ costs.addHeuristicDiskCost((1 + costWeight) * HEURISTIC_COST_BASE);
+ costs.addHeuristicCpuCost((long) ((1 + costWeight) * HEURISTIC_COST_BASE * HASHING_CPU_FACTOR));
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/03c6160e/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/dag/OptimizerNode.java
----------------------------------------------------------------------
diff --git a/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/dag/OptimizerNode.java b/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/dag/OptimizerNode.java
index ec9bd69..88592e8 100644
--- a/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/dag/OptimizerNode.java
+++ b/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/dag/OptimizerNode.java
@@ -491,6 +491,14 @@ public abstract class OptimizerNode implements Visitable<OptimizerNode>, Estimat
return this.estimatedNumRecords;
}
+ public void setEstimatedOutputSize(long estimatedOutputSize) {
+ this.estimatedOutputSize = estimatedOutputSize;
+ }
+
+ public void setEstimatedNumRecords(long estimatedNumRecords) {
+ this.estimatedNumRecords = estimatedNumRecords;
+ }
+
public float getEstimatedAvgWidthPerOutputRecord() {
if (this.estimatedOutputSize > 0 && this.estimatedNumRecords > 0) {
return ((float) this.estimatedOutputSize) / this.estimatedNumRecords;
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/03c6160e/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/plan/Channel.java
----------------------------------------------------------------------
diff --git a/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/plan/Channel.java b/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/plan/Channel.java
index 6f9418f..c604a45 100644
--- a/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/plan/Channel.java
+++ b/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/plan/Channel.java
@@ -317,12 +317,14 @@ public class Channel implements EstimateProvider, Cloneable, DumpableConnection<
@Override
public long getEstimatedOutputSize() {
- return this.source.template.getEstimatedOutputSize() * this.replicationFactor;
+ long estimate = this.source.template.getEstimatedOutputSize();
+ return estimate < 0 ? estimate : estimate * this.replicationFactor;
}
@Override
public long getEstimatedNumRecords() {
- return this.source.template.getEstimatedNumRecords() * this.replicationFactor;
+ long estimate = this.source.template.getEstimatedNumRecords();
+ return estimate < 0 ? estimate : estimate * this.replicationFactor;
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/03c6160e/stratosphere-compiler/src/test/java/eu/stratosphere/compiler/costs/DefaultCostEstimatorTest.java
----------------------------------------------------------------------
diff --git a/stratosphere-compiler/src/test/java/eu/stratosphere/compiler/costs/DefaultCostEstimatorTest.java b/stratosphere-compiler/src/test/java/eu/stratosphere/compiler/costs/DefaultCostEstimatorTest.java
new file mode 100644
index 0000000..3b85ff4
--- /dev/null
+++ b/stratosphere-compiler/src/test/java/eu/stratosphere/compiler/costs/DefaultCostEstimatorTest.java
@@ -0,0 +1,425 @@
+/***********************************************************************************************************************
+ *
+ * Copyright (C) 2010-2013 by the Stratosphere project (http://stratosphere.eu)
+ *
+ * Licensed 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 eu.stratosphere.compiler.costs;
+
+import static org.junit.Assert.*;
+
+import org.junit.Test;
+
+import eu.stratosphere.compiler.dag.EstimateProvider;
+
+/**
+ * Tests for the cost formulas in the {@link DefaultCostEstimator}. Most of the tests establish relative
+ * relationships.
+ */
+public class DefaultCostEstimatorTest {
+
+ // estimates
+
+ private static final long SMALL_DATA_SIZE = 10000;
+ private static final long SMALL_RECORD_COUNT = 100;
+
+ private static final long MEDIUM_DATA_SIZE = 500000000L;
+ private static final long MEDIUM_RECORD_COUNT = 500000L;
+
+ private static final long BIG_DATA_SIZE = 100000000000L;
+ private static final long BIG_RECORD_COUNT = 100000000L;
+
+ private static final EstimateProvider UNKNOWN_ESTIMATES = new UnknownEstimates();
+ private static final EstimateProvider ZERO_ESTIMATES = new Estimates(0, 0);
+ private static final EstimateProvider SMALL_ESTIMATES = new Estimates(SMALL_DATA_SIZE, SMALL_RECORD_COUNT);
+ private static final EstimateProvider MEDIUM_ESTIMATES = new Estimates(MEDIUM_DATA_SIZE, MEDIUM_RECORD_COUNT);
+ private static final EstimateProvider BIG_ESTIMATES = new Estimates(BIG_DATA_SIZE, BIG_RECORD_COUNT);
+
+ private final CostEstimator costEstimator = new DefaultCostEstimator();
+
+ // --------------------------------------------------------------------------------------------
+
+ @Test
+ public void testShipStrategiesIsolated() {
+ testShipStrategiesIsolated(UNKNOWN_ESTIMATES, 1);
+ testShipStrategiesIsolated(UNKNOWN_ESTIMATES, 10);
+ testShipStrategiesIsolated(ZERO_ESTIMATES, 1);
+ testShipStrategiesIsolated(ZERO_ESTIMATES, 10);
+ testShipStrategiesIsolated(SMALL_ESTIMATES, 1);
+ testShipStrategiesIsolated(SMALL_ESTIMATES, 10);
+ testShipStrategiesIsolated(BIG_ESTIMATES, 1);
+ testShipStrategiesIsolated(BIG_ESTIMATES, 10);
+ }
+
+ private void testShipStrategiesIsolated(EstimateProvider estimates, int targetParallelism) {
+ Costs random = new Costs();
+ costEstimator.addRandomPartitioningCost(estimates, random);
+
+ Costs hash = new Costs();
+ costEstimator.addHashPartitioningCost(estimates, hash);
+
+ Costs range = new Costs();
+ costEstimator.addRangePartitionCost(estimates, range);
+
+ Costs broadcast = new Costs();
+ costEstimator.addBroadcastCost(estimates, targetParallelism, broadcast);
+
+ int randomVsHash = random.compareTo(hash);
+ int hashVsRange = hash.compareTo(range);
+ int hashVsBroadcast = hash.compareTo(broadcast);
+ int rangeVsBroadcast = range.compareTo(broadcast);
+
+ // repartition random is at most as expensive as hash partitioning
+ assertTrue(randomVsHash <= 0);
+
+ // range partitioning is always more expensive than hash partitioning
+ assertTrue(hashVsRange < 0);
+
+ // broadcasting is always more expensive than hash partitioning
+ if (targetParallelism > 1) {
+ assertTrue(hashVsBroadcast < 0);
+ } else {
+ assertTrue(hashVsBroadcast <= 0);
+ }
+
+ // range partitioning is not more expensive than broadcasting
+ if (targetParallelism > 1) {
+ assertTrue(rangeVsBroadcast < 0);
+ }
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ @Test
+ public void testShipStrategyCombinationsPlain() {
+ Costs hashBothSmall = new Costs();
+ Costs hashSmallAndLarge = new Costs();
+ Costs hashBothLarge = new Costs();
+
+ Costs hashSmallBcLarge10 = new Costs();
+ Costs hashLargeBcSmall10 = new Costs();
+
+ Costs hashSmallBcLarge1000 = new Costs();
+ Costs hashLargeBcSmall1000 = new Costs();
+
+ Costs forwardSmallBcLarge10 = new Costs();
+ Costs forwardLargeBcSmall10 = new Costs();
+
+ Costs forwardSmallBcLarge1000 = new Costs();
+ Costs forwardLargeBcSmall1000 = new Costs();
+
+ costEstimator.addHashPartitioningCost(MEDIUM_ESTIMATES, hashBothSmall);
+ costEstimator.addHashPartitioningCost(MEDIUM_ESTIMATES, hashBothSmall);
+
+ costEstimator.addHashPartitioningCost(MEDIUM_ESTIMATES, hashSmallAndLarge);
+ costEstimator.addHashPartitioningCost(BIG_ESTIMATES, hashSmallAndLarge);
+
+ costEstimator.addHashPartitioningCost(BIG_ESTIMATES, hashBothLarge);
+ costEstimator.addHashPartitioningCost(BIG_ESTIMATES, hashBothLarge);
+
+ costEstimator.addHashPartitioningCost(MEDIUM_ESTIMATES, hashSmallBcLarge10);
+ costEstimator.addBroadcastCost(BIG_ESTIMATES, 10, hashSmallBcLarge10);
+
+ costEstimator.addHashPartitioningCost(BIG_ESTIMATES, hashLargeBcSmall10);
+ costEstimator.addBroadcastCost(MEDIUM_ESTIMATES, 10, hashLargeBcSmall10);
+
+ costEstimator.addHashPartitioningCost(MEDIUM_ESTIMATES, hashSmallBcLarge1000);
+ costEstimator.addBroadcastCost(BIG_ESTIMATES, 1000, hashSmallBcLarge1000);
+
+ costEstimator.addHashPartitioningCost(BIG_ESTIMATES, hashLargeBcSmall1000);
+ costEstimator.addBroadcastCost(MEDIUM_ESTIMATES, 1000, hashLargeBcSmall1000);
+
+ costEstimator.addBroadcastCost(BIG_ESTIMATES, 10, forwardSmallBcLarge10);
+
+ costEstimator.addBroadcastCost(MEDIUM_ESTIMATES, 10, forwardLargeBcSmall10);
+
+ costEstimator.addBroadcastCost(BIG_ESTIMATES, 1000, forwardSmallBcLarge1000);
+
+ costEstimator.addBroadcastCost(MEDIUM_ESTIMATES, 1000, forwardLargeBcSmall1000);
+
+ // hash cost is roughly monotonous
+ assertTrue(hashBothSmall.compareTo(hashSmallAndLarge) < 0);
+ assertTrue(hashSmallAndLarge.compareTo(hashBothLarge) < 0);
+
+ // broadcast the smaller is better
+ assertTrue(hashLargeBcSmall10.compareTo(hashSmallBcLarge10) < 0);
+ assertTrue(forwardLargeBcSmall10.compareTo(forwardSmallBcLarge10) < 0);
+ assertTrue(hashLargeBcSmall1000.compareTo(hashSmallBcLarge1000) < 0);
+ assertTrue(forwardLargeBcSmall1000.compareTo(forwardSmallBcLarge1000) < 0);
+
+ // broadcasting small and forwarding large is better than partition both, given size difference
+ assertTrue(forwardLargeBcSmall10.compareTo(hashSmallAndLarge) < 0);
+
+ // broadcasting too far is expensive again
+ assertTrue(forwardLargeBcSmall1000.compareTo(hashSmallAndLarge) > 0);
+
+ // assert weight is respected
+ assertTrue(hashSmallBcLarge10.compareTo(hashSmallBcLarge1000) < 0);
+ assertTrue(hashLargeBcSmall10.compareTo(hashLargeBcSmall1000) < 0);
+ assertTrue(forwardSmallBcLarge10.compareTo(forwardSmallBcLarge1000) < 0);
+ assertTrue(forwardLargeBcSmall10.compareTo(forwardLargeBcSmall1000) < 0);
+
+ // forward versus hash
+ assertTrue(forwardSmallBcLarge10.compareTo(hashSmallBcLarge10) < 0);
+ assertTrue(forwardSmallBcLarge1000.compareTo(hashSmallBcLarge1000) < 0);
+ assertTrue(forwardLargeBcSmall10.compareTo(hashLargeBcSmall10) < 0);
+ assertTrue(forwardLargeBcSmall1000.compareTo(hashLargeBcSmall1000) < 0);
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ @Test
+ public void testShipStrategyCombinationsWithUnknowns() {
+ testShipStrategyCombinationsWithUnknowns(UNKNOWN_ESTIMATES);
+ testShipStrategyCombinationsWithUnknowns(ZERO_ESTIMATES);
+ testShipStrategyCombinationsWithUnknowns(SMALL_ESTIMATES);
+ testShipStrategyCombinationsWithUnknowns(MEDIUM_ESTIMATES);
+ testShipStrategyCombinationsWithUnknowns(BIG_ESTIMATES);
+ }
+
+ private void testShipStrategyCombinationsWithUnknowns(EstimateProvider knownEstimates) {
+ Costs hashBoth = new Costs();
+ Costs bcKnown10 = new Costs();
+ Costs bcUnknown10 = new Costs();
+ Costs bcKnown1000 = new Costs();
+ Costs bcUnknown1000 = new Costs();
+
+ costEstimator.addHashPartitioningCost(knownEstimates, hashBoth);
+ costEstimator.addHashPartitioningCost(UNKNOWN_ESTIMATES, hashBoth);
+
+ costEstimator.addBroadcastCost(knownEstimates, 10, bcKnown10);
+
+ costEstimator.addBroadcastCost(UNKNOWN_ESTIMATES, 10, bcUnknown10);
+
+ costEstimator.addBroadcastCost(knownEstimates, 1000, bcKnown1000);
+
+ costEstimator.addBroadcastCost(UNKNOWN_ESTIMATES, 1000, bcUnknown1000);
+
+ // if we do not know one of them, hashing both should be cheaper than anything
+ assertTrue(hashBoth.compareTo(bcKnown10) < 0);
+ assertTrue(hashBoth.compareTo(bcUnknown10) < 0);
+ assertTrue(hashBoth.compareTo(bcKnown1000) < 0);
+ assertTrue(hashBoth.compareTo(bcUnknown1000) < 0);
+
+ // there should be no bias in broadcasting a known or unknown size input
+ assertTrue(bcKnown10.compareTo(bcUnknown10) == 0);
+ assertTrue(bcKnown1000.compareTo(bcUnknown1000) == 0);
+
+ // replication factor does matter
+ assertTrue(bcKnown10.compareTo(bcKnown1000) < 0);
+ assertTrue(bcUnknown10.compareTo(bcUnknown1000) < 0);
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ @Test
+ public void testJoinCostFormulasPlain() {
+
+ // hash join costs
+
+ Costs hashBothSmall = new Costs();
+ Costs hashBothLarge = new Costs();
+ Costs hashSmallBuild = new Costs();
+ Costs hashLargeBuild = new Costs();
+
+ costEstimator.addHybridHashCosts(SMALL_ESTIMATES, BIG_ESTIMATES, hashSmallBuild, 1);
+ costEstimator.addHybridHashCosts(BIG_ESTIMATES, SMALL_ESTIMATES, hashLargeBuild, 1);
+ costEstimator.addHybridHashCosts(SMALL_ESTIMATES, SMALL_ESTIMATES, hashBothSmall, 1);
+ costEstimator.addHybridHashCosts(BIG_ESTIMATES, BIG_ESTIMATES, hashBothLarge, 1);
+
+ assertTrue(hashBothSmall.compareTo(hashSmallBuild) < 0);
+ assertTrue(hashSmallBuild.compareTo(hashLargeBuild) < 0);
+ assertTrue(hashLargeBuild.compareTo(hashBothLarge) < 0);
+
+ // merge join costs
+
+ Costs mergeBothSmall = new Costs();
+ Costs mergeBothLarge = new Costs();
+ Costs mergeSmallFirst = new Costs();
+ Costs mergeSmallSecond = new Costs();
+
+ costEstimator.addLocalSortCost(SMALL_ESTIMATES, mergeSmallFirst);
+ costEstimator.addLocalSortCost(BIG_ESTIMATES, mergeSmallFirst);
+ costEstimator.addLocalMergeCost(SMALL_ESTIMATES, BIG_ESTIMATES, mergeSmallFirst, 1);
+
+ costEstimator.addLocalSortCost(BIG_ESTIMATES, mergeSmallSecond);
+ costEstimator.addLocalSortCost(SMALL_ESTIMATES, mergeSmallSecond);
+ costEstimator.addLocalMergeCost(BIG_ESTIMATES, SMALL_ESTIMATES, mergeSmallSecond, 1);
+
+ costEstimator.addLocalSortCost(SMALL_ESTIMATES, mergeBothSmall);
+ costEstimator.addLocalSortCost(SMALL_ESTIMATES, mergeBothSmall);
+ costEstimator.addLocalMergeCost(SMALL_ESTIMATES, SMALL_ESTIMATES, mergeBothSmall, 1);
+
+ costEstimator.addLocalSortCost(BIG_ESTIMATES, mergeBothLarge);
+ costEstimator.addLocalSortCost(BIG_ESTIMATES, mergeBothLarge);
+ costEstimator.addLocalMergeCost(BIG_ESTIMATES, BIG_ESTIMATES, mergeBothLarge, 1);
+
+
+ assertTrue(mergeBothSmall.compareTo(mergeSmallFirst) < 0);
+ assertTrue(mergeBothSmall.compareTo(mergeSmallSecond) < 0);
+ assertTrue(mergeSmallFirst.compareTo(mergeSmallSecond) == 0);
+ assertTrue(mergeSmallFirst.compareTo(mergeBothLarge) < 0);
+ assertTrue(mergeSmallSecond.compareTo(mergeBothLarge) < 0);
+
+ // compare merge join and hash join costs
+
+ assertTrue(hashBothSmall.compareTo(mergeBothSmall) < 0);
+ assertTrue(hashBothLarge.compareTo(mergeBothLarge) < 0);
+ assertTrue(hashSmallBuild.compareTo(mergeSmallFirst) < 0);
+ assertTrue(hashSmallBuild.compareTo(mergeSmallSecond) < 0);
+ assertTrue(hashLargeBuild.compareTo(mergeSmallFirst) < 0);
+ assertTrue(hashLargeBuild.compareTo(mergeSmallSecond) < 0);
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ @Test
+ public void testJoinCostFormulasWithWeights() {
+ testJoinCostFormulasWithWeights(UNKNOWN_ESTIMATES, SMALL_ESTIMATES);
+ testJoinCostFormulasWithWeights(SMALL_ESTIMATES, UNKNOWN_ESTIMATES);
+ testJoinCostFormulasWithWeights(UNKNOWN_ESTIMATES, MEDIUM_ESTIMATES);
+ testJoinCostFormulasWithWeights(MEDIUM_ESTIMATES, UNKNOWN_ESTIMATES);
+ testJoinCostFormulasWithWeights(BIG_ESTIMATES, MEDIUM_ESTIMATES);
+ testJoinCostFormulasWithWeights(MEDIUM_ESTIMATES, BIG_ESTIMATES);
+ }
+
+ private void testJoinCostFormulasWithWeights(EstimateProvider e1, EstimateProvider e2) {
+ Costs hf1 = new Costs();
+ Costs hf5 = new Costs();
+ Costs hs1 = new Costs();
+ Costs hs5 = new Costs();
+ Costs mm1 = new Costs();
+ Costs mm5 = new Costs();
+
+ costEstimator.addHybridHashCosts(e1, e2, hf1, 1);
+ costEstimator.addHybridHashCosts(e1, e2, hf5, 5);
+ costEstimator.addHybridHashCosts(e2, e1, hs1, 1);
+ costEstimator.addHybridHashCosts(e2, e1, hs5, 5);
+
+ costEstimator.addLocalSortCost(e1, mm1);
+ costEstimator.addLocalSortCost(e2, mm1);
+ costEstimator.addLocalMergeCost(e1, e2, mm1, 1);
+
+ costEstimator.addLocalSortCost(e1, mm5);
+ costEstimator.addLocalSortCost(e2, mm5);
+ mm5.multiplyWith(5);
+ costEstimator.addLocalMergeCost(e1, e2, mm5, 5);
+
+ // weight 1 versus weight 5
+ assertTrue(hf1.compareTo(hf5) < 0);
+ assertTrue(hs1.compareTo(hs5) < 0);
+ assertTrue(mm1.compareTo(mm5) < 0);
+
+ // hash versus merge
+ assertTrue(hf1.compareTo(mm1) < 0);
+ assertTrue(hs1.compareTo(mm1) < 0);
+ assertTrue(hf5.compareTo(mm5) < 0);
+ assertTrue(hs5.compareTo(mm5) < 0);
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ @Test
+ public void testHashJoinCostFormulasWithCaches() {
+
+ Costs hashBothUnknown10 = new Costs();
+ Costs hashBothUnknownCached10 = new Costs();
+
+ Costs hashBothSmall10 = new Costs();
+ Costs hashBothSmallCached10 = new Costs();
+
+ Costs hashSmallLarge10 = new Costs();
+ Costs hashSmallLargeCached10 = new Costs();
+
+ Costs hashLargeSmall10 = new Costs();
+ Costs hashLargeSmallCached10 = new Costs();
+
+ Costs hashLargeSmall1 = new Costs();
+ Costs hashLargeSmallCached1 = new Costs();
+
+ costEstimator.addHybridHashCosts(UNKNOWN_ESTIMATES, UNKNOWN_ESTIMATES, hashBothUnknown10, 10);
+ costEstimator.addCachedHybridHashCosts(UNKNOWN_ESTIMATES, UNKNOWN_ESTIMATES, hashBothUnknownCached10, 10);
+
+ costEstimator.addHybridHashCosts(MEDIUM_ESTIMATES, MEDIUM_ESTIMATES, hashBothSmall10, 10);
+ costEstimator.addCachedHybridHashCosts(MEDIUM_ESTIMATES, MEDIUM_ESTIMATES, hashBothSmallCached10, 10);
+
+ costEstimator.addHybridHashCosts(MEDIUM_ESTIMATES, BIG_ESTIMATES, hashSmallLarge10, 10);
+ costEstimator.addCachedHybridHashCosts(MEDIUM_ESTIMATES, BIG_ESTIMATES, hashSmallLargeCached10, 10);
+
+ costEstimator.addHybridHashCosts(BIG_ESTIMATES, MEDIUM_ESTIMATES, hashLargeSmall10, 10);
+ costEstimator.addCachedHybridHashCosts(BIG_ESTIMATES, MEDIUM_ESTIMATES, hashLargeSmallCached10, 10);
+
+ costEstimator.addHybridHashCosts(BIG_ESTIMATES, MEDIUM_ESTIMATES, hashLargeSmall1, 1);
+ costEstimator.addCachedHybridHashCosts(BIG_ESTIMATES, MEDIUM_ESTIMATES, hashLargeSmallCached1, 1);
+
+ // cached variant is always cheaper
+ assertTrue(hashBothUnknown10.compareTo(hashBothUnknownCached10) > 0);
+ assertTrue(hashBothSmall10.compareTo(hashBothSmallCached10) > 0);
+ assertTrue(hashSmallLarge10.compareTo(hashSmallLargeCached10) > 0);
+ assertTrue(hashLargeSmall10.compareTo(hashLargeSmallCached10) > 0);
+
+ // caching the large side is better, because then the small one is the one with additional I/O
+ assertTrue(hashLargeSmallCached10.compareTo(hashSmallLargeCached10) < 0);
+
+ // a weight of one makes the caching the same as the non-cached variant
+ assertTrue(hashLargeSmall1.compareTo(hashLargeSmallCached1) == 0);
+ }
+
+
+ // --------------------------------------------------------------------------------------------
+ // Estimate providers
+ // --------------------------------------------------------------------------------------------
+
+ private static final class UnknownEstimates implements EstimateProvider {
+
+ @Override
+ public long getEstimatedOutputSize() { return -1; }
+
+ @Override
+ public long getEstimatedNumRecords() { return -1; }
+
+ @Override
+ public float getEstimatedAvgWidthPerOutputRecord() { return -1.0f; }
+ }
+
+ private static final class Estimates implements EstimateProvider {
+
+ private final long size;
+ private final long records;
+ private final float width;
+
+ public Estimates(long size, long records) {
+ this(size, records, -1.0f);
+ }
+
+ public Estimates(long size, long records, float width) {
+ this.size = size;
+ this.records = records;
+ this.width = width;
+ }
+
+ @Override
+ public long getEstimatedOutputSize() {
+ return this.size;
+ }
+
+ @Override
+ public long getEstimatedNumRecords() {
+ return this.records;
+ }
+
+ @Override
+ public float getEstimatedAvgWidthPerOutputRecord() {
+ return this.width;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/03c6160e/stratosphere-compiler/src/test/java/eu/stratosphere/compiler/plan/ChannelTest.java
----------------------------------------------------------------------
diff --git a/stratosphere-compiler/src/test/java/eu/stratosphere/compiler/plan/ChannelTest.java b/stratosphere-compiler/src/test/java/eu/stratosphere/compiler/plan/ChannelTest.java
new file mode 100644
index 0000000..f3af5d7
--- /dev/null
+++ b/stratosphere-compiler/src/test/java/eu/stratosphere/compiler/plan/ChannelTest.java
@@ -0,0 +1,87 @@
+/***********************************************************************************************************************
+ *
+ * Copyright (C) 2010-2013 by the Stratosphere project (http://stratosphere.eu)
+ *
+ * Licensed 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 eu.stratosphere.compiler.plan;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import eu.stratosphere.api.common.operators.OperatorInformation;
+import eu.stratosphere.api.common.operators.base.GenericDataSourceBase;
+import eu.stratosphere.api.java.io.TextInputFormat;
+import eu.stratosphere.api.java.typeutils.BasicTypeInfo;
+import eu.stratosphere.compiler.dag.DataSourceNode;
+import eu.stratosphere.core.fs.Path;
+
+public class ChannelTest {
+
+ @Test
+ public void testGetEstimatesNoReplicationFactor() {
+ final long NUM_RECORD = 1001;
+ final long SIZE = 467131;
+
+ DataSourceNode source = getSourceNode();
+ SourcePlanNode planNode = new SourcePlanNode(source, "test node");
+ Channel channel = new Channel(planNode);
+
+ // no estimates here
+ Assert.assertEquals(-1, channel.getEstimatedOutputSize());
+ Assert.assertEquals(-1, channel.getEstimatedNumRecords());
+
+ // set estimates
+ source.setEstimatedNumRecords(NUM_RECORD);
+ source.setEstimatedOutputSize(SIZE);
+ Assert.assertEquals(SIZE, channel.getEstimatedOutputSize());
+ Assert.assertEquals(NUM_RECORD, channel.getEstimatedNumRecords());
+ }
+
+ @Test
+ public void testGetEstimatesWithReplicationFactor() {
+ final long NUM_RECORD = 1001;
+ final long SIZE = 467131;
+
+ final int REPLICATION = 23;
+
+ DataSourceNode source = getSourceNode();
+ SourcePlanNode planNode = new SourcePlanNode(source, "test node");
+ Channel channel = new Channel(planNode);
+ channel.setReplicationFactor(REPLICATION);
+
+ // no estimates here
+ Assert.assertEquals(-1, channel.getEstimatedOutputSize());
+ Assert.assertEquals(-1, channel.getEstimatedNumRecords());
+
+ // set estimates
+ source.setEstimatedNumRecords(NUM_RECORD);
+ source.setEstimatedOutputSize(SIZE);
+ Assert.assertEquals(SIZE * REPLICATION, channel.getEstimatedOutputSize());
+ Assert.assertEquals(NUM_RECORD * REPLICATION, channel.getEstimatedNumRecords());
+ }
+
+
+// private static final OptimizerNode getSingleInputNode() {
+// return new MapNode(new MapOperatorBase<String, String, GenericMap<String,String>>(
+// new IdentityMapper<String>(),
+// new UnaryOperatorInformation<String, String>(BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO),
+// "map"));
+// }
+
+ private static final DataSourceNode getSourceNode() {
+ return new DataSourceNode(new GenericDataSourceBase<String, TextInputFormat>(
+ new TextInputFormat(new Path("/ignored")),
+ new OperatorInformation<String>(BasicTypeInfo.STRING_TYPE_INFO),
+ "source"));
+ }
+}
[11/12] git commit: Clean up checkstyle and JavaDoc errors in core
project
Posted by se...@apache.org.
Clean up checkstyle and JavaDoc errors in core project
Project: http://git-wip-us.apache.org/repos/asf/incubator-flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-flink/commit/49ca3edd
Tree: http://git-wip-us.apache.org/repos/asf/incubator-flink/tree/49ca3edd
Diff: http://git-wip-us.apache.org/repos/asf/incubator-flink/diff/49ca3edd
Branch: refs/heads/master
Commit: 49ca3edd9508f09040c35dfad056debc1072a58b
Parents: 52f9cf6
Author: Stephan Ewen <se...@apache.org>
Authored: Wed Jul 9 03:20:30 2014 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Wed Jul 9 03:20:30 2014 +0200
----------------------------------------------------------------------
.../api/common/operators/base/BulkIterationBase.java | 2 +-
.../eu/stratosphere/api/common/typeutils/TypeComparator.java | 4 ++--
.../src/main/java/eu/stratosphere/core/memory/DataInputView.java | 1 -
.../core/memory/InputViewDataInputStreamWrapper.java | 1 +
.../core/memory/InputViewObjectInputStreamWrapper.java | 1 +
.../java/eu/stratosphere/core/memory/SeekableDataInputView.java | 3 ---
.../src/main/java/eu/stratosphere/types/NormalizableKey.java | 4 ++--
.../src/main/java/eu/stratosphere/types/Record.java | 2 +-
.../src/test/java/eu/stratosphere/types/RecordTest.java | 2 --
9 files changed, 8 insertions(+), 12 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/49ca3edd/stratosphere-core/src/main/java/eu/stratosphere/api/common/operators/base/BulkIterationBase.java
----------------------------------------------------------------------
diff --git a/stratosphere-core/src/main/java/eu/stratosphere/api/common/operators/base/BulkIterationBase.java b/stratosphere-core/src/main/java/eu/stratosphere/api/common/operators/base/BulkIterationBase.java
index 6de9959..af8003e 100644
--- a/stratosphere-core/src/main/java/eu/stratosphere/api/common/operators/base/BulkIterationBase.java
+++ b/stratosphere-core/src/main/java/eu/stratosphere/api/common/operators/base/BulkIterationBase.java
@@ -145,7 +145,7 @@ public class BulkIterationBase<T> extends SingleInputOperator<T, T, AbstractFunc
}
/**
- * @throws Exception
+ * @throws InvalidProgramException
*/
public void validate() throws InvalidProgramException {
if (this.input == null) {
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/49ca3edd/stratosphere-core/src/main/java/eu/stratosphere/api/common/typeutils/TypeComparator.java
----------------------------------------------------------------------
diff --git a/stratosphere-core/src/main/java/eu/stratosphere/api/common/typeutils/TypeComparator.java b/stratosphere-core/src/main/java/eu/stratosphere/api/common/typeutils/TypeComparator.java
index e519d89..aec4239 100644
--- a/stratosphere-core/src/main/java/eu/stratosphere/api/common/typeutils/TypeComparator.java
+++ b/stratosphere-core/src/main/java/eu/stratosphere/api/common/typeutils/TypeComparator.java
@@ -40,7 +40,7 @@ import eu.stratosphere.core.memory.MemorySegment;
* @see java.lang.Object#equals(Object)
* @see java.util.Comparator#compare(Object, Object)
*
- * @param T The data type that the comparator works on.
+ * @param <T> The data type that the comparator works on.
*/
public abstract class TypeComparator<T> implements Serializable {
@@ -192,7 +192,7 @@ public abstract class TypeComparator<T> implements Serializable {
/**
* Gets the number of bytes that the normalized key would maximally take. A value of
- * {@link java.lang.Integer.MAX_VALUE} is interpreted as infinite.
+ * {@link java.lang.Integer}.MAX_VALUE is interpreted as infinite.
*
* @return The number of bytes that the normalized key would maximally take.
*/
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/49ca3edd/stratosphere-core/src/main/java/eu/stratosphere/core/memory/DataInputView.java
----------------------------------------------------------------------
diff --git a/stratosphere-core/src/main/java/eu/stratosphere/core/memory/DataInputView.java b/stratosphere-core/src/main/java/eu/stratosphere/core/memory/DataInputView.java
index db9aa6d..4b85f74 100644
--- a/stratosphere-core/src/main/java/eu/stratosphere/core/memory/DataInputView.java
+++ b/stratosphere-core/src/main/java/eu/stratosphere/core/memory/DataInputView.java
@@ -30,7 +30,6 @@ public interface DataInputView extends DataInput {
*
* @param numBytes The number of bytes to skip.
*
- * @throws EOFException Thrown, when less then {@code numBytes} remain in the input.
* @throws IOException Thrown, if any I/O related problem occurred such that the input could not
* be advanced to the desired position.
*/
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/49ca3edd/stratosphere-core/src/main/java/eu/stratosphere/core/memory/InputViewDataInputStreamWrapper.java
----------------------------------------------------------------------
diff --git a/stratosphere-core/src/main/java/eu/stratosphere/core/memory/InputViewDataInputStreamWrapper.java b/stratosphere-core/src/main/java/eu/stratosphere/core/memory/InputViewDataInputStreamWrapper.java
index 8344109..16569ab 100644
--- a/stratosphere-core/src/main/java/eu/stratosphere/core/memory/InputViewDataInputStreamWrapper.java
+++ b/stratosphere-core/src/main/java/eu/stratosphere/core/memory/InputViewDataInputStreamWrapper.java
@@ -108,6 +108,7 @@ public class InputViewDataInputStreamWrapper implements DataInputView {
return in.readDouble();
}
+ @SuppressWarnings("deprecation")
@Override
public String readLine() throws IOException {
return in.readLine();
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/49ca3edd/stratosphere-core/src/main/java/eu/stratosphere/core/memory/InputViewObjectInputStreamWrapper.java
----------------------------------------------------------------------
diff --git a/stratosphere-core/src/main/java/eu/stratosphere/core/memory/InputViewObjectInputStreamWrapper.java b/stratosphere-core/src/main/java/eu/stratosphere/core/memory/InputViewObjectInputStreamWrapper.java
index 55c9d8f..52d62e0 100644
--- a/stratosphere-core/src/main/java/eu/stratosphere/core/memory/InputViewObjectInputStreamWrapper.java
+++ b/stratosphere-core/src/main/java/eu/stratosphere/core/memory/InputViewObjectInputStreamWrapper.java
@@ -108,6 +108,7 @@ public class InputViewObjectInputStreamWrapper implements DataInputView {
return in.readDouble();
}
+ @SuppressWarnings("deprecation")
@Override
public String readLine() throws IOException {
return in.readLine();
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/49ca3edd/stratosphere-core/src/main/java/eu/stratosphere/core/memory/SeekableDataInputView.java
----------------------------------------------------------------------
diff --git a/stratosphere-core/src/main/java/eu/stratosphere/core/memory/SeekableDataInputView.java b/stratosphere-core/src/main/java/eu/stratosphere/core/memory/SeekableDataInputView.java
index eda486b..672d257 100644
--- a/stratosphere-core/src/main/java/eu/stratosphere/core/memory/SeekableDataInputView.java
+++ b/stratosphere-core/src/main/java/eu/stratosphere/core/memory/SeekableDataInputView.java
@@ -23,9 +23,6 @@ public interface SeekableDataInputView extends DataInputView {
* Sets the read pointer to the given position.
*
* @param position The new read position.
- *
- * @throws IOException Thrown, if any I/O related problem occurred such that the input could not
- * be sought to the desired position.
*/
public void setReadPosition(long position);
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/49ca3edd/stratosphere-core/src/main/java/eu/stratosphere/types/NormalizableKey.java
----------------------------------------------------------------------
diff --git a/stratosphere-core/src/main/java/eu/stratosphere/types/NormalizableKey.java b/stratosphere-core/src/main/java/eu/stratosphere/types/NormalizableKey.java
index 2d20125..8aae4b6 100644
--- a/stratosphere-core/src/main/java/eu/stratosphere/types/NormalizableKey.java
+++ b/stratosphere-core/src/main/java/eu/stratosphere/types/NormalizableKey.java
@@ -32,11 +32,11 @@ public interface NormalizableKey<T> extends Key<T> {
/**
* Gets the maximal length of normalized keys that the data type would produce to determine
- * the order of instances solely by the normalized key. A value of {@link java.lang.Integer.MAX_VALUE}
+ * the order of instances solely by the normalized key. A value of {@link java.lang.Integer}.MAX_VALUE
* is interpreted as infinite.
* <p>
* For example, 32 bit integers return four, while Strings (potentially unlimited in length) return
- * {@link java.lang.Integer.MAX_VALUE}.
+ * {@link java.lang.Integer}.MAX_VALUE.
*
* @return The maximal length of normalized keys.
*/
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/49ca3edd/stratosphere-core/src/main/java/eu/stratosphere/types/Record.java
----------------------------------------------------------------------
diff --git a/stratosphere-core/src/main/java/eu/stratosphere/types/Record.java b/stratosphere-core/src/main/java/eu/stratosphere/types/Record.java
index 78e90f0..d07f1f6 100644
--- a/stratosphere-core/src/main/java/eu/stratosphere/types/Record.java
+++ b/stratosphere-core/src/main/java/eu/stratosphere/types/Record.java
@@ -344,7 +344,7 @@ public final class Record implements Value, CopyableValue<Record> {
* @param positions The positions of the fields to get.
* @param targets The values into which the content of the fields is put.
*
- * @throws A NullKeyFieldException in case of a failing field read.
+ * @throws NullKeyFieldException in case of a failing field read.
*/
public void getFieldsIntoCheckingNull(int[] positions, Value[] targets) {
for (int i = 0; i < positions.length; i++) {
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/49ca3edd/stratosphere-core/src/test/java/eu/stratosphere/types/RecordTest.java
----------------------------------------------------------------------
diff --git a/stratosphere-core/src/test/java/eu/stratosphere/types/RecordTest.java b/stratosphere-core/src/test/java/eu/stratosphere/types/RecordTest.java
index 89b3ce6..0118f10 100644
--- a/stratosphere-core/src/test/java/eu/stratosphere/types/RecordTest.java
+++ b/stratosphere-core/src/test/java/eu/stratosphere/types/RecordTest.java
@@ -16,9 +16,7 @@ package eu.stratosphere.types;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
-import java.io.DataInput;
import java.io.DataInputStream;
-import java.io.DataOutput;
import java.io.DataOutputStream;
import java.io.IOException;
import java.io.PipedInputStream;
[04/12] git commit: Fix various JavaDoc errors
Posted by se...@apache.org.
Fix various JavaDoc errors
Project: http://git-wip-us.apache.org/repos/asf/incubator-flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-flink/commit/53c7a894
Tree: http://git-wip-us.apache.org/repos/asf/incubator-flink/tree/53c7a894
Diff: http://git-wip-us.apache.org/repos/asf/incubator-flink/diff/53c7a894
Branch: refs/heads/master
Commit: 53c7a8944dc92c457aeaf089f35fec5c53eb5a36
Parents: 47a68ad
Author: Stephan Ewen <se...@apache.org>
Authored: Tue Jul 8 16:15:23 2014 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Tue Jul 8 16:55:07 2014 +0200
----------------------------------------------------------------------
.../java/eu/stratosphere/api/java/DataSet.java | 20 ++++++++++----------
.../java/aggregation/AggregationFunction.java | 1 -
.../api/java/functions/FunctionAnnotation.java | 6 +++---
.../api/java/functions/JoinFunction.java | 4 ++--
.../api/java/functions/ReduceFunction.java | 2 +-
.../api/java/operators/Grouping.java | 6 +++---
.../api/java/operators/SortedGrouping.java | 8 ++++----
.../api/java/operators/UdfOperator.java | 15 ++++++++++-----
.../api/java/operators/UnsortedGrouping.java | 8 ++++----
.../java/record/functions/CoGroupFunction.java | 2 +-
.../java/record/functions/CrossFunction.java | 2 +-
.../api/java/record/functions/JoinFunction.java | 2 +-
.../java/record/functions/ReduceFunction.java | 2 +-
.../ExternalProcessFixedLengthInputFormat.java | 4 +---
.../record/io/ExternalProcessInputFormat.java | 4 ++--
.../record/io/ExternalProcessInputSplit.java | 2 +-
.../record/operators/CollectionDataSource.java | 6 +++---
.../runtime/record/RecordComparator.java | 8 ++------
.../runtime/record/RecordComparatorFactory.java | 2 +-
19 files changed, 51 insertions(+), 53 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/53c7a894/stratosphere-java/src/main/java/eu/stratosphere/api/java/DataSet.java
----------------------------------------------------------------------
diff --git a/stratosphere-java/src/main/java/eu/stratosphere/api/java/DataSet.java b/stratosphere-java/src/main/java/eu/stratosphere/api/java/DataSet.java
index 9ebb10b..4c3d31f 100644
--- a/stratosphere-java/src/main/java/eu/stratosphere/api/java/DataSet.java
+++ b/stratosphere-java/src/main/java/eu/stratosphere/api/java/DataSet.java
@@ -288,7 +288,7 @@ public abstract class DataSet<T> {
* @return A GroupReduceOperator that represents the reduced DataSet.
*
* @see GroupReduceFunction
- * @see GroupReduceOperator
+ * @see ReduceGroupOperator
* @see DataSet
*/
public <R> ReduceGroupOperator<T, R> reduceGroup(GroupReduceFunction<T, R> reducer) {
@@ -383,9 +383,9 @@ public abstract class DataSet<T> {
* can be applied.
* <ul>
* <li>{@link UnsortedGrouping#sortGroup(int, eu.stratosphere.api.common.operators.Order)} to get a {@link SortedGrouping}.
- * <li>{@link Grouping#aggregate(Aggregations, int)} to apply an Aggregate transformation.
- * <li>{@link Grouping#reduce(ReduceFunction)} to apply a Reduce transformation.
- * <li>{@link Grouping#reduceGroup(GroupReduceFunction)} to apply a GroupReduce transformation.
+ * <li>{@link UnsortedGrouping#aggregate(Aggregations, int)} to apply an Aggregate transformation.
+ * <li>{@link UnsortedGrouping#reduce(ReduceFunction)} to apply a Reduce transformation.
+ * <li>{@link UnsortedGrouping#reduceGroup(GroupReduceFunction)} to apply a GroupReduce transformation.
* </ul>
*
* @param fields One or more field positions on which the DataSet will be grouped.
@@ -412,9 +412,9 @@ public abstract class DataSet<T> {
* can be applied.
* <ul>
* <li>{@link UnsortedGrouping#sortGroup(int, eu.stratosphere.api.common.operators.Order)} to get a {@link SortedGrouping}.
- * <li>{@link Grouping#aggregate(Aggregations, int)} to apply an Aggregate transformation.
- * <li>{@link Grouping#reduce(ReduceFunction)} to apply a Reduce transformation.
- * <li>{@link Grouping#reduceGroup(GroupReduceFunction)} to apply a GroupReduce transformation.
+ * <li>{@link UnsortedGrouping#aggregate(Aggregations, int)} to apply an Aggregate transformation.
+ * <li>{@link UnsortedGrouping#reduce(ReduceFunction)} to apply a Reduce transformation.
+ * <li>{@link UnsortedGrouping#reduceGroup(GroupReduceFunction)} to apply a GroupReduce transformation.
* </ul>
*
* @param fields One or more field expressions on which the DataSet will be grouped.
@@ -516,7 +516,7 @@ public abstract class DataSet<T> {
* The CoGroupFunction can iterate over the elements of both groups and return any number
* of elements including none.</br>
* This method returns a {@link CoGroupOperatorSets} on which
- * {@link CoGroupOperatorSets#where(} needs to be called to define the grouping key of the first
+ * {@link CoGroupOperatorSets#where()} needs to be called to define the grouping key of the first
* (i.e., this) DataSet.
*
* @param other The other DataSet of the CoGroup transformation.
@@ -794,8 +794,8 @@ public abstract class DataSet<T> {
* Writes a {@link Tuple} DataSet as a CSV file to the specified location.<br/>
* <b>Note: Only a Tuple DataSet can written as a CSV file.</b><br/>
* For each Tuple field the result of {@link Object#toString()} is written.
- * Tuple fields are separated by the default field delimiter {@link CsvOutputFormat.DEFAULT_FIELD_DELIMITER}.<br/>
- * Tuples are are separated by the default line delimiter {@link CsvOutputFormat.DEFAULT_LINE_DELIMITER}.
+ * Tuple fields are separated by the default field delimiter {@code "comma" (,)}.<br/>
+ * Tuples are are separated by the newline character ({@code \n}).
*
* @param filePath The path pointing to the location the CSV file is written to.
* @return The DataSink that writes the DataSet.
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/53c7a894/stratosphere-java/src/main/java/eu/stratosphere/api/java/aggregation/AggregationFunction.java
----------------------------------------------------------------------
diff --git a/stratosphere-java/src/main/java/eu/stratosphere/api/java/aggregation/AggregationFunction.java b/stratosphere-java/src/main/java/eu/stratosphere/api/java/aggregation/AggregationFunction.java
index cf75ed2..fb6eb88 100644
--- a/stratosphere-java/src/main/java/eu/stratosphere/api/java/aggregation/AggregationFunction.java
+++ b/stratosphere-java/src/main/java/eu/stratosphere/api/java/aggregation/AggregationFunction.java
@@ -16,7 +16,6 @@ package eu.stratosphere.api.java.aggregation;
/**
- * @param <R> The type of the running aggregate.
* @param <T> The type to be aggregated.
*/
public abstract class AggregationFunction<T> implements java.io.Serializable {
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/53c7a894/stratosphere-java/src/main/java/eu/stratosphere/api/java/functions/FunctionAnnotation.java
----------------------------------------------------------------------
diff --git a/stratosphere-java/src/main/java/eu/stratosphere/api/java/functions/FunctionAnnotation.java b/stratosphere-java/src/main/java/eu/stratosphere/api/java/functions/FunctionAnnotation.java
index cecc644..5b37b48 100644
--- a/stratosphere-java/src/main/java/eu/stratosphere/api/java/functions/FunctionAnnotation.java
+++ b/stratosphere-java/src/main/java/eu/stratosphere/api/java/functions/FunctionAnnotation.java
@@ -177,7 +177,7 @@ public class FunctionAnnotation {
* <p>
* The annotation takes one String array specifying the positions of the input types that do not remain constant.
* When this annotation is used, it is assumed that all other values remain at the same position in input and output.
- * To model more complex situations use the {@link @ConstantFields}s annotation.
+ * To model more complex situations use the {@link ConstantFields}s annotation.
* <p>
* This annotation is mutually exclusive with the {@link ConstantFields} annotation.
* <p>
@@ -212,7 +212,7 @@ public class FunctionAnnotation {
* <p>
* The annotation takes one String array specifying the positions of the input types that do not remain constant.
* When this annotation is used, it is assumed that all other values remain at the same position in input and output.
- * To model more complex situations use the {@link @ConstantFields}s annotation.
+ * To model more complex situations use the {@link ConstantFields}s annotation.
* <p>
* This annotation is mutually exclusive with the {@link ConstantFieldsFirst}
* <p>
@@ -249,7 +249,7 @@ public class FunctionAnnotation {
* <p>
* The annotation takes one String array specifying the positions of the input types that do not remain constant.
* When this annotation is used, it is assumed that all other values remain at the same position in input and output.
- * To model more complex situations use the {@link @ConstantFields}s annotation.
+ * To model more complex situations use the {@link ConstantFields}s annotation.
* <p>
* This annotation is mutually exclusive with the {@link ConstantFieldsSecond}
* <p>
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/53c7a894/stratosphere-java/src/main/java/eu/stratosphere/api/java/functions/JoinFunction.java
----------------------------------------------------------------------
diff --git a/stratosphere-java/src/main/java/eu/stratosphere/api/java/functions/JoinFunction.java b/stratosphere-java/src/main/java/eu/stratosphere/api/java/functions/JoinFunction.java
index f704255..7d6f0d7 100644
--- a/stratosphere-java/src/main/java/eu/stratosphere/api/java/functions/JoinFunction.java
+++ b/stratosphere-java/src/main/java/eu/stratosphere/api/java/functions/JoinFunction.java
@@ -75,8 +75,8 @@ public abstract class JoinFunction<IN1, IN2, OUT> extends AbstractFunction imple
* By default, this method delegates to the method {@link #join(Object, Object)}. If this method
* is overridden, that method will no longer be called.
*
- * @param first The element from first input.
- * @param second The element from second input.
+ * @param value1 The element from first input.
+ * @param value2 The element from second input.
* @param out A collector to emit resulting element (zero, one, or many).
*
* @throws Exception This method may throw exceptions. Throwing an exception will cause the operation
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/53c7a894/stratosphere-java/src/main/java/eu/stratosphere/api/java/functions/ReduceFunction.java
----------------------------------------------------------------------
diff --git a/stratosphere-java/src/main/java/eu/stratosphere/api/java/functions/ReduceFunction.java b/stratosphere-java/src/main/java/eu/stratosphere/api/java/functions/ReduceFunction.java
index d3f2f65..09072ec 100644
--- a/stratosphere-java/src/main/java/eu/stratosphere/api/java/functions/ReduceFunction.java
+++ b/stratosphere-java/src/main/java/eu/stratosphere/api/java/functions/ReduceFunction.java
@@ -38,7 +38,7 @@ import eu.stratosphere.api.common.functions.GenericReduce;
* <p>
* Like all functions, the ReduceFunction needs to be serializable, as defined in {@link java.io.Serializable}.
*
- * @param <IN> Type of the elements that this function processes.
+ * @param <T> Type of the elements that this function processes.
*/
public abstract class ReduceFunction<T> extends AbstractFunction implements GenericReduce<T> {
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/53c7a894/stratosphere-java/src/main/java/eu/stratosphere/api/java/operators/Grouping.java
----------------------------------------------------------------------
diff --git a/stratosphere-java/src/main/java/eu/stratosphere/api/java/operators/Grouping.java b/stratosphere-java/src/main/java/eu/stratosphere/api/java/operators/Grouping.java
index 25a0ed1..881659d 100644
--- a/stratosphere-java/src/main/java/eu/stratosphere/api/java/operators/Grouping.java
+++ b/stratosphere-java/src/main/java/eu/stratosphere/api/java/operators/Grouping.java
@@ -21,9 +21,9 @@ import eu.stratosphere.api.java.DataSet;
* Grouping is an intermediate step for a transformation on a grouped DataSet.<br/>
* The following transformation can be applied on Grouping:
* <ul>
- * <li>{@link Grouping#reduce(ReduceFunction)},</li>
- * <li>{@link Grouping#reduceGroup(GroupReduceFunction)}, and</li>
- * <li>{@link Grouping#aggregate(Aggregations, int)}.</li>
+ * <li>{@link UnsortedGrouping#reduce(eu.stratosphere.api.java.functions.ReduceFunction)},</li>
+ * <li>{@link UnsortedGrouping#reduceGroup(eu.stratosphere.api.java.functions.GroupReduceFunction)}, and</li>
+ * <li>{@link UnsortedGrouping#aggregate(eu.stratosphere.api.java.aggregation.Aggregations, int)}.</li>
* </ul>
*
* @param <T> The type of the elements of the grouped DataSet.
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/53c7a894/stratosphere-java/src/main/java/eu/stratosphere/api/java/operators/SortedGrouping.java
----------------------------------------------------------------------
diff --git a/stratosphere-java/src/main/java/eu/stratosphere/api/java/operators/SortedGrouping.java b/stratosphere-java/src/main/java/eu/stratosphere/api/java/operators/SortedGrouping.java
index dc26a2b..d961e4a 100644
--- a/stratosphere-java/src/main/java/eu/stratosphere/api/java/operators/SortedGrouping.java
+++ b/stratosphere-java/src/main/java/eu/stratosphere/api/java/operators/SortedGrouping.java
@@ -25,7 +25,7 @@ import eu.stratosphere.api.java.functions.GroupReduceFunction;
* SortedGrouping is an intermediate step for a transformation on a grouped and sorted DataSet.<br/>
* The following transformation can be applied on sorted groups:
* <ul>
- * <li>{@link Grouping#reduce(ReduceFunction)},</li>
+ * <li>{@link SortedGrouping#reduceGroup(GroupReduceFunction)},</li>
* </ul>
*
* @param <T> The type of the elements of the sorted and grouped DataSet.
@@ -68,7 +68,7 @@ public class SortedGrouping<T> extends Grouping<T> {
* @return A GroupReduceOperator that represents the reduced DataSet.
*
* @see GroupReduceFunction
- * @see GroupReduceOperator
+ * @see ReduceGroupOperator
* @see DataSet
*/
public <R> ReduceGroupOperator<T, R> reduceGroup(GroupReduceFunction<T, R> reducer) {
@@ -83,7 +83,7 @@ public class SortedGrouping<T> extends Grouping<T> {
// --------------------------------------------------------------------------------------------
/**
- * Sorts {@link Tuple} elements within a group on the specified field in the specified {@link Order}.</br>
+ * Sorts {@link eu.stratosphere.api.java.tuple.Tuple} elements within a group on the specified field in the specified {@link Order}.</br>
* <b>Note: Only groups of Tuple elements can be sorted.</b><br/>
* Groups can be sorted by multiple fields by chaining {@link #sortGroup(int, Order)} calls.
*
@@ -91,7 +91,7 @@ public class SortedGrouping<T> extends Grouping<T> {
* @param order The Order in which the specified Tuple field is sorted.
* @return A SortedGrouping with specified order of group element.
*
- * @see Tuple
+ * @see eu.stratosphere.api.java.tuple.Tuple
* @see Order
*/
public SortedGrouping<T> sortGroup(int field, Order order) {
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/53c7a894/stratosphere-java/src/main/java/eu/stratosphere/api/java/operators/UdfOperator.java
----------------------------------------------------------------------
diff --git a/stratosphere-java/src/main/java/eu/stratosphere/api/java/operators/UdfOperator.java b/stratosphere-java/src/main/java/eu/stratosphere/api/java/operators/UdfOperator.java
index ce72a3b..76e2c26 100644
--- a/stratosphere-java/src/main/java/eu/stratosphere/api/java/operators/UdfOperator.java
+++ b/stratosphere-java/src/main/java/eu/stratosphere/api/java/operators/UdfOperator.java
@@ -22,7 +22,8 @@ import eu.stratosphere.configuration.Configuration;
/**
* This interface marks operators as operators that execute user-defined functions (UDFs), such as
- * {@link MapFunction}, {@link ReduceFunction} or {@link CoGroupFunction}.
+ * {@link eu.stratosphere.api.java.functions.MapFunction}, {@link eu.stratosphere.api.java.functions.ReduceFunction},
+ * or {@link eu.stratosphere.api.java.functions.CoGroupFunction}.
* The UDF operators stand in contrast to operators that execute built-in operations, like aggregations.
*/
public interface UdfOperator<O extends UdfOperator<O>> {
@@ -33,7 +34,8 @@ public interface UdfOperator<O extends UdfOperator<O>> {
/**
* Gets the configuration parameters that will be passed to the UDF's open method
- * {@link AbstractFunction#open(Configuration)}. The configuration is set via the {@link #withParameters(Configuration)}
+ * {@link eu.stratosphere.api.common.functions.AbstractFunction#open(Configuration)}.
+ * The configuration is set via the {@link #withParameters(Configuration)}
* method.
*
* @return The configuration parameters for the UDF.
@@ -62,7 +64,7 @@ public interface UdfOperator<O extends UdfOperator<O>> {
/**
* Sets the configuration parameters for the UDF. These are optional parameters that are passed
- * to the UDF in the {@link AbstractFunction#open(Configuration)} method.
+ * to the UDF in the {@link eu.stratosphere.api.common.functions.AbstractFunction#open(Configuration)} method.
*
* @param parameters The configuration parameters for the UDF.
* @return The operator itself, to allow chaining function calls.
@@ -72,8 +74,11 @@ public interface UdfOperator<O extends UdfOperator<O>> {
/**
* Adds a certain data set as a broadcast set to this operator. Broadcasted data sets are available at all
* parallel instances of this operator. A broadcast data set is registered under a certain name, and can be
- * retrieved under that name from the operators runtime context via {@link RuntimeContext#getBroadcastVariable(String)}.
- * The runtime context itself is available in all UDFs via {@link AbstractFunction#getRuntimeContext()}.
+ * retrieved under that name from the operators runtime context via
+ * {@link eu.stratosphere.api.common.functions.RuntimeContext#getBroadcastVariable(String)}.
+ *
+ * The runtime context itself is available in all UDFs via
+ * {@link eu.stratosphere.api.common.functions.AbstractFunction#getRuntimeContext()}.
*
* @param data The data set to be broadcasted.
* @param name The name under which the broadcast data set retrieved.
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/53c7a894/stratosphere-java/src/main/java/eu/stratosphere/api/java/operators/UnsortedGrouping.java
----------------------------------------------------------------------
diff --git a/stratosphere-java/src/main/java/eu/stratosphere/api/java/operators/UnsortedGrouping.java b/stratosphere-java/src/main/java/eu/stratosphere/api/java/operators/UnsortedGrouping.java
index b61d427..3b2ef50 100644
--- a/stratosphere-java/src/main/java/eu/stratosphere/api/java/operators/UnsortedGrouping.java
+++ b/stratosphere-java/src/main/java/eu/stratosphere/api/java/operators/UnsortedGrouping.java
@@ -31,7 +31,7 @@ public class UnsortedGrouping<T> extends Grouping<T> {
// --------------------------------------------------------------------------------------------
/**
- * Applies an Aggregate transformation on a grouped {@link Tuple} {@link DataSet}.<br/>
+ * Applies an Aggregate transformation on a grouped {@link eu.stratosphere.api.java.tuple.Tuple} {@link DataSet}.<br/>
* <b>Note: Only Tuple DataSets can be aggregated.</b>
* The transformation applies a built-in {@link Aggregations Aggregation} on a specified field
* of a Tuple group. Additional aggregation functions can be added to the resulting
@@ -41,7 +41,7 @@ public class UnsortedGrouping<T> extends Grouping<T> {
* @param field The index of the Tuple field on which the aggregation function is applied.
* @return An AggregateOperator that represents the aggregated DataSet.
*
- * @see Tuple
+ * @see eu.stratosphere.api.java.tuple.Tuple
* @see Aggregations
* @see AggregateOperator
* @see DataSet
@@ -128,7 +128,7 @@ public class UnsortedGrouping<T> extends Grouping<T> {
// --------------------------------------------------------------------------------------------
/**
- * Sorts {@link Tuple} elements within a group on the specified field in the specified {@link Order}.</br>
+ * Sorts {@link eu.stratosphere.api.java.tuple.Tuple} elements within a group on the specified field in the specified {@link Order}.</br>
* <b>Note: Only groups of Tuple elements can be sorted.</b><br/>
* Groups can be sorted by multiple fields by chaining {@link #sortGroup(int, Order)} calls.
*
@@ -136,7 +136,7 @@ public class UnsortedGrouping<T> extends Grouping<T> {
* @param order The Order in which the specified Tuple field is sorted.
* @return A SortedGrouping with specified order of group element.
*
- * @see Tuple
+ * @see eu.stratosphere.api.java.tuple.Tuple
* @see Order
*/
public SortedGrouping<T> sortGroup(int field, Order order) {
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/53c7a894/stratosphere-java/src/main/java/eu/stratosphere/api/java/record/functions/CoGroupFunction.java
----------------------------------------------------------------------
diff --git a/stratosphere-java/src/main/java/eu/stratosphere/api/java/record/functions/CoGroupFunction.java b/stratosphere-java/src/main/java/eu/stratosphere/api/java/record/functions/CoGroupFunction.java
index 3a81d34..39d62c7 100644
--- a/stratosphere-java/src/main/java/eu/stratosphere/api/java/record/functions/CoGroupFunction.java
+++ b/stratosphere-java/src/main/java/eu/stratosphere/api/java/record/functions/CoGroupFunction.java
@@ -21,7 +21,7 @@ import eu.stratosphere.types.Record;
import eu.stratosphere.util.Collector;
/**
- * The CoGroupFunction is the base class for functions that are invoked by a {@link CoGroupOperator}.
+ * The CoGroupFunction is the base class for functions that are invoked by a {@link eu.stratosphere.api.java.operators.CoGroupOperator}.
*/
public abstract class CoGroupFunction extends AbstractFunction implements GenericCoGrouper<Record, Record, Record> {
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/53c7a894/stratosphere-java/src/main/java/eu/stratosphere/api/java/record/functions/CrossFunction.java
----------------------------------------------------------------------
diff --git a/stratosphere-java/src/main/java/eu/stratosphere/api/java/record/functions/CrossFunction.java b/stratosphere-java/src/main/java/eu/stratosphere/api/java/record/functions/CrossFunction.java
index de30eb4..35d66f5 100644
--- a/stratosphere-java/src/main/java/eu/stratosphere/api/java/record/functions/CrossFunction.java
+++ b/stratosphere-java/src/main/java/eu/stratosphere/api/java/record/functions/CrossFunction.java
@@ -19,7 +19,7 @@ import eu.stratosphere.types.Record;
import eu.stratosphere.util.Collector;
/**
- * The CrossFunction is the base class for functions that are invoked by a {@link CrossOperator}.
+ * The CrossFunction is the base class for functions that are invoked by a {@link eu.stratosphere.api.java.operators.CrossOperator}.
*/
public abstract class CrossFunction extends AbstractFunction implements GenericCrosser<Record, Record, Record> {
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/53c7a894/stratosphere-java/src/main/java/eu/stratosphere/api/java/record/functions/JoinFunction.java
----------------------------------------------------------------------
diff --git a/stratosphere-java/src/main/java/eu/stratosphere/api/java/record/functions/JoinFunction.java b/stratosphere-java/src/main/java/eu/stratosphere/api/java/record/functions/JoinFunction.java
index 375912c..78c8c97 100644
--- a/stratosphere-java/src/main/java/eu/stratosphere/api/java/record/functions/JoinFunction.java
+++ b/stratosphere-java/src/main/java/eu/stratosphere/api/java/record/functions/JoinFunction.java
@@ -19,7 +19,7 @@ import eu.stratosphere.types.Record;
import eu.stratosphere.util.Collector;
/**
- * The JoinFunction must implementation by functions of a {@link JoinOperator}.
+ * The JoinFunction must implementation by functions of a {@link eu.stratosphere.api.java.operators.JoinOperator}.
* It resembles an equality join of both inputs on their key fields.
*/
public abstract class JoinFunction extends AbstractFunction implements GenericJoiner<Record, Record, Record> {
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/53c7a894/stratosphere-java/src/main/java/eu/stratosphere/api/java/record/functions/ReduceFunction.java
----------------------------------------------------------------------
diff --git a/stratosphere-java/src/main/java/eu/stratosphere/api/java/record/functions/ReduceFunction.java b/stratosphere-java/src/main/java/eu/stratosphere/api/java/record/functions/ReduceFunction.java
index 0884b0b..26abca2 100644
--- a/stratosphere-java/src/main/java/eu/stratosphere/api/java/record/functions/ReduceFunction.java
+++ b/stratosphere-java/src/main/java/eu/stratosphere/api/java/record/functions/ReduceFunction.java
@@ -23,7 +23,7 @@ import eu.stratosphere.util.Collector;
/**
* The ReduceFunction must be extended to provide a reducer implementation, as invoked by a
- * {@link ReduceOperator}.
+ * {@link eu.stratosphere.api.java.operators.ReduceOperator}.
*/
public abstract class ReduceFunction extends AbstractFunction implements GenericGroupReduce<Record, Record>, GenericCombine<Record> {
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/53c7a894/stratosphere-java/src/main/java/eu/stratosphere/api/java/record/io/ExternalProcessFixedLengthInputFormat.java
----------------------------------------------------------------------
diff --git a/stratosphere-java/src/main/java/eu/stratosphere/api/java/record/io/ExternalProcessFixedLengthInputFormat.java b/stratosphere-java/src/main/java/eu/stratosphere/api/java/record/io/ExternalProcessFixedLengthInputFormat.java
index 1e48bef..2c1acb2 100644
--- a/stratosphere-java/src/main/java/eu/stratosphere/api/java/record/io/ExternalProcessFixedLengthInputFormat.java
+++ b/stratosphere-java/src/main/java/eu/stratosphere/api/java/record/io/ExternalProcessFixedLengthInputFormat.java
@@ -28,10 +28,8 @@ import eu.stratosphere.types.Record;
* The input format requires ({@link ExternalProcessInputSplit} objects that hold the command to execute.
*
* <b>Warning:</b> This format does not consume the standard error stream (stderr) of the started process. This might cause deadlocks.
- *
*
- * @param <T>, The type of the input split (must extend ExternalProcessInputSplit)
- *
+ * @param <T> The type of the input split (must extend ExternalProcessInputSplit)
*/
public abstract class ExternalProcessFixedLengthInputFormat<T extends ExternalProcessInputSplit> extends ExternalProcessInputFormat<T> {
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/53c7a894/stratosphere-java/src/main/java/eu/stratosphere/api/java/record/io/ExternalProcessInputFormat.java
----------------------------------------------------------------------
diff --git a/stratosphere-java/src/main/java/eu/stratosphere/api/java/record/io/ExternalProcessInputFormat.java b/stratosphere-java/src/main/java/eu/stratosphere/api/java/record/io/ExternalProcessInputFormat.java
index 15f27d9..535d3c7 100644
--- a/stratosphere-java/src/main/java/eu/stratosphere/api/java/record/io/ExternalProcessInputFormat.java
+++ b/stratosphere-java/src/main/java/eu/stratosphere/api/java/record/io/ExternalProcessInputFormat.java
@@ -26,12 +26,12 @@ import eu.stratosphere.core.io.GenericInputSplit;
* The input format checks the exit code of the process to validate whether the process terminated correctly. A list of allowed exit codes can be provided.
* The input format requires ({@link ExternalProcessInputSplit} objects that hold the command to execute.
*
- * <b> Attention! </b><br>
+ * <b>Attention! </b><br/>
* You must take care to read from (and process) both output streams of the process, standard out (stdout) and standard error (stderr).
* Otherwise, the input format might get deadlocked!
*
*
- * @param <T>, The type of the input split (must extend ExternalProcessInputSplit)
+ * @param <T> The type of the input split (must extend ExternalProcessInputSplit)
*/
public abstract class ExternalProcessInputFormat<T extends ExternalProcessInputSplit> extends GenericInputFormat {
private static final long serialVersionUID = 1L;
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/53c7a894/stratosphere-java/src/main/java/eu/stratosphere/api/java/record/io/ExternalProcessInputSplit.java
----------------------------------------------------------------------
diff --git a/stratosphere-java/src/main/java/eu/stratosphere/api/java/record/io/ExternalProcessInputSplit.java b/stratosphere-java/src/main/java/eu/stratosphere/api/java/record/io/ExternalProcessInputSplit.java
index e6677a4..4c856ed 100644
--- a/stratosphere-java/src/main/java/eu/stratosphere/api/java/record/io/ExternalProcessInputSplit.java
+++ b/stratosphere-java/src/main/java/eu/stratosphere/api/java/record/io/ExternalProcessInputSplit.java
@@ -21,7 +21,7 @@ import eu.stratosphere.core.memory.DataInputView;
import eu.stratosphere.core.memory.DataOutputView;
/**
- * The ExternalProcessInputSplit contains all informations for {@link InputFormat} that read their data from external processes.
+ * The ExternalProcessInputSplit contains all informations for {@link eu.stratosphere.api.common.io.InputFormat} that read their data from external processes.
* Each parallel instance of an InputFormat starts an external process and reads its output.
* The command to start the external process must be executable on all nodes.
*
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/53c7a894/stratosphere-java/src/main/java/eu/stratosphere/api/java/record/operators/CollectionDataSource.java
----------------------------------------------------------------------
diff --git a/stratosphere-java/src/main/java/eu/stratosphere/api/java/record/operators/CollectionDataSource.java b/stratosphere-java/src/main/java/eu/stratosphere/api/java/record/operators/CollectionDataSource.java
index f4f85d6..fbb9355 100644
--- a/stratosphere-java/src/main/java/eu/stratosphere/api/java/record/operators/CollectionDataSource.java
+++ b/stratosphere-java/src/main/java/eu/stratosphere/api/java/record/operators/CollectionDataSource.java
@@ -28,13 +28,13 @@ import eu.stratosphere.types.Record;
* Use this operator if you want to pass data from the application submitting the
* Stratosphere job to the cluster.
* There are two main ways to use the CollectionDataSource:
- * * Using a @link {@link SerializableIterator}
*
+ * Using an {@link java.util.Iterator} that is also {@link java.io.Serializable}.
* <pre>
- * CollectionDataSource source = new CollectionDataSource(new SerializableIteratorTest(), "IterSource");
+ * CollectionDataSource source = new CollectionDataSource(mySerializableIterator, "IterSource");
* </pre>
*
- * * Using a Collection of Java Objects.
+ * Using a Collection of Java Objects.
*
* <pre>
* CollectionDataSource source2 = new CollectionDataSource(new List<String>(), "Collection source");
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/53c7a894/stratosphere-java/src/main/java/eu/stratosphere/api/java/typeutils/runtime/record/RecordComparator.java
----------------------------------------------------------------------
diff --git a/stratosphere-java/src/main/java/eu/stratosphere/api/java/typeutils/runtime/record/RecordComparator.java b/stratosphere-java/src/main/java/eu/stratosphere/api/java/typeutils/runtime/record/RecordComparator.java
index b487cd0..fe28283 100644
--- a/stratosphere-java/src/main/java/eu/stratosphere/api/java/typeutils/runtime/record/RecordComparator.java
+++ b/stratosphere-java/src/main/java/eu/stratosphere/api/java/typeutils/runtime/record/RecordComparator.java
@@ -82,7 +82,7 @@ public final class RecordComparator extends TypeComparator<Record> {
*
* @param keyFields The positions of the key fields.
* @param keyTypes The types (classes) of the key fields.
- * @param sortOrder The direction for sorting. A value of <i>true</i> indicates ascending for an attribute,
+ * @param sortDirection The direction for sorting. A value of <i>true</i> indicates ascending for an attribute,
* a value of <i>false</i> indicated descending. If the parameter is <i>null</i>, then
* all order comparisons will assume ascending order on all fields.
*/
@@ -150,11 +150,7 @@ public final class RecordComparator extends TypeComparator<Record> {
/**
* Copy constructor.
*
- * @param keyFields
- * @param keys
- * @param normalKeyLengths
- * @param leadingNormalKeys
- * @param normalKeyPrefixLen
+ * @param toCopy Comparator to copy.
*/
private RecordComparator(RecordComparator toCopy) {
this.keyFields = toCopy.keyFields;
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/53c7a894/stratosphere-java/src/main/java/eu/stratosphere/api/java/typeutils/runtime/record/RecordComparatorFactory.java
----------------------------------------------------------------------
diff --git a/stratosphere-java/src/main/java/eu/stratosphere/api/java/typeutils/runtime/record/RecordComparatorFactory.java b/stratosphere-java/src/main/java/eu/stratosphere/api/java/typeutils/runtime/record/RecordComparatorFactory.java
index 49a4dce..ebc5fd6 100644
--- a/stratosphere-java/src/main/java/eu/stratosphere/api/java/typeutils/runtime/record/RecordComparatorFactory.java
+++ b/stratosphere-java/src/main/java/eu/stratosphere/api/java/typeutils/runtime/record/RecordComparatorFactory.java
@@ -22,7 +22,7 @@ import eu.stratosphere.types.Key;
import eu.stratosphere.types.Record;
/**
- * A factory for a {@link TypeComparator} for {@link Record}. The comparator uses a subset of
+ * A factory for a {@link eu.stratosphere.api.common.typeutils.TypeComparator} for {@link Record}. The comparator uses a subset of
* the fields for the comparison. That subset of fields (positions and types) is read from the
* supplied configuration.
*/
[12/12] git commit: Fix optimizer error in channel instantiation
(wrong parameterization)
Posted by se...@apache.org.
Fix optimizer error in channel instantiation (wrong parameterization)
Project: http://git-wip-us.apache.org/repos/asf/incubator-flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-flink/commit/26bc321c
Tree: http://git-wip-us.apache.org/repos/asf/incubator-flink/tree/26bc321c
Diff: http://git-wip-us.apache.org/repos/asf/incubator-flink/diff/26bc321c
Branch: refs/heads/master
Commit: 26bc321c31c6625f782a1adfbc97a91e1cc0d288
Parents: 49ca3ed
Author: Stephan Ewen <se...@apache.org>
Authored: Wed Jul 9 11:23:18 2014 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Wed Jul 9 11:23:18 2014 +0200
----------------------------------------------------------------------
.../stratosphere/compiler/costs/DefaultCostEstimator.java | 2 +-
.../java/eu/stratosphere/compiler/dag/TwoInputNode.java | 8 ++++++--
.../compiler/examples/RelationalQueryCompilerTest.java | 1 +
.../test/compiler/iterations/ConnectedComponentsTest.java | 10 ++++++----
4 files changed, 14 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/26bc321c/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/costs/DefaultCostEstimator.java
----------------------------------------------------------------------
diff --git a/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/costs/DefaultCostEstimator.java b/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/costs/DefaultCostEstimator.java
index 93eb66a..ef3e51c 100644
--- a/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/costs/DefaultCostEstimator.java
+++ b/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/costs/DefaultCostEstimator.java
@@ -42,7 +42,7 @@ public class DefaultCostEstimator extends CostEstimator {
private static final float HASHING_CPU_FACTOR = 4;
- private static final float SORTING_CPU_FACTOR = 7;
+ private static final float SORTING_CPU_FACTOR = 9;
// --------------------------------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/26bc321c/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/dag/TwoInputNode.java
----------------------------------------------------------------------
diff --git a/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/dag/TwoInputNode.java b/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/dag/TwoInputNode.java
index 97a92d0..dd4e241 100644
--- a/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/dag/TwoInputNode.java
+++ b/stratosphere-compiler/src/main/java/eu/stratosphere/compiler/dag/TwoInputNode.java
@@ -474,12 +474,16 @@ public abstract class TwoInputNode extends OptimizerNode {
if (dps.areCoFulfilled(lpp.getProperties1(), lpp.getProperties2(),
in1.getLocalProperties(), in2.getLocalProperties()))
{
+ // copy, because setting required properties and instantiation may
+ // change the channels and should not affect prior candidates
Channel in1Copy = in1.clone();
in1Copy.setRequiredLocalProps(lpp.getProperties1());
- in2.setRequiredLocalProps(lpp.getProperties2());
+
+ Channel in2Copy = in2.clone();
+ in2Copy.setRequiredLocalProps(lpp.getProperties2());
// all right, co compatible
- instantiate(dps, in1Copy, in2, broadcastPlanChannels, target, estimator, rgps1, rgps2, ilp1, ilp2);
+ instantiate(dps, in1Copy, in2Copy, broadcastPlanChannels, target, estimator, rgps1, rgps2, ilp1, ilp2);
break;
} else {
// meet, but not co-compatible
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/26bc321c/stratosphere-tests/src/test/java/eu/stratosphere/test/compiler/examples/RelationalQueryCompilerTest.java
----------------------------------------------------------------------
diff --git a/stratosphere-tests/src/test/java/eu/stratosphere/test/compiler/examples/RelationalQueryCompilerTest.java b/stratosphere-tests/src/test/java/eu/stratosphere/test/compiler/examples/RelationalQueryCompilerTest.java
index 461a85f..6bac03c 100644
--- a/stratosphere-tests/src/test/java/eu/stratosphere/test/compiler/examples/RelationalQueryCompilerTest.java
+++ b/stratosphere-tests/src/test/java/eu/stratosphere/test/compiler/examples/RelationalQueryCompilerTest.java
@@ -60,6 +60,7 @@ public class RelationalQueryCompilerTest extends CompilerTestBase {
// compile
final OptimizedPlan plan = compileNoStats(p);
+
final OptimizerPlanNodeResolver or = getOptimizerPlanNodeResolver(plan);
// get the nodes from the final plan
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/26bc321c/stratosphere-tests/src/test/java/eu/stratosphere/test/compiler/iterations/ConnectedComponentsTest.java
----------------------------------------------------------------------
diff --git a/stratosphere-tests/src/test/java/eu/stratosphere/test/compiler/iterations/ConnectedComponentsTest.java b/stratosphere-tests/src/test/java/eu/stratosphere/test/compiler/iterations/ConnectedComponentsTest.java
index 9c8c0c8..e9074b4 100644
--- a/stratosphere-tests/src/test/java/eu/stratosphere/test/compiler/iterations/ConnectedComponentsTest.java
+++ b/stratosphere-tests/src/test/java/eu/stratosphere/test/compiler/iterations/ConnectedComponentsTest.java
@@ -102,7 +102,9 @@ public class ConnectedComponentsTest extends CompilerTestBase {
Assert.assertEquals(DriverStrategy.NONE, vertexSource.getDriverStrategy());
Assert.assertEquals(DriverStrategy.NONE, edgesSource.getDriverStrategy());
- Assert.assertEquals(DriverStrategy.HYBRIDHASH_BUILD_SECOND, neighborsJoin.getDriverStrategy());
+ Assert.assertEquals(DriverStrategy.HYBRIDHASH_BUILD_SECOND_CACHED, neighborsJoin.getDriverStrategy());
+ Assert.assertTrue(!neighborsJoin.getInput1().getTempMode().isCached());
+ Assert.assertTrue(!neighborsJoin.getInput2().getTempMode().isCached());
Assert.assertEquals(set0, neighborsJoin.getKeysForInput1());
Assert.assertEquals(set0, neighborsJoin.getKeysForInput2());
@@ -120,7 +122,6 @@ public class ConnectedComponentsTest extends CompilerTestBase {
Assert.assertEquals(ShipStrategyType.FORWARD, neighborsJoin.getInput1().getShipStrategy()); // workset
Assert.assertEquals(ShipStrategyType.PARTITION_HASH, neighborsJoin.getInput2().getShipStrategy()); // edges
Assert.assertEquals(set0, neighborsJoin.getInput2().getShipStrategyKeys());
- Assert.assertTrue(neighborsJoin.getInput2().getTempMode().isCached());
Assert.assertEquals(ShipStrategyType.PARTITION_HASH, minIdReducer.getInput().getShipStrategy());
Assert.assertEquals(set0, minIdReducer.getInput().getShipStrategyKeys());
@@ -182,7 +183,9 @@ public class ConnectedComponentsTest extends CompilerTestBase {
Assert.assertEquals(DriverStrategy.NONE, vertexSource.getDriverStrategy());
Assert.assertEquals(DriverStrategy.NONE, edgesSource.getDriverStrategy());
- Assert.assertEquals(DriverStrategy.HYBRIDHASH_BUILD_SECOND, neighborsJoin.getDriverStrategy());
+ Assert.assertEquals(DriverStrategy.HYBRIDHASH_BUILD_SECOND_CACHED, neighborsJoin.getDriverStrategy());
+ Assert.assertTrue(!neighborsJoin.getInput1().getTempMode().isCached());
+ Assert.assertTrue(!neighborsJoin.getInput2().getTempMode().isCached());
Assert.assertEquals(set0, neighborsJoin.getKeysForInput1());
Assert.assertEquals(set0, neighborsJoin.getKeysForInput2());
@@ -200,7 +203,6 @@ public class ConnectedComponentsTest extends CompilerTestBase {
Assert.assertEquals(ShipStrategyType.FORWARD, neighborsJoin.getInput1().getShipStrategy()); // workset
Assert.assertEquals(ShipStrategyType.PARTITION_HASH, neighborsJoin.getInput2().getShipStrategy()); // edges
Assert.assertEquals(set0, neighborsJoin.getInput2().getShipStrategyKeys());
- Assert.assertTrue(neighborsJoin.getInput2().getTempMode().isCached());
Assert.assertEquals(ShipStrategyType.PARTITION_HASH, minIdReducer.getInput().getShipStrategy());
Assert.assertEquals(set0, minIdReducer.getInput().getShipStrategyKeys());
[07/12] git commit: Add dedicated test for the type extraction on
input formats.
Posted by se...@apache.org.
Add dedicated test for the type extraction on input formats.
Fix various JavaDoc errors.
Project: http://git-wip-us.apache.org/repos/asf/incubator-flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-flink/commit/ce65fb61
Tree: http://git-wip-us.apache.org/repos/asf/incubator-flink/tree/ce65fb61
Diff: http://git-wip-us.apache.org/repos/asf/incubator-flink/diff/ce65fb61
Branch: refs/heads/master
Commit: ce65fb6161321c72310d3ba8c52b31ec557233e9
Parents: 5cec8e8
Author: Stephan Ewen <se...@apache.org>
Authored: Tue Jul 8 17:24:43 2014 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Wed Jul 9 03:09:36 2014 +0200
----------------------------------------------------------------------
.../io/AvroInputFormatTypeExtractionTest.java | 77 +++++++
.../stratosphere/spargel/java/OutgoingEdge.java | 2 +-
.../eu/stratosphere/client/CliFrontend.java | 4 +-
.../eu/stratosphere/client/LocalExecutor.java | 6 +-
.../eu/stratosphere/client/program/Client.java | 1 -
.../client/program/PackagedProgram.java | 20 +-
.../client/web/WebInterfaceServer.java | 4 +-
.../stratosphere/api/common/io/InputFormat.java | 2 +-
.../type/extractor/PojoTypeInformationTest.java | 6 +
.../TypeExtractorInputFormatsTest.java | 230 +++++++++++++++++++
10 files changed, 330 insertions(+), 22 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/ce65fb61/stratosphere-addons/avro/src/test/java/eu/stratosphere/api/java/io/AvroInputFormatTypeExtractionTest.java
----------------------------------------------------------------------
diff --git a/stratosphere-addons/avro/src/test/java/eu/stratosphere/api/java/io/AvroInputFormatTypeExtractionTest.java b/stratosphere-addons/avro/src/test/java/eu/stratosphere/api/java/io/AvroInputFormatTypeExtractionTest.java
new file mode 100644
index 0000000..66467fa
--- /dev/null
+++ b/stratosphere-addons/avro/src/test/java/eu/stratosphere/api/java/io/AvroInputFormatTypeExtractionTest.java
@@ -0,0 +1,77 @@
+/***********************************************************************************************************************
+ *
+ * Copyright (C) 2010-2013 by the Stratosphere project (http://stratosphere.eu)
+ *
+ * Licensed 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 eu.stratosphere.api.java.io;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import eu.stratosphere.api.common.io.InputFormat;
+import eu.stratosphere.api.java.DataSet;
+import eu.stratosphere.api.java.ExecutionEnvironment;
+import eu.stratosphere.api.java.typeutils.PojoTypeInfo;
+import eu.stratosphere.api.java.typeutils.TypeExtractor;
+import eu.stratosphere.core.fs.Path;
+import eu.stratosphere.types.TypeInformation;
+
+public class AvroInputFormatTypeExtractionTest {
+
+ @Test
+ public void testTypeExtraction() {
+ try {
+ InputFormat<MyAvroType, ?> format = new AvroInputFormat<MyAvroType>(new Path("file:///ignore/this/file"), MyAvroType.class);
+
+ TypeInformation<?> typeInfoDirect = TypeExtractor.getInputFormatTypes(format);
+
+ ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+ DataSet<MyAvroType> input = env.createInput(format);
+ TypeInformation<?> typeInfoDataSet = input.getType();
+
+
+ Assert.assertTrue(typeInfoDirect instanceof PojoTypeInfo);
+ Assert.assertTrue(typeInfoDataSet instanceof PojoTypeInfo);
+
+ Assert.assertEquals(MyAvroType.class, typeInfoDirect.getTypeClass());
+ Assert.assertEquals(MyAvroType.class, typeInfoDataSet.getTypeClass());
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ Assert.fail(e.getMessage());
+ }
+ }
+
+ public static final class MyAvroType {
+
+ public String theString;
+
+ private double aDouble;
+
+ public double getaDouble() {
+ return aDouble;
+ }
+
+ public void setaDouble(double aDouble) {
+ this.aDouble = aDouble;
+ }
+
+ public void setTheString(String theString) {
+ this.theString = theString;
+ }
+
+ public String getTheString() {
+ return theString;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/ce65fb61/stratosphere-addons/spargel/src/main/java/eu/stratosphere/spargel/java/OutgoingEdge.java
----------------------------------------------------------------------
diff --git a/stratosphere-addons/spargel/src/main/java/eu/stratosphere/spargel/java/OutgoingEdge.java b/stratosphere-addons/spargel/src/main/java/eu/stratosphere/spargel/java/OutgoingEdge.java
index 9abe0d7..0a63e56 100644
--- a/stratosphere-addons/spargel/src/main/java/eu/stratosphere/spargel/java/OutgoingEdge.java
+++ b/stratosphere-addons/spargel/src/main/java/eu/stratosphere/spargel/java/OutgoingEdge.java
@@ -16,7 +16,7 @@ package eu.stratosphere.spargel.java;
* <tt>Edge</tt> objects represent edges between vertices. Edges are defined by their source and target
* vertex id. Edges may have an associated value (for example a weight or a distance), if the
* graph algorithm was initialized with the
- * {@link VertexCentricIteration#withEdgesWithValue(eu.stratosphere.api.java.DataSet, VertexUpdateFunction, MessagingFunction)}
+ * {@link VertexCentricIteration#withValuedEdges(eu.stratosphere.api.java.DataSet, VertexUpdateFunction, MessagingFunction, int)}
* method.
*
* @param <VertexKey> The type of the vertex key.
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/ce65fb61/stratosphere-clients/src/main/java/eu/stratosphere/client/CliFrontend.java
----------------------------------------------------------------------
diff --git a/stratosphere-clients/src/main/java/eu/stratosphere/client/CliFrontend.java b/stratosphere-clients/src/main/java/eu/stratosphere/client/CliFrontend.java
index a3f73db..7d97143 100644
--- a/stratosphere-clients/src/main/java/eu/stratosphere/client/CliFrontend.java
+++ b/stratosphere-clients/src/main/java/eu/stratosphere/client/CliFrontend.java
@@ -815,8 +815,6 @@ public class CliFrontend {
/**
* Prints the help for the client.
- *
- * @param options A map with options for actions.
*/
private void printHelp() {
System.out.println("./stratosphere <ACTION> [GENERAL_OPTIONS] [ARGUMENTS]");
@@ -878,7 +876,7 @@ public class CliFrontend {
/**
* Displays exceptions.
*
- * @param e the exception to display.
+ * @param t The exception to display.
*/
private int handleError(Throwable t) {
System.out.println("Error: " + t.getMessage());
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/ce65fb61/stratosphere-clients/src/main/java/eu/stratosphere/client/LocalExecutor.java
----------------------------------------------------------------------
diff --git a/stratosphere-clients/src/main/java/eu/stratosphere/client/LocalExecutor.java b/stratosphere-clients/src/main/java/eu/stratosphere/client/LocalExecutor.java
index 0f50f96..6822e10 100644
--- a/stratosphere-clients/src/main/java/eu/stratosphere/client/LocalExecutor.java
+++ b/stratosphere-clients/src/main/java/eu/stratosphere/client/LocalExecutor.java
@@ -285,7 +285,7 @@ public class LocalExecutor extends PlanExecutor {
/**
* Executes the program represented by the given Pact plan.
*
- * @param pa The program's plan.
+ * @param plan The program's plan.
* @return The net runtime of the program, in milliseconds.
*
* @throws Exception Thrown, if either the startup of the local execution context, or the execution
@@ -325,7 +325,9 @@ public class LocalExecutor extends PlanExecutor {
/**
* Return unoptimized plan as JSON.
- * @return
+ *
+ * @param plan The program plan.
+ * @return The plan as a JSON object.
*/
public static String getPlanAsJSON(Plan plan) {
PlanJSONDumpGenerator gen = new PlanJSONDumpGenerator();
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/ce65fb61/stratosphere-clients/src/main/java/eu/stratosphere/client/program/Client.java
----------------------------------------------------------------------
diff --git a/stratosphere-clients/src/main/java/eu/stratosphere/client/program/Client.java b/stratosphere-clients/src/main/java/eu/stratosphere/client/program/Client.java
index ec66f4a..16dfd60 100644
--- a/stratosphere-clients/src/main/java/eu/stratosphere/client/program/Client.java
+++ b/stratosphere-clients/src/main/java/eu/stratosphere/client/program/Client.java
@@ -270,7 +270,6 @@ public class Client {
* or if the submission failed. That might be either due to an I/O problem,
* i.e. the job-manager is unreachable, or due to the fact that the execution
* on the nephele system failed.
- * @throws JobInstantiationException Thrown, if the plan assembler function causes an exception.
*/
public JobExecutionResult run(JobWithJars prog, int parallelism, boolean wait) throws CompilerException, ProgramInvocationException {
return run(getOptimizedPlan(prog, parallelism), prog.getJarFiles(), wait);
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/ce65fb61/stratosphere-clients/src/main/java/eu/stratosphere/client/program/PackagedProgram.java
----------------------------------------------------------------------
diff --git a/stratosphere-clients/src/main/java/eu/stratosphere/client/program/PackagedProgram.java b/stratosphere-clients/src/main/java/eu/stratosphere/client/program/PackagedProgram.java
index edf36b3..9fe828a 100644
--- a/stratosphere-clients/src/main/java/eu/stratosphere/client/program/PackagedProgram.java
+++ b/stratosphere-clients/src/main/java/eu/stratosphere/client/program/PackagedProgram.java
@@ -102,7 +102,7 @@ public class PackagedProgram {
*
* @param jarFile
* The jar file which contains the plan.
- * @param className
+ * @param entryPointClassName
* Name of the class which generates the plan. Overrides the class defined
* in the jar file manifest
* @param args
@@ -177,7 +177,8 @@ public class PackagedProgram {
/**
* Returns the plan with all required jars.
- * @throws JobInstantiationException
+ *
+ * @return The plan with attached jar files.
* @throws ProgramInvocationException
*/
public JobWithJars getPlanWithJars() throws ProgramInvocationException {
@@ -261,9 +262,6 @@ public class PackagedProgram {
* @throws ProgramInvocationException
* This invocation is thrown if the Program can't be properly loaded. Causes
* may be a missing / wrong class or manifest files.
- * @throws JobInstantiationException
- * Thrown if an error occurred in the user-provided pact assembler. This may indicate
- * missing parameters for generation.
*/
public String getDescription() throws ProgramInvocationException {
if (ProgramDescription.class.isAssignableFrom(this.mainClass)) {
@@ -487,18 +485,16 @@ public class PackagedProgram {
* or it is read from the manifest of the jar. The assembler is handed the given options
* for its assembly.
*
- * @param clazz
- * The name of the assembler class, or null, if the class should be read from
- * the manifest.
+ * @param program The program to create the plan for.
* @param options
* The options for the assembler.
* @return The plan created by the program.
- * @throws JobInstantiationException
+ * @throws ProgramInvocationException
* Thrown, if an error occurred in the user-provided pact assembler.
*/
- private static Plan createPlanFromProgram(Program assembler, String[] options) throws ProgramInvocationException {
+ private static Plan createPlanFromProgram(Program program, String[] options) throws ProgramInvocationException {
try {
- return assembler.getPlan(options);
+ return program.getPlan(options);
} catch (Throwable t) {
throw new ProgramInvocationException("Error while calling the program: " + t.getMessage(), t);
}
@@ -509,7 +505,7 @@ public class PackagedProgram {
* to the system's temp directory.
*
* @return The file names of the extracted temporary files.
- * @throws IOException Thrown, if the extraction process failed.
+ * @throws ProgramInvocationException Thrown, if the extraction process failed.
*/
private static List<File> extractContainedLibaries(File jarFile) throws ProgramInvocationException {
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/ce65fb61/stratosphere-clients/src/main/java/eu/stratosphere/client/web/WebInterfaceServer.java
----------------------------------------------------------------------
diff --git a/stratosphere-clients/src/main/java/eu/stratosphere/client/web/WebInterfaceServer.java b/stratosphere-clients/src/main/java/eu/stratosphere/client/web/WebInterfaceServer.java
index 7fe8abb..587bfb6 100644
--- a/stratosphere-clients/src/main/java/eu/stratosphere/client/web/WebInterfaceServer.java
+++ b/stratosphere-clients/src/main/java/eu/stratosphere/client/web/WebInterfaceServer.java
@@ -247,8 +247,8 @@ public class WebInterfaceServer {
* if the directory exists and creates it if necessary. It also checks read permissions and
* write permission, if necessary.
*
- * @param dir
- * The String describing the directory path.
+ * @param f
+ * The file describing the directory path.
* @param needWritePermission
* A flag indicating whether to check write access.
* @throws IOException
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/ce65fb61/stratosphere-core/src/main/java/eu/stratosphere/api/common/io/InputFormat.java
----------------------------------------------------------------------
diff --git a/stratosphere-core/src/main/java/eu/stratosphere/api/common/io/InputFormat.java b/stratosphere-core/src/main/java/eu/stratosphere/api/common/io/InputFormat.java
index ce3cd95..247ac10 100644
--- a/stratosphere-core/src/main/java/eu/stratosphere/api/common/io/InputFormat.java
+++ b/stratosphere-core/src/main/java/eu/stratosphere/api/common/io/InputFormat.java
@@ -88,7 +88,7 @@ public interface InputFormat<OT, T extends InputSplit> extends Serializable {
* instances may remain idle.
* @return The splits of this input that can be processed in parallel.
*
- * @throws Exception Thrown, when the creation of the splits was erroneous.
+ * @throws IOException Thrown, when the creation of the splits was erroneous.
*/
T[] createInputSplits(int minNumSplits) throws IOException;
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/ce65fb61/stratosphere-java/src/test/java/eu/stratosphere/api/java/type/extractor/PojoTypeInformationTest.java
----------------------------------------------------------------------
diff --git a/stratosphere-java/src/test/java/eu/stratosphere/api/java/type/extractor/PojoTypeInformationTest.java b/stratosphere-java/src/test/java/eu/stratosphere/api/java/type/extractor/PojoTypeInformationTest.java
index 772050a..6907077 100644
--- a/stratosphere-java/src/test/java/eu/stratosphere/api/java/type/extractor/PojoTypeInformationTest.java
+++ b/stratosphere-java/src/test/java/eu/stratosphere/api/java/type/extractor/PojoTypeInformationTest.java
@@ -84,4 +84,10 @@ public class PojoTypeInformationTest {
TypeInformation<Recursive1Pojo> type = TypeExtractor.getForClass(Recursive1Pojo.class);
assertTrue("Extracted type is not a Pojo type but should be.", type instanceof PojoTypeInfo);
}
+
+ @Test
+ public void testRecursivePojoObjectTypeExtraction() {
+ TypeInformation<Recursive1Pojo> type = TypeExtractor.getForObject(new Recursive1Pojo());
+ assertTrue("Extracted type is not a Pojo type but should be.", type instanceof PojoTypeInfo);
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/ce65fb61/stratosphere-java/src/test/java/eu/stratosphere/api/java/type/extractor/TypeExtractorInputFormatsTest.java
----------------------------------------------------------------------
diff --git a/stratosphere-java/src/test/java/eu/stratosphere/api/java/type/extractor/TypeExtractorInputFormatsTest.java b/stratosphere-java/src/test/java/eu/stratosphere/api/java/type/extractor/TypeExtractorInputFormatsTest.java
new file mode 100644
index 0000000..d2c3642
--- /dev/null
+++ b/stratosphere-java/src/test/java/eu/stratosphere/api/java/type/extractor/TypeExtractorInputFormatsTest.java
@@ -0,0 +1,230 @@
+/***********************************************************************************************************************
+ *
+ * Copyright (C) 2010-2013 by the Stratosphere project (http://stratosphere.eu)
+ *
+ * Licensed 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 eu.stratosphere.api.java.type.extractor;
+
+
+import static org.junit.Assert.*;
+
+import java.io.IOException;
+
+import org.junit.Test;
+
+import eu.stratosphere.api.common.io.GenericInputFormat;
+import eu.stratosphere.api.common.io.InputFormat;
+import eu.stratosphere.api.common.io.statistics.BaseStatistics;
+import eu.stratosphere.api.java.tuple.Tuple3;
+import eu.stratosphere.api.java.typeutils.BasicTypeInfo;
+import eu.stratosphere.api.java.typeutils.ResultTypeQueryable;
+import eu.stratosphere.api.java.typeutils.TupleTypeInfo;
+import eu.stratosphere.api.java.typeutils.TypeExtractor;
+import eu.stratosphere.configuration.Configuration;
+import eu.stratosphere.core.io.InputSplit;
+import eu.stratosphere.types.TypeInformation;
+
+@SuppressWarnings("serial")
+public class TypeExtractorInputFormatsTest {
+
+ @Test
+ public void testExtractInputFormatType() {
+ try {
+ InputFormat<?, ?> format = new DummyFloatInputFormat();
+ TypeInformation<?> typeInfo = TypeExtractor.getInputFormatTypes(format);
+ assertEquals(BasicTypeInfo.FLOAT_TYPE_INFO, typeInfo);
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testExtractDerivedInputFormatType() {
+ try {
+ // simple type
+ {
+ InputFormat<?, ?> format = new DerivedInputFormat();
+ TypeInformation<?> typeInfo = TypeExtractor.getInputFormatTypes(format);
+ assertEquals(BasicTypeInfo.SHORT_TYPE_INFO, typeInfo);
+ }
+
+ // composite type
+ {
+ InputFormat<?, ?> format = new DerivedTupleInputFormat();
+ TypeInformation<?> typeInfo = TypeExtractor.getInputFormatTypes(format);
+
+ assertTrue(typeInfo.isTupleType());
+ assertTrue(typeInfo instanceof TupleTypeInfo);
+
+ @SuppressWarnings("unchecked")
+ TupleTypeInfo<Tuple3<String, Short, Double>> tupleInfo = (TupleTypeInfo<Tuple3<String, Short, Double>>) typeInfo;
+
+ assertEquals(3, tupleInfo.getArity());
+ assertEquals(BasicTypeInfo.STRING_TYPE_INFO, tupleInfo.getTypeAt(0));
+ assertEquals(BasicTypeInfo.SHORT_TYPE_INFO, tupleInfo.getTypeAt(1));
+ assertEquals(BasicTypeInfo.DOUBLE_TYPE_INFO, tupleInfo.getTypeAt(2));
+ }
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testMultiLevelDerivedInputFormatType() {
+ try {
+
+ // composite type
+ {
+ InputFormat<?, ?> format = new FinalRelativeInputFormat();
+ TypeInformation<?> typeInfo = TypeExtractor.getInputFormatTypes(format);
+
+ assertTrue(typeInfo.isTupleType());
+ assertTrue(typeInfo instanceof TupleTypeInfo);
+
+ @SuppressWarnings("unchecked")
+ TupleTypeInfo<Tuple3<String, Integer, Double>> tupleInfo = (TupleTypeInfo<Tuple3<String, Integer, Double>>) typeInfo;
+
+ assertEquals(3, tupleInfo.getArity());
+ assertEquals(BasicTypeInfo.STRING_TYPE_INFO, tupleInfo.getTypeAt(0));
+ assertEquals(BasicTypeInfo.INT_TYPE_INFO, tupleInfo.getTypeAt(1));
+ assertEquals(BasicTypeInfo.DOUBLE_TYPE_INFO, tupleInfo.getTypeAt(2));
+ }
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testQueryableFormatType() {
+ try {
+ InputFormat<?, ?> format = new QueryableInputFormat();
+ TypeInformation<?> typeInfo = TypeExtractor.getInputFormatTypes(format);
+ assertEquals(BasicTypeInfo.DOUBLE_TYPE_INFO, typeInfo);
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Test formats
+ // --------------------------------------------------------------------------------------------
+
+ public static final class DummyFloatInputFormat implements InputFormat<Float, InputSplit> {
+
+ @Override
+ public void configure(Configuration parameters) {}
+
+ @Override
+ public BaseStatistics getStatistics(BaseStatistics cachedStatistics) { return null; }
+
+ @Override
+ public InputSplit[] createInputSplits(int minNumSplits) { return null; }
+
+ @Override
+ public Class<? extends InputSplit> getInputSplitType() { return null; }
+
+ @Override
+ public void open(InputSplit split) {}
+
+ @Override
+ public boolean reachedEnd() { return false; }
+
+ @Override
+ public Float nextRecord(Float reuse) throws IOException { return null; }
+
+ @Override
+ public void close() {}
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ public static final class DerivedInputFormat extends GenericInputFormat<Short> {
+
+ @Override
+ public boolean reachedEnd() { return false; }
+
+ @Override
+ public Short nextRecord(Short reuse) { return null; }
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ public static final class DerivedTupleInputFormat extends GenericInputFormat<Tuple3<String, Short, Double>> {
+
+ @Override
+ public boolean reachedEnd() { return false; }
+
+ @Override
+ public Tuple3<String, Short, Double> nextRecord(Tuple3<String, Short, Double> reuse) { return null; }
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ public static class RelativeInputFormat<T> extends GenericInputFormat<Tuple3<String, T, Double>> {
+
+ @Override
+ public boolean reachedEnd() { return false; }
+
+ @Override
+ public Tuple3<String, T, Double> nextRecord(Tuple3<String, T, Double> reuse) { return null; }
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ public static final class FinalRelativeInputFormat extends RelativeInputFormat<Integer> {
+
+ @Override
+ public Tuple3<String, Integer, Double> nextRecord(Tuple3<String, Integer, Double> reuse) { return null; }
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ public static final class QueryableInputFormat implements InputFormat<Float, InputSplit>, ResultTypeQueryable<Double> {
+
+ @Override
+ public void configure(Configuration parameters) {}
+
+ @Override
+ public BaseStatistics getStatistics(BaseStatistics cachedStatistics) { return null; }
+
+ @Override
+ public InputSplit[] createInputSplits(int minNumSplits) { return null; }
+
+ @Override
+ public Class<? extends InputSplit> getInputSplitType() { return null; }
+
+ @Override
+ public void open(InputSplit split) {}
+
+ @Override
+ public boolean reachedEnd() { return false; }
+
+ @Override
+ public Float nextRecord(Float reuse) throws IOException { return null; }
+
+ @Override
+ public void close() {}
+
+ @Override
+ public TypeInformation<Double> getProducedType() {
+ return BasicTypeInfo.DOUBLE_TYPE_INFO;
+ }
+ }
+}