[FLINK-1916] [FLINK-2361] [runtime] Fix EOFException and entry loss i…

…n CompactingHashTable Also a lot of code cleanups in CompactingHashTable
apache · Aug 5, 2015 · 925ac1f · 925ac1f
1 parent 5226f0b
commit 925ac1f
Show file tree

Hide file tree

Showing 12 changed files with 564 additions and 409 deletions.
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/GenericPairComparator.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/GenericPairComparator.java
@@ -20,9 +20,6 @@
 
 import java.io.Serializable;
 
-import org.apache.flink.api.common.typeutils.TypeComparator;
-import org.apache.flink.api.common.typeutils.TypePairComparator;
-
 
 public class GenericPairComparator<T1, T2> extends TypePairComparator<T1, T2>
 		implements Serializable {

diff --git a/...main/java/org/apache/flink/runtime/iterative/io/SolutionSetFastUpdateOutputCollector.java b/...main/java/org/apache/flink/runtime/iterative/io/SolutionSetFastUpdateOutputCollector.java
@@ -20,7 +20,6 @@
 
 import java.io.IOException;
 
-import org.apache.flink.api.common.typeutils.TypeSerializer;
 import org.apache.flink.runtime.operators.hash.CompactingHashTable;
 import org.apache.flink.util.Collector;
 
@@ -38,23 +37,20 @@ public class SolutionSetFastUpdateOutputCollector<T> implements Collector<T> {
 	private final Collector<T> delegate;
 
 	private final CompactingHashTable<T> solutionSet;
-
-	private final T tmpHolder;
 
-	public SolutionSetFastUpdateOutputCollector(CompactingHashTable<T> solutionSet, TypeSerializer<T> serializer) {
-		this(solutionSet, serializer, null);
+	public SolutionSetFastUpdateOutputCollector(CompactingHashTable<T> solutionSet) {
+		this(solutionSet, null);
 	}
 
-	public SolutionSetFastUpdateOutputCollector(CompactingHashTable<T> solutionSet, TypeSerializer<T> serializer, Collector<T> delegate) {
+	public SolutionSetFastUpdateOutputCollector(CompactingHashTable<T> solutionSet, Collector<T> delegate) {
 		this.solutionSet = solutionSet;
 		this.delegate = delegate;
-		this.tmpHolder = serializer.createInstance();
 	}
 
 	@Override
 	public void collect(T record) {
 		try {
-			solutionSet.insertOrReplaceRecord(record, tmpHolder);
+			solutionSet.insertOrReplaceRecord(record);
 			if (delegate != null) {
 				delegate.collect(record);
 			}

diff --git a/...src/main/java/org/apache/flink/runtime/iterative/io/SolutionSetUpdateOutputCollector.java b/...src/main/java/org/apache/flink/runtime/iterative/io/SolutionSetUpdateOutputCollector.java
@@ -16,12 +16,10 @@
  * limitations under the License.
  */
 
-
 package org.apache.flink.runtime.iterative.io;
 
 import java.io.IOException;
 
-import org.apache.flink.api.common.typeutils.TypeSerializer;
 import org.apache.flink.runtime.operators.hash.CompactingHashTable;
 import org.apache.flink.util.Collector;
 
@@ -40,23 +38,20 @@ public class SolutionSetUpdateOutputCollector<T> implements Collector<T> {
 	private final Collector<T> delegate;
 
 	private final CompactingHashTable<T> solutionSet;
-
-	private final T tmpHolder;
 
-	public SolutionSetUpdateOutputCollector(CompactingHashTable<T> solutionSet, TypeSerializer<T> serializer) {
-		this(solutionSet, serializer, null);
+	public SolutionSetUpdateOutputCollector(CompactingHashTable<T> solutionSet) {
+		this(solutionSet, null);
 	}
 
-	public SolutionSetUpdateOutputCollector(CompactingHashTable<T> solutionSet, TypeSerializer<T> serializer, Collector<T> delegate) {
+	public SolutionSetUpdateOutputCollector(CompactingHashTable<T> solutionSet, Collector<T> delegate) {
 		this.solutionSet = solutionSet;
 		this.delegate = delegate;
-		this.tmpHolder = serializer.createInstance();
 	}
 
 	@Override
 	public void collect(T record) {
 		try {
-			solutionSet.insertOrReplaceRecord(record, tmpHolder);
+			solutionSet.insertOrReplaceRecord(record);
 			if (delegate != null) {
 				delegate.collect(record);
 			}

diff --git a/...time/src/main/java/org/apache/flink/runtime/iterative/task/AbstractIterativePactTask.java b/...time/src/main/java/org/apache/flink/runtime/iterative/task/AbstractIterativePactTask.java
@@ -329,8 +329,7 @@ protected Collector<OT> createSolutionSetUpdateOutputCollector(Collector<OT> del
 		if (ss instanceof CompactingHashTable) {
 			@SuppressWarnings("unchecked")
 			CompactingHashTable<OT> solutionSet = (CompactingHashTable<OT>) ss;
-			TypeSerializer<OT> serializer = getOutputSerializer();
-			return new SolutionSetUpdateOutputCollector<OT>(solutionSet, serializer, delegate);
+			return new SolutionSetUpdateOutputCollector<OT>(solutionSet, delegate);
 		}
 		else if (ss instanceof JoinHashMap) {
 			@SuppressWarnings("unchecked")

diff --git a/...ntime/src/main/java/org/apache/flink/runtime/operators/hash/AbstractMutableHashTable.java b/...ntime/src/main/java/org/apache/flink/runtime/operators/hash/AbstractMutableHashTable.java
@@ -61,15 +61,13 @@ public TypeComparator<T> getBuildSideComparator() {
 
 	public abstract void abort();
 
-	public abstract void buildTable(final MutableObjectIterator<T> input) throws IOException;
-
 	public abstract List<MemorySegment> getFreeMemory();
 
 	// ------------- Modifier -------------
 
 	public abstract void insert(T record) throws IOException;
 
-	public abstract void insertOrReplaceRecord(T record, T tempHolder) throws IOException;
+	public abstract void insertOrReplaceRecord(T record) throws IOException;
 
 	// ------------- Accessors -------------