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 2015/01/08 11:59:04 UTC
[09/13] flink git commit: [FLINK-1285] Make execution mode
configurable
http://git-wip-us.apache.org/repos/asf/flink/blob/b7b32a05/flink-runtime/src/main/java/org/apache/flink/runtime/operators/JoinWithSolutionSetFirstDriver.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/JoinWithSolutionSetFirstDriver.java b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/JoinWithSolutionSetFirstDriver.java
index 4f9f1ba..a1c8a4a 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/JoinWithSolutionSetFirstDriver.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/JoinWithSolutionSetFirstDriver.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.operators;
+import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.functions.FlatJoinFunction;
import org.apache.flink.api.common.operators.util.JoinHashMap;
import org.apache.flink.api.common.typeutils.TypeComparator;
@@ -48,7 +49,9 @@ public class JoinWithSolutionSetFirstDriver<IT1, IT2, OT> implements ResettableP
private IT2 probeSideRecord;
protected volatile boolean running;
-
+
+ private boolean objectReuseEnabled = false;
+
// --------------------------------------------------------------------------------------------
@Override
@@ -125,10 +128,15 @@ public class JoinWithSolutionSetFirstDriver<IT1, IT2, OT> implements ResettableP
TypeComparatorFactory<IT2> probeSideComparatorFactory = config.getDriverComparator(0, classLoader);
this.probeSideComparator = probeSideComparatorFactory.createComparator();
-
- solutionSideRecord = solutionSetSerializer.createInstance();
- probeSideRecord = probeSideSerializer.createInstance();
-
+
+ ExecutionConfig executionConfig = taskContext.getExecutionConfig();
+ objectReuseEnabled = executionConfig.isObjectReuseEnabled();
+
+ if (objectReuseEnabled) {
+ solutionSideRecord = solutionSetSerializer.createInstance();
+ probeSideRecord = probeSideSerializer.createInstance();
+ }
+
TypePairComparatorFactory<IT1, IT2> factory = taskContext.getTaskConfig().getPairComparatorFactory(taskContext.getUserCodeClassLoader());
pairComparator = factory.createComparator21(solutionSetComparator, this.probeSideComparator);
}
@@ -146,33 +154,60 @@ public class JoinWithSolutionSetFirstDriver<IT1, IT2, OT> implements ResettableP
final Collector<OT> collector = taskContext.getOutputCollector();
final MutableObjectIterator<IT2> probeSideInput = taskContext.<IT2>getInput(0);
-
- IT2 probeSideRecord = this.probeSideRecord;
-
- if (hashTable != null) {
- final CompactingHashTable<IT1> join = hashTable;
- final CompactingHashTable<IT1>.HashTableProber<IT2> prober = join.getProber(probeSideComparator, pairComparator);
-
- IT1 buildSideRecord = this.solutionSideRecord;
-
- while (this.running && ((probeSideRecord = probeSideInput.next(probeSideRecord)) != null)) {
- buildSideRecord = prober.getMatchFor(probeSideRecord, buildSideRecord);
- joinFunction.join(buildSideRecord, probeSideRecord, collector);
+
+ if (objectReuseEnabled) {
+ IT2 probeSideRecord = this.probeSideRecord;
+
+ if (hashTable != null) {
+ final CompactingHashTable<IT1> join = hashTable;
+ final CompactingHashTable<IT1>.HashTableProber<IT2> prober = join.getProber(probeSideComparator, pairComparator);
+
+
+ IT1 buildSideRecord = this.solutionSideRecord;
+
+ while (this.running && ((probeSideRecord = probeSideInput.next(probeSideRecord)) != null)) {
+ buildSideRecord = prober.getMatchFor(probeSideRecord, buildSideRecord);
+ joinFunction.join(buildSideRecord, probeSideRecord, collector);
+ }
+ } else if (objectMap != null) {
+ final JoinHashMap<IT1> hashTable = this.objectMap;
+ final JoinHashMap<IT1>.Prober<IT2> prober = this.objectMap.createProber(probeSideComparator, pairComparator);
+ final TypeSerializer<IT1> buildSerializer = hashTable.getBuildSerializer();
+
+ while (this.running && ((probeSideRecord = probeSideInput.next(probeSideRecord)) != null)) {
+ IT1 match = prober.lookupMatch(probeSideRecord);
+ joinFunction.join(buildSerializer.copy(match), probeSideRecord, collector);
+ }
+ } else {
+ throw new RuntimeException();
}
- }
- else if (objectMap != null) {
- final JoinHashMap<IT1> hashTable = this.objectMap;
- final JoinHashMap<IT1>.Prober<IT2> prober = this.objectMap.createProber(probeSideComparator, pairComparator);
- final TypeSerializer<IT1> buildSerializer = hashTable.getBuildSerializer();
-
- while (this.running && ((probeSideRecord = probeSideInput.next(probeSideRecord)) != null)) {
- IT1 match = prober.lookupMatch(probeSideRecord);
- joinFunction.join(buildSerializer.copy(match), probeSideRecord, collector);
+ } else {
+ IT2 probeSideRecord;
+
+ if (hashTable != null) {
+ final CompactingHashTable<IT1> join = hashTable;
+ final CompactingHashTable<IT1>.HashTableProber<IT2> prober = join.getProber(probeSideComparator, pairComparator);
+
+
+ IT1 buildSideRecord;
+
+ while (this.running && ((probeSideRecord = probeSideInput.next()) != null)) {
+ buildSideRecord = prober.getMatchFor(probeSideRecord);
+ joinFunction.join(buildSideRecord, probeSideRecord, collector);
+ }
+ } else if (objectMap != null) {
+ final JoinHashMap<IT1> hashTable = this.objectMap;
+ final JoinHashMap<IT1>.Prober<IT2> prober = this.objectMap.createProber(probeSideComparator, pairComparator);
+ final TypeSerializer<IT1> buildSerializer = hashTable.getBuildSerializer();
+
+ while (this.running && ((probeSideRecord = probeSideInput.next()) != null)) {
+ IT1 match = prober.lookupMatch(probeSideRecord);
+ joinFunction.join(buildSerializer.copy(match), probeSideRecord, collector);
+ }
+ } else {
+ throw new RuntimeException();
}
}
- else {
- throw new RuntimeException();
- }
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/b7b32a05/flink-runtime/src/main/java/org/apache/flink/runtime/operators/JoinWithSolutionSetSecondDriver.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/JoinWithSolutionSetSecondDriver.java b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/JoinWithSolutionSetSecondDriver.java
index 7866cab..32a75dc 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/JoinWithSolutionSetSecondDriver.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/JoinWithSolutionSetSecondDriver.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.operators;
+import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.functions.FlatJoinFunction;
import org.apache.flink.api.common.operators.util.JoinHashMap;
import org.apache.flink.api.common.typeutils.TypeComparator;
@@ -48,7 +49,9 @@ public class JoinWithSolutionSetSecondDriver<IT1, IT2, OT> implements Resettable
private IT1 probeSideRecord;
protected volatile boolean running;
-
+
+ private boolean objectReuseEnabled = false;
+
// --------------------------------------------------------------------------------------------
@Override
@@ -128,8 +131,13 @@ public class JoinWithSolutionSetSecondDriver<IT1, IT2, OT> implements Resettable
this.probeSideComparator = probeSideComparatorFactory.createComparator();
- solutionSideRecord = solutionSetSerializer.createInstance();
- probeSideRecord = probeSideSerializer.createInstance();
+ ExecutionConfig executionConfig = taskContext.getExecutionConfig();
+ objectReuseEnabled = executionConfig.isObjectReuseEnabled();
+
+ if (objectReuseEnabled) {
+ solutionSideRecord = solutionSetSerializer.createInstance();
+ probeSideRecord = probeSideSerializer.createInstance();
+ }
TypePairComparatorFactory<IT1, IT2> factory = taskContext.getTaskConfig().getPairComparatorFactory(taskContext.getUserCodeClassLoader());
pairComparator = factory.createComparator12(this.probeSideComparator, solutionSetComparator);
@@ -148,32 +156,60 @@ public class JoinWithSolutionSetSecondDriver<IT1, IT2, OT> implements Resettable
final FlatJoinFunction<IT1, IT2, OT> joinFunction = taskContext.getStub();
final Collector<OT> collector = taskContext.getOutputCollector();
final MutableObjectIterator<IT1> probeSideInput = taskContext.getInput(0);
-
- IT1 probeSideRecord = this.probeSideRecord;
-
- if (hashTable != null) {
- final CompactingHashTable<IT2> join = hashTable;
- final CompactingHashTable<IT2>.HashTableProber<IT1> prober = join.getProber(probeSideComparator, pairComparator);
-
- IT2 buildSideRecord = this.solutionSideRecord;
-
- while (this.running && ((probeSideRecord = probeSideInput.next(probeSideRecord)) != null)) {
- buildSideRecord = prober.getMatchFor(probeSideRecord, buildSideRecord);
- joinFunction.join(probeSideRecord, buildSideRecord, collector);
+
+ if (objectReuseEnabled) {
+ IT1 probeSideRecord = this.probeSideRecord;
+
+ if (hashTable != null) {
+ final CompactingHashTable<IT2> join = hashTable;
+ final CompactingHashTable<IT2>.HashTableProber<IT1> prober = join.getProber(probeSideComparator, pairComparator);
+
+
+ IT2 buildSideRecord = this.solutionSideRecord;
+
+ while (this.running && ((probeSideRecord = probeSideInput.next(probeSideRecord)) != null)) {
+ buildSideRecord = prober.getMatchFor(probeSideRecord, buildSideRecord);
+ joinFunction.join(probeSideRecord, buildSideRecord, collector);
+ }
+ } else if (objectMap != null) {
+ final JoinHashMap<IT2> hashTable = this.objectMap;
+ final JoinHashMap<IT2>.Prober<IT1> prober = this.objectMap.createProber(probeSideComparator, pairComparator);
+ final TypeSerializer<IT2> buildSerializer = hashTable.getBuildSerializer();
+
+ while (this.running && ((probeSideRecord = probeSideInput.next(probeSideRecord)) != null)) {
+ IT2 match = prober.lookupMatch(probeSideRecord);
+ joinFunction.join(probeSideRecord, buildSerializer.copy(match), collector);
+ }
+ } else {
+ throw new RuntimeException();
}
- }
- else if (objectMap != null) {
- final JoinHashMap<IT2> hashTable = this.objectMap;
- final JoinHashMap<IT2>.Prober<IT1> prober = this.objectMap.createProber(probeSideComparator, pairComparator);
- final TypeSerializer<IT2> buildSerializer = hashTable.getBuildSerializer();
-
- while (this.running && ((probeSideRecord = probeSideInput.next(probeSideRecord)) != null)) {
- IT2 match = prober.lookupMatch(probeSideRecord);
- joinFunction.join(probeSideRecord, buildSerializer.copy(match), collector);
+ } else {
+ IT1 probeSideRecord;
+
+ if (hashTable != null) {
+ final CompactingHashTable<IT2> join = hashTable;
+ final CompactingHashTable<IT2>.HashTableProber<IT1> prober = join.getProber(probeSideComparator, pairComparator);
+
+
+ IT2 buildSideRecord;
+
+ while (this.running && ((probeSideRecord = probeSideInput.next()) != null)) {
+ buildSideRecord = prober.getMatchFor(probeSideRecord);
+ joinFunction.join(probeSideRecord, buildSideRecord, collector);
+ }
+ } else if (objectMap != null) {
+ final JoinHashMap<IT2> hashTable = this.objectMap;
+ final JoinHashMap<IT2>.Prober<IT1> prober = this.objectMap.createProber(probeSideComparator, pairComparator);
+ final TypeSerializer<IT2> buildSerializer = hashTable.getBuildSerializer();
+
+ while (this.running && ((probeSideRecord = probeSideInput.next()) != null)) {
+ IT2 match = prober.lookupMatch(probeSideRecord);
+ joinFunction.join(probeSideRecord, buildSerializer.copy(match), collector);
+ }
+ } else {
+ throw new RuntimeException();
}
- }
- else {
- throw new RuntimeException();
+
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/b7b32a05/flink-runtime/src/main/java/org/apache/flink/runtime/operators/MapDriver.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/MapDriver.java b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/MapDriver.java
index 2d98954..d6679dd 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/MapDriver.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/MapDriver.java
@@ -19,6 +19,7 @@
package org.apache.flink.runtime.operators;
+import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.MutableObjectIterator;
@@ -41,12 +42,17 @@ public class MapDriver<IT, OT> implements PactDriver<MapFunction<IT, OT>, OT> {
private PactTaskContext<MapFunction<IT, OT>, OT> taskContext;
private volatile boolean running;
+
+ private boolean objectReuseEnabled = false;
@Override
public void setup(PactTaskContext<MapFunction<IT, OT>, OT> context) {
this.taskContext = context;
this.running = true;
+
+ ExecutionConfig executionConfig = taskContext.getExecutionConfig();
+ this.objectReuseEnabled = executionConfig.isObjectReuseEnabled();
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/b7b32a05/flink-runtime/src/main/java/org/apache/flink/runtime/operators/MapPartitionDriver.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/MapPartitionDriver.java b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/MapPartitionDriver.java
index 41aa312..401c951 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/MapPartitionDriver.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/MapPartitionDriver.java
@@ -19,10 +19,14 @@
package org.apache.flink.runtime.operators;
+import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.functions.MapPartitionFunction;
-import org.apache.flink.runtime.util.MutableToRegularIteratorWrapper;
+import org.apache.flink.runtime.util.NonReusingMutableToRegularIteratorWrapper;
+import org.apache.flink.runtime.util.ReusingMutableToRegularIteratorWrapper;
import org.apache.flink.util.Collector;
import org.apache.flink.util.MutableObjectIterator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* MapPartition task which is executed by a Nephele task manager. The task has a single
@@ -39,8 +43,12 @@ import org.apache.flink.util.MutableObjectIterator;
*/
public class MapPartitionDriver<IT, OT> implements PactDriver<MapPartitionFunction<IT, OT>, OT> {
+ private static final Logger LOG = LoggerFactory.getLogger(MapPartitionDriver.class);
+
private PactTaskContext<MapPartitionFunction<IT, OT>, OT> taskContext;
+ private boolean objectReuseEnabled = false;
+
@Override
public void setup(PactTaskContext<MapPartitionFunction<IT, OT>, OT> context) {
this.taskContext = context;
@@ -65,7 +73,12 @@ public class MapPartitionDriver<IT, OT> implements PactDriver<MapPartitionFuncti
@Override
public void prepare() {
- // nothing, since a mapper does not need any preparation
+ ExecutionConfig executionConfig = taskContext.getExecutionConfig();
+ this.objectReuseEnabled = executionConfig.isObjectReuseEnabled();
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("MapPartitionDriver object reuse: " + (this.objectReuseEnabled ? "ENABLED" : "DISABLED") + ".");
+ }
}
@Override
@@ -75,8 +88,15 @@ public class MapPartitionDriver<IT, OT> implements PactDriver<MapPartitionFuncti
final MapPartitionFunction<IT, OT> function = this.taskContext.getStub();
final Collector<OT> output = this.taskContext.getOutputCollector();
- final MutableToRegularIteratorWrapper<IT> inIter = new MutableToRegularIteratorWrapper<IT>(input, this.taskContext.<IT>getInputSerializer(0).getSerializer());
- function.mapPartition(inIter, output);
+ if (objectReuseEnabled) {
+ final ReusingMutableToRegularIteratorWrapper<IT> inIter = new ReusingMutableToRegularIteratorWrapper<IT>(input, this.taskContext.<IT>getInputSerializer(0).getSerializer());
+
+ function.mapPartition(inIter, output);
+ } else {
+ final NonReusingMutableToRegularIteratorWrapper<IT> inIter = new NonReusingMutableToRegularIteratorWrapper<IT>(input, this.taskContext.<IT>getInputSerializer(0).getSerializer());
+
+ function.mapPartition(inIter, output);
+ }
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/b7b32a05/flink-runtime/src/main/java/org/apache/flink/runtime/operators/MatchDriver.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/MatchDriver.java b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/MatchDriver.java
index 82f24d7..2d051ad 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/MatchDriver.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/MatchDriver.java
@@ -19,6 +19,9 @@
package org.apache.flink.runtime.operators;
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.runtime.operators.hash.NonReusingBuildFirstHashMatchIterator;
+import org.apache.flink.runtime.operators.hash.NonReusingBuildSecondHashMatchIterator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.flink.api.common.functions.FlatJoinFunction;
@@ -27,8 +30,8 @@ import org.apache.flink.api.common.typeutils.TypePairComparatorFactory;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.runtime.io.disk.iomanager.IOManager;
import org.apache.flink.runtime.memorymanager.MemoryManager;
-import org.apache.flink.runtime.operators.hash.BuildFirstHashMatchIterator;
-import org.apache.flink.runtime.operators.hash.BuildSecondHashMatchIterator;
+import org.apache.flink.runtime.operators.hash.ReusingBuildFirstHashMatchIterator;
+import org.apache.flink.runtime.operators.hash.ReusingBuildSecondHashMatchIterator;
import org.apache.flink.runtime.operators.sort.MergeMatchIterator;
import org.apache.flink.runtime.operators.util.JoinTaskIterator;
import org.apache.flink.runtime.operators.util.TaskConfig;
@@ -53,7 +56,9 @@ public class MatchDriver<IT1, IT2, OT> implements PactDriver<FlatJoinFunction<IT
private volatile JoinTaskIterator<IT1, IT2, OT> matchIterator; // the iterator that does the actual matching
protected volatile boolean running;
-
+
+ private boolean objectReuseEnabled = false;
+
// ------------------------------------------------------------------------
@Override
@@ -96,7 +101,7 @@ public class MatchDriver<IT1, IT2, OT> implements PactDriver<FlatJoinFunction<IT
final MutableObjectIterator<IT1> in1 = this.taskContext.getInput(0);
final MutableObjectIterator<IT2> in2 = this.taskContext.getInput(1);
-
+
// get the key positions and types
final TypeSerializer<IT1> serializer1 = this.taskContext.<IT1>getInputSerializer(0).getSerializer();
final TypeSerializer<IT2> serializer2 = this.taskContext.<IT2>getInputSerializer(1).getSerializer();
@@ -109,25 +114,44 @@ public class MatchDriver<IT1, IT2, OT> implements PactDriver<FlatJoinFunction<IT
throw new Exception("Missing pair comparator factory for Match driver");
}
+ ExecutionConfig executionConfig = taskContext.getExecutionConfig();
+ this.objectReuseEnabled = executionConfig.isObjectReuseEnabled();
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("MatchDriver object reuse: " + (this.objectReuseEnabled ? "ENABLED" : "DISABLED") + ".");
+ }
+
// create and return MatchTaskIterator according to provided local strategy.
- switch (ls) {
- case MERGE:
- this.matchIterator = new MergeMatchIterator<IT1, IT2, OT>(in1, in2, serializer1, comparator1,
- serializer2, comparator2, pairComparatorFactory.createComparator12(comparator1, comparator2),
- memoryManager, ioManager, numPages, this.taskContext.getOwningNepheleTask());
- break;
- case HYBRIDHASH_BUILD_FIRST:
- this.matchIterator = new BuildFirstHashMatchIterator<IT1, IT2, OT>(in1, in2, serializer1, comparator1,
- serializer2, comparator2, pairComparatorFactory.createComparator21(comparator1, comparator2),
- memoryManager, ioManager, this.taskContext.getOwningNepheleTask(), fractionAvailableMemory);
- break;
- case HYBRIDHASH_BUILD_SECOND:
- this.matchIterator = new BuildSecondHashMatchIterator<IT1, IT2, OT>(in1, in2, serializer1, comparator1,
- serializer2, comparator2, pairComparatorFactory.createComparator12(comparator1, comparator2),
- memoryManager, ioManager, this.taskContext.getOwningNepheleTask(), fractionAvailableMemory);
- break;
- default:
- throw new Exception("Unsupported driver strategy for Match driver: " + ls.name());
+ if (this.objectReuseEnabled) {
+ switch (ls) {
+ case MERGE:
+ this.matchIterator = new MergeMatchIterator<IT1, IT2, OT>(in1, in2, serializer1, comparator1, serializer2, comparator2, pairComparatorFactory.createComparator12(comparator1, comparator2), memoryManager, ioManager, numPages, this.taskContext.getOwningNepheleTask());
+
+ break;
+ case HYBRIDHASH_BUILD_FIRST:
+ this.matchIterator = new ReusingBuildFirstHashMatchIterator<IT1, IT2, OT>(in1, in2, serializer1, comparator1, serializer2, comparator2, pairComparatorFactory.createComparator21(comparator1, comparator2), memoryManager, ioManager, this.taskContext.getOwningNepheleTask(), fractionAvailableMemory);
+ break;
+ case HYBRIDHASH_BUILD_SECOND:
+ this.matchIterator = new ReusingBuildSecondHashMatchIterator<IT1, IT2, OT>(in1, in2, serializer1, comparator1, serializer2, comparator2, pairComparatorFactory.createComparator12(comparator1, comparator2), memoryManager, ioManager, this.taskContext.getOwningNepheleTask(), fractionAvailableMemory);
+ break;
+ default:
+ throw new Exception("Unsupported driver strategy for Match driver: " + ls.name());
+ }
+ } else {
+ switch (ls) {
+ case MERGE:
+ this.matchIterator = new MergeMatchIterator<IT1, IT2, OT>(in1, in2, serializer1, comparator1, serializer2, comparator2, pairComparatorFactory.createComparator12(comparator1, comparator2), memoryManager, ioManager, numPages, this.taskContext.getOwningNepheleTask());
+
+ break;
+ case HYBRIDHASH_BUILD_FIRST:
+ this.matchIterator = new NonReusingBuildFirstHashMatchIterator<IT1, IT2, OT>(in1, in2, serializer1, comparator1, serializer2, comparator2, pairComparatorFactory.createComparator21(comparator1, comparator2), memoryManager, ioManager, this.taskContext.getOwningNepheleTask(), fractionAvailableMemory);
+ break;
+ case HYBRIDHASH_BUILD_SECOND:
+ this.matchIterator = new NonReusingBuildSecondHashMatchIterator<IT1, IT2, OT>(in1, in2, serializer1, comparator1, serializer2, comparator2, pairComparatorFactory.createComparator12(comparator1, comparator2), memoryManager, ioManager, this.taskContext.getOwningNepheleTask(), fractionAvailableMemory);
+ break;
+ default:
+ throw new Exception("Unsupported driver strategy for Match driver: " + ls.name());
+ }
}
// open MatchTaskIterator - this triggers the sorting or hash-table building
http://git-wip-us.apache.org/repos/asf/flink/blob/b7b32a05/flink-runtime/src/main/java/org/apache/flink/runtime/operators/NoOpDriver.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/NoOpDriver.java b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/NoOpDriver.java
index 3eeb2d5..1fb4813 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/NoOpDriver.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/NoOpDriver.java
@@ -19,9 +19,13 @@
package org.apache.flink.runtime.operators;
+import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.functions.AbstractRichFunction;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.util.Collector;
import org.apache.flink.util.MutableObjectIterator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* A driver that does nothing but forward data from its input to its output.
@@ -29,12 +33,15 @@ import org.apache.flink.util.MutableObjectIterator;
* @param <T> The data type.
*/
public class NoOpDriver<T> implements PactDriver<AbstractRichFunction, T> {
-
+
+ private static final Logger LOG = LoggerFactory.getLogger(MapPartitionDriver.class);
+
private PactTaskContext<AbstractRichFunction, T> taskContext;
private volatile boolean running;
-
-
+
+ private boolean objectReuseEnabled = false;
+
@Override
public void setup(PactTaskContext<AbstractRichFunction, T> context) {
this.taskContext = context;
@@ -57,17 +64,34 @@ public class NoOpDriver<T> implements PactDriver<AbstractRichFunction, T> {
}
@Override
- public void prepare() {}
+ public void prepare() {
+ ExecutionConfig executionConfig = taskContext.getExecutionConfig();
+ this.objectReuseEnabled = executionConfig.isObjectReuseEnabled();
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("NoOpDriver object reuse: " + (this.objectReuseEnabled ? "ENABLED" : "DISABLED") + ".");
+ }
+ }
@Override
public void run() throws Exception {
// cache references on the stack
final MutableObjectIterator<T> input = this.taskContext.getInput(0);
final Collector<T> output = this.taskContext.getOutputCollector();
- T record = this.taskContext.<T>getInputSerializer(0).getSerializer().createInstance();
- while (this.running && ((record = input.next(record)) != null)) {
- output.collect(record);
+ if (objectReuseEnabled) {
+ T record = this.taskContext.<T>getInputSerializer(0).getSerializer().createInstance();
+
+ while (this.running && ((record = input.next(record)) != null)) {
+ output.collect(record);
+ }
+ } else {
+ T record;
+ TypeSerializer<T> serializer = this.taskContext.<T>getInputSerializer(0).getSerializer();
+ while (this.running && ((record = input.next(serializer.createInstance())) != null)) {
+ output.collect(record);
+ }
+
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/b7b32a05/flink-runtime/src/main/java/org/apache/flink/runtime/operators/PactTaskContext.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/PactTaskContext.java b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/PactTaskContext.java
index f6d46fa..f6d2f4a 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/PactTaskContext.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/PactTaskContext.java
@@ -19,6 +19,7 @@
package org.apache.flink.runtime.operators;
+import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.typeutils.TypeComparator;
import org.apache.flink.api.common.typeutils.TypeSerializerFactory;
import org.apache.flink.runtime.io.disk.iomanager.IOManager;
@@ -43,7 +44,9 @@ import org.apache.flink.util.MutableObjectIterator;
public interface PactTaskContext<S, OT> {
TaskConfig getTaskConfig();
-
+
+ ExecutionConfig getExecutionConfig();
+
ClassLoader getUserCodeClassLoader();
MemoryManager getMemoryManager();
http://git-wip-us.apache.org/repos/asf/flink/blob/b7b32a05/flink-runtime/src/main/java/org/apache/flink/runtime/operators/ReduceCombineDriver.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/ReduceCombineDriver.java b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/ReduceCombineDriver.java
index 35fb739..f3ea700 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/ReduceCombineDriver.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/ReduceCombineDriver.java
@@ -22,6 +22,7 @@ package org.apache.flink.runtime.operators;
import java.io.IOException;
import java.util.List;
+import org.apache.flink.api.common.ExecutionConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.flink.api.common.functions.ReduceFunction;
@@ -62,16 +63,17 @@ public class ReduceCombineDriver<T> implements PactDriver<ReduceFunction<T>, T>
private Collector<T> output;
-
private MemoryManager memManager;
private InMemorySorter<T> sorter;
private QuickSort sortAlgo = new QuickSort();
-
-
+
private boolean running;
+ private boolean objectReuseEnabled = false;
+
+
// ------------------------------------------------------------------------
@Override
@@ -124,6 +126,13 @@ public class ReduceCombineDriver<T> implements PactDriver<ReduceFunction<T>, T>
} else {
this.sorter = new NormalizedKeySorter<T>(this.serializer, this.comparator.duplicate(), memory);
}
+
+ ExecutionConfig executionConfig = taskContext.getExecutionConfig();
+ this.objectReuseEnabled = executionConfig.isObjectReuseEnabled();
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("ReduceCombineDriver object reuse: " + (this.objectReuseEnabled ? "ENABLED" : "DISABLED") + ".");
+ }
}
@Override
@@ -172,26 +181,60 @@ public class ReduceCombineDriver<T> implements PactDriver<ReduceFunction<T>, T>
final Collector<T> output = this.output;
final MutableObjectIterator<T> input = sorter.getIterator();
-
- T value = input.next(serializer.createInstance());
-
- // iterate over key groups
- while (this.running && value != null) {
- comparator.setReference(value);
- T res = value;
-
- // iterate within a key group
- while ((value = input.next(serializer.createInstance())) != null) {
- if (comparator.equalToReference(value)) {
- // same group, reduce
- res = function.reduce(res, value);
- } else {
- // new key group
- break;
+
+ if (objectReuseEnabled) {
+ // We only need two objects. The user function is expected to return
+ // the first input as the result. The output value is also expected
+ // to have the same key fields as the input elements.
+
+ T reuse1 = serializer.createInstance();
+ T reuse2 = serializer.createInstance();
+
+ T value = input.next(reuse1);
+
+ // iterate over key groups
+ while (this.running && value != null) {
+ comparator.setReference(value);
+ T res = value;
+
+ // iterate within a key group
+ while ((value = input.next(reuse2)) != null) {
+ if (comparator.equalToReference(value)) {
+ // same group, reduce
+ res = function.reduce(res, value);
+ } else {
+ // new key group
+ break;
+ }
}
+
+ output.collect(res);
+
+ if (value != null) {
+ value = serializer.copy(value, reuse1);
+ }
+ }
+ } else {
+ T value = input.next(serializer.createInstance());
+
+ // iterate over key groups
+ while (this.running && value != null) {
+ comparator.setReference(value);
+ T res = value;
+
+ // iterate within a key group
+ while ((value = input.next(serializer.createInstance())) != null) {
+ if (comparator.equalToReference(value)) {
+ // same group, reduce
+ res = function.reduce(res, value);
+ } else {
+ // new key group
+ break;
+ }
+ }
+
+ output.collect(res);
}
-
- output.collect(res);
}
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/b7b32a05/flink-runtime/src/main/java/org/apache/flink/runtime/operators/ReduceDriver.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/ReduceDriver.java b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/ReduceDriver.java
index 9cdcdda..6528f30 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/ReduceDriver.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/ReduceDriver.java
@@ -19,6 +19,7 @@
package org.apache.flink.runtime.operators;
+import org.apache.flink.api.common.ExecutionConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.flink.api.common.functions.ReduceFunction;
@@ -52,6 +53,8 @@ public class ReduceDriver<T> implements PactDriver<ReduceFunction<T>, T> {
private volatile boolean running;
+ private boolean objectReuseEnabled = false;
+
// ------------------------------------------------------------------------
@Override
@@ -88,6 +91,13 @@ public class ReduceDriver<T> implements PactDriver<ReduceFunction<T>, T> {
this.serializer = this.taskContext.<T>getInputSerializer(0).getSerializer();
this.comparator = this.taskContext.getDriverComparator(0);
this.input = this.taskContext.getInput(0);
+
+ ExecutionConfig executionConfig = taskContext.getExecutionConfig();
+ this.objectReuseEnabled = executionConfig.isObjectReuseEnabled();
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("ReduceDriver object reuse: " + (this.objectReuseEnabled ? "ENABLED" : "DISABLED") + ".");
+ }
}
@Override
@@ -104,26 +114,60 @@ public class ReduceDriver<T> implements PactDriver<ReduceFunction<T>, T> {
final ReduceFunction<T> function = this.taskContext.getStub();
final Collector<T> output = this.taskContext.getOutputCollector();
-
- T value = input.next(serializer.createInstance());
-
- // iterate over key groups
- while (this.running && value != null) {
- comparator.setReference(value);
- T res = value;
-
- // iterate within a key group
- while ((value = input.next(serializer.createInstance())) != null) {
- if (comparator.equalToReference(value)) {
- // same group, reduce
- res = function.reduce(res, value);
- } else {
- // new key group
- break;
+
+ if (objectReuseEnabled) {
+ // We only need two objects. The user function is expected to return
+ // the first input as the result. The output value is also expected
+ // to have the same key fields as the input elements.
+
+ T reuse1 = serializer.createInstance();
+ T reuse2 = serializer.createInstance();
+
+ T value = input.next(reuse1);
+
+ // iterate over key groups
+ while (this.running && value != null) {
+ comparator.setReference(value);
+ T res = value;
+
+ // iterate within a key group
+ while ((value = input.next(reuse2)) != null) {
+ if (comparator.equalToReference(value)) {
+ // same group, reduce
+ res = function.reduce(res, value);
+ } else {
+ // new key group
+ break;
+ }
+ }
+
+ output.collect(res);
+
+ if (value != null) {
+ value = serializer.copy(value, reuse1);
}
}
-
- output.collect(res);
+ } else {
+ T value = input.next(serializer.createInstance());
+
+ // iterate over key groups
+ while (this.running && value != null) {
+ comparator.setReference(value);
+ T res = value;
+
+ // iterate within a key group
+ while ((value = input.next(serializer.createInstance())) != null) {
+ if (comparator.equalToReference(value)) {
+ // same group, reduce
+ res = function.reduce(res, value);
+ } else {
+ // new key group
+ break;
+ }
+ }
+
+ output.collect(res);
+ }
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/b7b32a05/flink-runtime/src/main/java/org/apache/flink/runtime/operators/RegularPactTask.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/RegularPactTask.java b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/RegularPactTask.java
index 74bc4cb..db36f6d 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/RegularPactTask.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/RegularPactTask.java
@@ -20,6 +20,7 @@ package org.apache.flink.runtime.operators;
import akka.actor.ActorRef;
import org.apache.flink.runtime.messages.JobManagerMessages;
+import org.apache.flink.api.common.ExecutionConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.flink.api.common.accumulators.Accumulator;
@@ -1057,7 +1058,7 @@ public class RegularPactTask<S extends Function, OT> extends AbstractInvokable i
ClassLoader userCodeClassLoader = getUserCodeClassLoader();
- this.output = initOutputs(this, userCodeClassLoader, this.config, this.chainedTasks, this.eventualOutputs);
+ this.output = initOutputs(this, userCodeClassLoader, this.config, this.chainedTasks, this.eventualOutputs, this.getExecutionConfig());
}
public DistributedRuntimeUDFContext createRuntimeContext(String taskName) {
@@ -1076,6 +1077,25 @@ public class RegularPactTask<S extends Function, OT> extends AbstractInvokable i
}
@Override
+ public ExecutionConfig getExecutionConfig() {
+ try {
+ ExecutionConfig c = (ExecutionConfig) InstantiationUtil.readObjectFromConfig(
+ getOwningNepheleTask().getJobConfiguration(),
+ ExecutionConfig.CONFIG_KEY,
+ this.getClass().getClassLoader());
+ if (c != null) {
+ return c;
+ } else {
+ return new ExecutionConfig();
+ }
+ } catch (IOException e) {
+ throw new RuntimeException("Could not load ExecutionConfig from Job Configuration: " + e);
+ } catch (ClassNotFoundException e) {
+ throw new RuntimeException("Could not load ExecutionConfig from Job Configuration: " + e);
+ }
+ }
+
+ @Override
public MemoryManager getMemoryManager() {
return getEnvironment().getMemoryManager();
}
@@ -1318,7 +1338,7 @@ public class RegularPactTask<S extends Function, OT> extends AbstractInvokable i
*/
@SuppressWarnings("unchecked")
public static <T> Collector<T> initOutputs(AbstractInvokable nepheleTask, ClassLoader cl, TaskConfig config,
- List<ChainedDriver<?, ?>> chainedTasksTarget, List<BufferWriter> eventualOutputs)
+ List<ChainedDriver<?, ?>> chainedTasksTarget, List<BufferWriter> eventualOutputs, ExecutionConfig executionConfig)
throws Exception
{
final int numOutputs = config.getNumOutputs();
@@ -1355,7 +1375,7 @@ public class RegularPactTask<S extends Function, OT> extends AbstractInvokable i
previous = getOutputCollector(nepheleTask, chainedStubConf, cl, eventualOutputs, chainedStubConf.getNumOutputs());
}
- ct.setup(chainedStubConf, taskName, previous, nepheleTask, cl);
+ ct.setup(chainedStubConf, taskName, previous, nepheleTask, cl, executionConfig);
chainedTasksTarget.add(0, ct);
previous = ct;
http://git-wip-us.apache.org/repos/asf/flink/blob/b7b32a05/flink-runtime/src/main/java/org/apache/flink/runtime/operators/chaining/ChainedDriver.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/chaining/ChainedDriver.java b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/chaining/ChainedDriver.java
index 28cebf0..c36e742 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/chaining/ChainedDriver.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/chaining/ChainedDriver.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.operators.chaining;
+import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.functions.Function;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.runtime.execution.Environment;
@@ -43,9 +44,13 @@ public abstract class ChainedDriver<IT, OT> implements Collector<IT> {
private DistributedRuntimeUDFContext udfContext;
+ protected ExecutionConfig executionConfig;
+
+ protected boolean objectReuseEnabled = false;
+
public void setup(TaskConfig config, String taskName, Collector<OT> outputCollector,
- AbstractInvokable parent, ClassLoader userCodeClassLoader)
+ AbstractInvokable parent, ClassLoader userCodeClassLoader, ExecutionConfig executionConfig)
{
this.config = config;
this.taskName = taskName;
@@ -60,6 +65,9 @@ public abstract class ChainedDriver<IT, OT> implements Collector<IT> {
env.getIndexInSubtaskGroup(), userCodeClassLoader, env.getCopyTask());
}
+ this.executionConfig = executionConfig;
+ this.objectReuseEnabled = executionConfig.isObjectReuseEnabled();
+
setup(parent);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/b7b32a05/flink-runtime/src/main/java/org/apache/flink/runtime/operators/chaining/SynchronousChainedCombineDriver.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/chaining/SynchronousChainedCombineDriver.java b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/chaining/SynchronousChainedCombineDriver.java
index 5492635..65426ef 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/chaining/SynchronousChainedCombineDriver.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/chaining/SynchronousChainedCombineDriver.java
@@ -38,11 +38,17 @@ import org.apache.flink.runtime.operators.sort.FixedLengthRecordSorter;
import org.apache.flink.runtime.operators.sort.InMemorySorter;
import org.apache.flink.runtime.operators.sort.NormalizedKeySorter;
import org.apache.flink.runtime.operators.sort.QuickSort;
-import org.apache.flink.runtime.util.KeyGroupedIterator;
+import org.apache.flink.runtime.util.NonReusingKeyGroupedIterator;
+import org.apache.flink.runtime.util.ReusingKeyGroupedIterator;
import org.apache.flink.util.Collector;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
public class SynchronousChainedCombineDriver<T> extends ChainedDriver<T, T> {
+ private static final Logger LOG = LoggerFactory.getLogger(SynchronousChainedCombineDriver.class);
+
+
/**
* Fix length records with a length below this threshold will be in-place sorted, if possible.
*/
@@ -110,6 +116,10 @@ public class SynchronousChainedCombineDriver<T> extends ChainedDriver<T, T> {
} else {
this.sorter = new NormalizedKeySorter<T>(this.serializer, this.sortingComparator.duplicate(), memory);
}
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("SynchronousChainedCombineDriver object reuse: " + (this.objectReuseEnabled ? "ENABLED" : "DISABLED") + ".");
+ }
}
@Override
@@ -183,19 +193,37 @@ public class SynchronousChainedCombineDriver<T> extends ChainedDriver<T, T> {
private void sortAndCombine() throws Exception {
final InMemorySorter<T> sorter = this.sorter;
- if (!sorter.isEmpty()) {
- this.sortAlgo.sort(sorter);
- // run the combiner
- final KeyGroupedIterator<T> keyIter = new KeyGroupedIterator<T>(sorter.getIterator(), this.serializer,
- this.groupingComparator);
+ if (objectReuseEnabled) {
+ if (!sorter.isEmpty()) {
+ this.sortAlgo.sort(sorter);
+ // run the combiner
+ final ReusingKeyGroupedIterator<T> keyIter = new ReusingKeyGroupedIterator<T>(sorter.getIterator(), this.serializer, this.groupingComparator);
+
+
+ // cache references on the stack
+ final FlatCombineFunction<T> stub = this.combiner;
+ final Collector<T> output = this.outputCollector;
+
+ // run stub implementation
+ while (this.running && keyIter.nextKey()) {
+ stub.combine(keyIter.getValues(), output);
+ }
+ }
+ } else {
+ if (!sorter.isEmpty()) {
+ this.sortAlgo.sort(sorter);
+ // run the combiner
+ final NonReusingKeyGroupedIterator<T> keyIter = new NonReusingKeyGroupedIterator<T>(sorter.getIterator(), this.serializer, this.groupingComparator);
+
- // cache references on the stack
- final FlatCombineFunction<T> stub = this.combiner;
- final Collector<T> output = this.outputCollector;
+ // cache references on the stack
+ final FlatCombineFunction<T> stub = this.combiner;
+ final Collector<T> output = this.outputCollector;
- // run stub implementation
- while (this.running && keyIter.nextKey()) {
- stub.combine(keyIter.getValues(), output);
+ // run stub implementation
+ while (this.running && keyIter.nextKey()) {
+ stub.combine(keyIter.getValues(), output);
+ }
}
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/b7b32a05/flink-runtime/src/main/java/org/apache/flink/runtime/operators/hash/BuildFirstHashMatchIterator.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/hash/BuildFirstHashMatchIterator.java b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/hash/BuildFirstHashMatchIterator.java
deleted file mode 100644
index a65633f..0000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/hash/BuildFirstHashMatchIterator.java
+++ /dev/null
@@ -1,168 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-
-package org.apache.flink.runtime.operators.hash;
-
-import java.io.IOException;
-import java.util.List;
-
-import org.apache.flink.api.common.functions.FlatJoinFunction;
-import org.apache.flink.api.common.typeutils.TypeComparator;
-import org.apache.flink.api.common.typeutils.TypePairComparator;
-import org.apache.flink.api.common.typeutils.TypeSerializer;
-import org.apache.flink.core.memory.MemorySegment;
-import org.apache.flink.runtime.io.disk.iomanager.IOManager;
-import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
-import org.apache.flink.runtime.memorymanager.MemoryAllocationException;
-import org.apache.flink.runtime.memorymanager.MemoryManager;
-import org.apache.flink.runtime.operators.util.JoinTaskIterator;
-import org.apache.flink.util.Collector;
-import org.apache.flink.util.MutableObjectIterator;
-
-
-/**
- * An implementation of the {@link org.apache.flink.runtime.operators.util.JoinTaskIterator} that uses a hybrid-hash-join
- * internally to match the records with equal key. The build side of the hash is the first input of the match.
- */
-public class BuildFirstHashMatchIterator<V1, V2, O> implements JoinTaskIterator<V1, V2, O> {
-
- protected final MutableHashTable<V1, V2> hashJoin;
-
- private final V1 nextBuildSideObject;
-
- private final V1 tempBuildSideRecord;
-
- private final V2 probeCopy;
-
- protected final TypeSerializer<V2> probeSideSerializer;
-
- private final MemoryManager memManager;
-
- private final MutableObjectIterator<V1> firstInput;
-
- private final MutableObjectIterator<V2> secondInput;
-
- private volatile boolean running = true;
-
- // --------------------------------------------------------------------------------------------
-
- public BuildFirstHashMatchIterator(MutableObjectIterator<V1> firstInput, MutableObjectIterator<V2> secondInput,
- 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
- {
- this.memManager = memManager;
- this.firstInput = firstInput;
- this.secondInput = secondInput;
- this.probeSideSerializer = serializer2;
-
- this.nextBuildSideObject = serializer1.createInstance();
- this.tempBuildSideRecord = serializer1.createInstance();
- this.probeCopy = serializer2.createInstance();
-
- this.hashJoin = getHashJoin(serializer1, comparator1, serializer2, comparator2, pairComparator,
- memManager, ioManager, ownerTask, memoryFraction);
- }
-
- // --------------------------------------------------------------------------------------------
-
- @Override
- public void open() throws IOException, MemoryAllocationException, InterruptedException {
- this.hashJoin.open(this.firstInput, this.secondInput);
- }
-
-
- @Override
- public void close() {
- // close the join
- this.hashJoin.close();
-
- // free the memory
- final List<MemorySegment> segments = this.hashJoin.getFreedMemory();
- this.memManager.release(segments);
- }
-
- @Override
- public final boolean callWithNextKey(FlatJoinFunction<V1, V2, O> matchFunction, Collector<O> collector)
- throws Exception
- {
- if (this.hashJoin.nextRecord())
- {
- // we have a next record, get the iterators to the probe and build side values
- final MutableHashTable.HashBucketIterator<V1, V2> buildSideIterator = this.hashJoin.getBuildSideIterator();
- V1 nextBuildSideRecord = this.nextBuildSideObject;
-
- // get the first build side value
- if ((nextBuildSideRecord = buildSideIterator.next(nextBuildSideRecord)) != null) {
- V1 tmpRec = this.tempBuildSideRecord;
- final V2 probeRecord = this.hashJoin.getCurrentProbeRecord();
-
- // check if there is another build-side value
- if ((tmpRec = buildSideIterator.next(tmpRec)) != null) {
- // more than one build-side value --> copy the probe side
- V2 probeCopy = this.probeCopy;
- probeCopy = this.probeSideSerializer.copy(probeRecord, probeCopy);
-
- // call match on the first pair
- matchFunction.join(nextBuildSideRecord, probeCopy, collector);
-
- // call match on the second pair
- probeCopy = this.probeSideSerializer.copy(probeRecord, probeCopy);
- matchFunction.join(tmpRec, probeCopy, collector);
-
- while (this.running && ((nextBuildSideRecord = buildSideIterator.next(nextBuildSideRecord)) != null)) {
- // call match on the next pair
- // make sure we restore the value of the probe side record
- probeCopy = this.probeSideSerializer.copy(probeRecord, probeCopy);
- matchFunction.join(nextBuildSideRecord, probeCopy, collector);
- }
- }
- else {
- // only single pair matches
- matchFunction.join(nextBuildSideRecord, probeRecord, collector);
- }
- }
- return true;
- }
- else {
- return false;
- }
- }
-
- @Override
- public void abort() {
- this.running = false;
- this.hashJoin.abort();
- }
-
- // --------------------------------------------------------------------------------------------
-
- 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 MutableHashTable<BT, PT>(buildSideSerializer, probeSideSerializer, buildSideComparator, probeSideComparator, pairComparator, memorySegments, ioManager);
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/b7b32a05/flink-runtime/src/main/java/org/apache/flink/runtime/operators/hash/BuildFirstReOpenableHashMatchIterator.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/hash/BuildFirstReOpenableHashMatchIterator.java b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/hash/BuildFirstReOpenableHashMatchIterator.java
deleted file mode 100644
index 9a05264..0000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/hash/BuildFirstReOpenableHashMatchIterator.java
+++ /dev/null
@@ -1,78 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-
-package org.apache.flink.runtime.operators.hash;
-
-import java.io.IOException;
-import java.util.List;
-
-import org.apache.flink.api.common.typeutils.TypeComparator;
-import org.apache.flink.api.common.typeutils.TypePairComparator;
-import org.apache.flink.api.common.typeutils.TypeSerializer;
-import org.apache.flink.core.memory.MemorySegment;
-import org.apache.flink.runtime.io.disk.iomanager.IOManager;
-import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
-import org.apache.flink.runtime.memorymanager.MemoryAllocationException;
-import org.apache.flink.runtime.memorymanager.MemoryManager;
-import org.apache.flink.util.MutableObjectIterator;
-
-public class BuildFirstReOpenableHashMatchIterator<V1, V2, O> extends BuildFirstHashMatchIterator<V1, V2, O> {
-
-
- private final ReOpenableMutableHashTable<V1, V2> reopenHashTable;
-
- public BuildFirstReOpenableHashMatchIterator(
- MutableObjectIterator<V1> firstInput,
- MutableObjectIterator<V2> secondInput,
- 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
- {
- 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,
- 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<V2> probeInput) throws IOException {
- reopenHashTable.reopenProbe(probeInput);
- }
-
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/b7b32a05/flink-runtime/src/main/java/org/apache/flink/runtime/operators/hash/BuildSecondHashMatchIterator.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/hash/BuildSecondHashMatchIterator.java b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/hash/BuildSecondHashMatchIterator.java
deleted file mode 100644
index e1226da..0000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/hash/BuildSecondHashMatchIterator.java
+++ /dev/null
@@ -1,166 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-
-package org.apache.flink.runtime.operators.hash;
-
-import java.io.IOException;
-import java.util.List;
-
-import org.apache.flink.api.common.functions.FlatJoinFunction;
-import org.apache.flink.api.common.typeutils.TypeComparator;
-import org.apache.flink.api.common.typeutils.TypePairComparator;
-import org.apache.flink.api.common.typeutils.TypeSerializer;
-import org.apache.flink.core.memory.MemorySegment;
-import org.apache.flink.runtime.io.disk.iomanager.IOManager;
-import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
-import org.apache.flink.runtime.memorymanager.MemoryAllocationException;
-import org.apache.flink.runtime.memorymanager.MemoryManager;
-import org.apache.flink.runtime.operators.util.JoinTaskIterator;
-import org.apache.flink.util.Collector;
-import org.apache.flink.util.MutableObjectIterator;
-
-
-/**
- * An implementation of the {@link org.apache.flink.runtime.operators.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 class BuildSecondHashMatchIterator<V1, V2, O> implements JoinTaskIterator<V1, V2, O> {
-
- protected final MutableHashTable<V2, V1> hashJoin;
-
- private final V2 nextBuildSideObject;
-
- private final V2 tempBuildSideRecord;
-
- private final V1 probeCopy;
-
- protected final TypeSerializer<V1> probeSideSerializer;
-
- private final MemoryManager memManager;
-
- private final MutableObjectIterator<V1> firstInput;
-
- private final MutableObjectIterator<V2> secondInput;
-
- private volatile boolean running = true;
-
- // --------------------------------------------------------------------------------------------
-
- public BuildSecondHashMatchIterator(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
- {
- this.memManager = memManager;
- this.firstInput = firstInput;
- this.secondInput = secondInput;
- this.probeSideSerializer = serializer1;
-
- this.nextBuildSideObject = serializer2.createInstance();
- this.tempBuildSideRecord = serializer2.createInstance();
- this.probeCopy = serializer1.createInstance();
-
- this.hashJoin = getHashJoin(serializer2, comparator2, serializer1, comparator1, pairComparator,
- memManager, ioManager, ownerTask, memoryFraction);
- }
-
- // --------------------------------------------------------------------------------------------
-
- @Override
- public void open() throws IOException, MemoryAllocationException, InterruptedException {
- this.hashJoin.open(this.secondInput, this.firstInput);
- }
-
- @Override
- public void close() {
- // close the join
- this.hashJoin.close();
-
- // free the memory
- final List<MemorySegment> segments = this.hashJoin.getFreedMemory();
- this.memManager.release(segments);
- }
-
- @Override
- public boolean callWithNextKey(FlatJoinFunction<V1, V2, O> matchFunction, Collector<O> collector)
- throws Exception
- {
- if (this.hashJoin.nextRecord())
- {
- // we have a next record, get the iterators to the probe and build side values
- final MutableHashTable.HashBucketIterator<V2, V1> buildSideIterator = this.hashJoin.getBuildSideIterator();
- V2 nextBuildSideRecord = this.nextBuildSideObject;
-
- // get the first build side value
- if ((nextBuildSideRecord = buildSideIterator.next(nextBuildSideRecord)) != null) {
- V2 tmpRec = this.tempBuildSideRecord;
- final V1 probeRecord = this.hashJoin.getCurrentProbeRecord();
-
- // check if there is another build-side value
- if ((tmpRec = buildSideIterator.next(tmpRec)) != null) {
- // more than one build-side value --> copy the probe side
- V1 probeCopy = this.probeCopy;
- probeCopy = this.probeSideSerializer.copy(probeRecord, probeCopy);
-
- // call match on the first pair
- matchFunction.join(probeCopy, nextBuildSideRecord, collector);
-
- // call match on the second pair
- probeCopy = this.probeSideSerializer.copy(probeRecord, probeCopy);
- matchFunction.join(probeCopy, tmpRec, collector);
-
- while (this.running && ((nextBuildSideRecord = buildSideIterator.next(nextBuildSideRecord)) != null)) {
- // call match on the next pair
- // make sure we restore the value of the probe side record
- probeCopy = this.probeSideSerializer.copy(probeRecord, probeCopy);
- matchFunction.join(probeCopy, nextBuildSideRecord, collector);
- }
- }
- else {
- // only single pair matches
- matchFunction.join(probeRecord, nextBuildSideRecord, collector);
- }
- }
- return true;
- }
- else {
- return false;
- }
- }
-
- @Override
- public void abort() {
- this.running = false;
- this.hashJoin.abort();
- }
-
- 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 MutableHashTable<BT, PT>(buildSideSerializer, probeSideSerializer, buildSideComparator, probeSideComparator, pairComparator, memorySegments, ioManager);
- }
-
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/b7b32a05/flink-runtime/src/main/java/org/apache/flink/runtime/operators/hash/BuildSecondReOpenableHashMatchIterator.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/hash/BuildSecondReOpenableHashMatchIterator.java b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/hash/BuildSecondReOpenableHashMatchIterator.java
deleted file mode 100644
index e005851..0000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/hash/BuildSecondReOpenableHashMatchIterator.java
+++ /dev/null
@@ -1,77 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-
-package org.apache.flink.runtime.operators.hash;
-
-import java.io.IOException;
-import java.util.List;
-
-import org.apache.flink.api.common.typeutils.TypeComparator;
-import org.apache.flink.api.common.typeutils.TypePairComparator;
-import org.apache.flink.api.common.typeutils.TypeSerializer;
-import org.apache.flink.core.memory.MemorySegment;
-import org.apache.flink.runtime.io.disk.iomanager.IOManager;
-import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
-import org.apache.flink.runtime.memorymanager.MemoryAllocationException;
-import org.apache.flink.runtime.memorymanager.MemoryManager;
-import org.apache.flink.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/flink/blob/b7b32a05/flink-runtime/src/main/java/org/apache/flink/runtime/operators/hash/CompactingHashTable.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/hash/CompactingHashTable.java b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/hash/CompactingHashTable.java
index 4c39f28..3dd400f 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/hash/CompactingHashTable.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/hash/CompactingHashTable.java
@@ -1332,7 +1332,7 @@ public class CompactingHashTable<T> extends AbstractMutableHashTable<T>{
super(probeTypeComparator, pairComparator);
}
- public T getMatchFor(PT probeSideRecord, T targetForMatch) {
+ public T getMatchFor(PT probeSideRecord, T reuse) {
if(closed.get()) {
return null;
}
@@ -1372,13 +1372,13 @@ public class CompactingHashTable<T> extends AbstractMutableHashTable<T>{
// deserialize the key to check whether it is really equal, or whether we had only a hash collision
try {
- targetForMatch = p.readRecordAt(pointer, targetForMatch);
+ reuse = p.readRecordAt(pointer, reuse);
- if (this.pairComparator.equalToReference(targetForMatch)) {
+ if (this.pairComparator.equalToReference(reuse)) {
this.partition = p;
this.bucket = bucket;
this.pointerOffsetInBucket = pointerOffset;
- return targetForMatch;
+ return reuse;
}
}
catch (IOException e) {
@@ -1404,7 +1404,80 @@ public class CompactingHashTable<T> extends AbstractMutableHashTable<T>{
numInSegment = 0;
}
}
-
+
+ public T getMatchFor(PT probeSideRecord) {
+ if(closed.get()) {
+ return null;
+ }
+ final int searchHashCode = hash(this.probeTypeComparator.hash(probeSideRecord));
+
+ final int posHashCode = searchHashCode % numBuckets;
+
+ // get the bucket for the given hash code
+ MemorySegment bucket = buckets[posHashCode >> bucketsPerSegmentBits];
+ int bucketInSegmentOffset = (posHashCode & bucketsPerSegmentMask) << NUM_INTRA_BUCKET_BITS;
+
+ // get the basic characteristics of the bucket
+ final int partitionNumber = bucket.get(bucketInSegmentOffset + HEADER_PARTITION_OFFSET);
+ final InMemoryPartition<T> p = partitions.get(partitionNumber);
+ final MemorySegment[] overflowSegments = p.overflowSegments;
+
+ this.pairComparator.setReference(probeSideRecord);
+
+ int countInSegment = bucket.getInt(bucketInSegmentOffset + HEADER_COUNT_OFFSET);
+ int numInSegment = 0;
+ int posInSegment = bucketInSegmentOffset + BUCKET_HEADER_LENGTH;
+
+ // loop over all segments that are involved in the bucket (original bucket plus overflow buckets)
+ while (true) {
+
+ while (numInSegment < countInSegment) {
+
+ final int thisCode = bucket.getInt(posInSegment);
+ posInSegment += HASH_CODE_LEN;
+
+ // check if the hash code matches
+ if (thisCode == searchHashCode) {
+ // get the pointer to the pair
+ final int pointerOffset = bucketInSegmentOffset + BUCKET_POINTER_START_OFFSET + (numInSegment * POINTER_LEN);
+ final long pointer = bucket.getLong(pointerOffset);
+ numInSegment++;
+
+ // deserialize the key to check whether it is really equal, or whether we had only a hash collision
+ try {
+ T result = p.readRecordAt(pointer);
+
+ if (this.pairComparator.equalToReference(result)) {
+ this.partition = p;
+ this.bucket = bucket;
+ this.pointerOffsetInBucket = pointerOffset;
+ return result;
+ }
+ }
+ catch (IOException e) {
+ throw new RuntimeException("Error deserializing record from the hashtable: " + e.getMessage(), e);
+ }
+ }
+ else {
+ numInSegment++;
+ }
+ }
+
+ // this segment is done. check if there is another chained bucket
+ final long forwardPointer = bucket.getLong(bucketInSegmentOffset + HEADER_FORWARD_OFFSET);
+ if (forwardPointer == BUCKET_FORWARD_POINTER_NOT_SET) {
+ return null;
+ }
+
+ final int overflowSegNum = (int) (forwardPointer >>> 32);
+ bucket = overflowSegments[overflowSegNum];
+ bucketInSegmentOffset = (int) (forwardPointer & 0xffffffff);
+ countInSegment = bucket.getInt(bucketInSegmentOffset + HEADER_COUNT_OFFSET);
+ posInSegment = bucketInSegmentOffset + BUCKET_HEADER_LENGTH;
+ numInSegment = 0;
+ }
+ }
+
public void updateMatch(T record) throws IOException {
if(closed.get()) {
return;
http://git-wip-us.apache.org/repos/asf/flink/blob/b7b32a05/flink-runtime/src/main/java/org/apache/flink/runtime/operators/hash/HashMatchIteratorBase.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/hash/HashMatchIteratorBase.java b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/hash/HashMatchIteratorBase.java
new file mode 100644
index 0000000..4e0112a
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/hash/HashMatchIteratorBase.java
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.runtime.operators.hash;
+
+import org.apache.flink.api.common.typeutils.TypeComparator;
+import org.apache.flink.api.common.typeutils.TypePairComparator;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
+import org.apache.flink.runtime.memorymanager.MemoryAllocationException;
+import org.apache.flink.runtime.memorymanager.MemoryManager;
+
+import java.util.List;
+
+/**
+ * Common methods for all Hash Join Iterators.
+ */
+public class HashMatchIteratorBase {
+ 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 MutableHashTable<BT, PT>(buildSideSerializer, probeSideSerializer, buildSideComparator, probeSideComparator, pairComparator, memorySegments, ioManager);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/b7b32a05/flink-runtime/src/main/java/org/apache/flink/runtime/operators/hash/InMemoryPartition.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/hash/InMemoryPartition.java b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/hash/InMemoryPartition.java
index ca3eb4e..7fb997e 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/hash/InMemoryPartition.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/hash/InMemoryPartition.java
@@ -235,6 +235,11 @@ public class InMemoryPartition<T> {
this.readView.setReadPosition(pointer);
return this.serializer.deserialize(reuse, this.readView);
}
+
+ public T readRecordAt(long pointer) throws IOException {
+ this.readView.setReadPosition(pointer);
+ return this.serializer.deserialize(this.readView);
+ }
/**
* UNSAFE!! overwrites record
http://git-wip-us.apache.org/repos/asf/flink/blob/b7b32a05/flink-runtime/src/main/java/org/apache/flink/runtime/operators/hash/NonReusingBuildFirstHashMatchIterator.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/hash/NonReusingBuildFirstHashMatchIterator.java b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/hash/NonReusingBuildFirstHashMatchIterator.java
new file mode 100644
index 0000000..c2d7805
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/hash/NonReusingBuildFirstHashMatchIterator.java
@@ -0,0 +1,152 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+
+package org.apache.flink.runtime.operators.hash;
+
+import org.apache.flink.api.common.functions.FlatJoinFunction;
+import org.apache.flink.api.common.typeutils.TypeComparator;
+import org.apache.flink.api.common.typeutils.TypePairComparator;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
+import org.apache.flink.runtime.memorymanager.MemoryAllocationException;
+import org.apache.flink.runtime.memorymanager.MemoryManager;
+import org.apache.flink.runtime.operators.util.JoinTaskIterator;
+import org.apache.flink.util.Collector;
+import org.apache.flink.util.MutableObjectIterator;
+
+import java.io.IOException;
+import java.util.List;
+
+
+/**
+ * An implementation of the {@link org.apache.flink.runtime.operators.util.JoinTaskIterator} that uses a hybrid-hash-join
+ * internally to match the records with equal key. The build side of the hash is the first input of the match.
+ * This implementation DOES NOT reuse objects.
+ */
+public class NonReusingBuildFirstHashMatchIterator<V1, V2, O> extends HashMatchIteratorBase implements JoinTaskIterator<V1, V2, O> {
+
+ protected final MutableHashTable<V1, V2> hashJoin;
+
+ protected final TypeSerializer<V2> probeSideSerializer;
+
+ private final MemoryManager memManager;
+
+ private final MutableObjectIterator<V1> firstInput;
+
+ private final MutableObjectIterator<V2> secondInput;
+
+ private volatile boolean running = true;
+
+ // --------------------------------------------------------------------------------------------
+
+ public NonReusingBuildFirstHashMatchIterator(
+ MutableObjectIterator<V1> firstInput,
+ MutableObjectIterator<V2> secondInput,
+ 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
+ {
+ this.memManager = memManager;
+ this.firstInput = firstInput;
+ this.secondInput = secondInput;
+ this.probeSideSerializer = serializer2;
+
+ this.hashJoin = getHashJoin(serializer1, comparator1, serializer2, comparator2,
+ pairComparator, memManager, ioManager, ownerTask, memoryFraction);
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ @Override
+ public void open() throws IOException, MemoryAllocationException, InterruptedException {
+ this.hashJoin.open(this.firstInput, this.secondInput);
+ }
+
+
+ @Override
+ public void close() {
+ // close the join
+ this.hashJoin.close();
+
+ // free the memory
+ final List<MemorySegment> segments = this.hashJoin.getFreedMemory();
+ this.memManager.release(segments);
+ }
+
+ @Override
+ public final boolean callWithNextKey(FlatJoinFunction<V1, V2, O> matchFunction, Collector<O> collector)
+ throws Exception
+ {
+ if (this.hashJoin.nextRecord())
+ {
+ // we have a next record, get the iterators to the probe and build side values
+ final MutableHashTable.HashBucketIterator<V1, V2> buildSideIterator = this.hashJoin.getBuildSideIterator();
+ V1 nextBuildSideRecord;
+
+ // get the first build side value
+ if ((nextBuildSideRecord = buildSideIterator.next()) != null) {
+ V1 tmpRec;
+ final V2 probeRecord = this.hashJoin.getCurrentProbeRecord();
+
+ // check if there is another build-side value
+ if ((tmpRec = buildSideIterator.next()) != null) {
+ // more than one build-side value --> copy the probe side
+ V2 probeCopy;
+ probeCopy = this.probeSideSerializer.copy(probeRecord);
+
+ // call match on the first pair
+ matchFunction.join(nextBuildSideRecord, probeCopy, collector);
+
+ // call match on the second pair
+ probeCopy = this.probeSideSerializer.copy(probeRecord);
+ matchFunction.join(tmpRec, probeCopy, collector);
+
+ while (this.running && ((nextBuildSideRecord = buildSideIterator.next()) != null)) {
+ // call match on the next pair
+ // make sure we restore the value of the probe side record
+ probeCopy = this.probeSideSerializer.copy(probeRecord);
+ matchFunction.join(nextBuildSideRecord, probeCopy, collector);
+ }
+ }
+ else {
+ // only single pair matches
+ matchFunction.join(nextBuildSideRecord, probeRecord, collector);
+ }
+ }
+ return true;
+ }
+ else {
+ return false;
+ }
+ }
+
+ @Override
+ public void abort() {
+ this.running = false;
+ this.hashJoin.abort();
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/b7b32a05/flink-runtime/src/main/java/org/apache/flink/runtime/operators/hash/NonReusingBuildFirstReOpenableHashMatchIterator.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/hash/NonReusingBuildFirstReOpenableHashMatchIterator.java b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/hash/NonReusingBuildFirstReOpenableHashMatchIterator.java
new file mode 100644
index 0000000..ee870a6
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/hash/NonReusingBuildFirstReOpenableHashMatchIterator.java
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+
+package org.apache.flink.runtime.operators.hash;
+
+import org.apache.flink.api.common.typeutils.TypeComparator;
+import org.apache.flink.api.common.typeutils.TypePairComparator;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
+import org.apache.flink.runtime.memorymanager.MemoryAllocationException;
+import org.apache.flink.runtime.memorymanager.MemoryManager;
+import org.apache.flink.util.MutableObjectIterator;
+
+import java.io.IOException;
+import java.util.List;
+
+public class NonReusingBuildFirstReOpenableHashMatchIterator<V1, V2, O> extends NonReusingBuildFirstHashMatchIterator<V1, V2, O> {
+
+
+ private final ReOpenableMutableHashTable<V1, V2> reopenHashTable;
+
+ public NonReusingBuildFirstReOpenableHashMatchIterator(
+ MutableObjectIterator<V1> firstInput,
+ MutableObjectIterator<V2> secondInput,
+ 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
+ {
+ 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,
+ 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 java.io.IOException
+ */
+ public void reopenProbe(MutableObjectIterator<V2> probeInput) throws IOException {
+ reopenHashTable.reopenProbe(probeInput);
+ }
+
+}