You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by gr...@apache.org on 2016/07/14 01:11:58 UTC
[2/4] flink git commit: [FLINK-3477] [runtime] Add hash-based combine
strategy for ReduceFunction
http://git-wip-us.apache.org/repos/asf/flink/blob/52e191a5/flink-runtime/src/main/java/org/apache/flink/runtime/operators/hash/MutableHashTable.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/hash/MutableHashTable.java b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/hash/MutableHashTable.java
index 351cd3f..bcaa225 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/hash/MutableHashTable.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/hash/MutableHashTable.java
@@ -1480,28 +1480,17 @@ public class MutableHashTable<BT, PT> implements MemorySegmentSource {
public static byte assignPartition(int bucket, byte numPartitions) {
return (byte) (bucket % numPartitions);
}
-
+
/**
- * This function hashes an integer value. It is adapted from Bob Jenkins' website
- * <a href="http://www.burtleburtle.net/bob/hash/integer.html">http://www.burtleburtle.net/bob/hash/integer.html</a>.
- * The hash function has the <i>full avalanche</i> property, meaning that every bit of the value to be hashed
- * affects every bit of the hash value.
- *
- * @param code The integer to be hashed.
- * @return The hash code for the integer.
- */
+ * The level parameter is needed so that we can have different hash functions when we recursively apply
+ * the partitioning, so that the working set eventually fits into memory.
+ */
public static int hash(int code, int level) {
final int rotation = level * 11;
-
- code = (code << rotation) | (code >>> -rotation);
- code = (code + 0x7ed55d16) + (code << 12);
- code = (code ^ 0xc761c23c) ^ (code >>> 19);
- code = (code + 0x165667b1) + (code << 5);
- code = (code + 0xd3a2646c) ^ (code << 9);
- code = (code + 0xfd7046c5) + (code << 3);
- code = (code ^ 0xb55a4f09) ^ (code >>> 16);
- return code >= 0 ? code : -(code + 1);
+ code = Integer.rotateLeft(code, rotation);
+
+ return MathUtils.jenkinsHash(code);
}
public TypeComparator<PT> getProbeSideComparator () {
http://git-wip-us.apache.org/repos/asf/flink/blob/52e191a5/flink-runtime/src/test/java/org/apache/flink/runtime/operators/hash/CompactingHashTableTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/hash/CompactingHashTableTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/hash/CompactingHashTableTest.java
index abcbf73..63cc59f 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/hash/CompactingHashTableTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/hash/CompactingHashTableTest.java
@@ -20,29 +20,28 @@ package org.apache.flink.runtime.operators.hash;
import static org.junit.Assert.*;
+import org.apache.flink.api.common.typeutils.SameTypePairComparator;
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.api.common.typeutils.base.LongComparator;
-import org.apache.flink.api.common.typeutils.base.LongSerializer;
-import org.apache.flink.api.common.typeutils.base.StringSerializer;
import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.flink.api.java.typeutils.runtime.TupleComparator;
-import org.apache.flink.api.java.typeutils.runtime.TupleSerializer;
import org.apache.flink.core.memory.MemorySegment;
import org.apache.flink.core.memory.MemorySegmentFactory;
+import org.apache.flink.runtime.operators.testutils.types.IntList;
+import org.apache.flink.runtime.operators.testutils.types.IntPair;
import org.apache.flink.util.MutableObjectIterator;
import org.junit.Test;
+import org.powermock.reflect.Whitebox;
+import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.BitSet;
import java.util.List;
+import java.util.Random;
-public class CompactingHashTableTest {
-
- private final TypeSerializer<Tuple2<Long, String>> serializer;
- private final TypeComparator<Tuple2<Long, String>> comparator;
+public class CompactingHashTableTest extends MutableHashTableTestBase {
private final TypeComparator<Long> probeComparator;
@@ -50,16 +49,7 @@ public class CompactingHashTableTest {
public CompactingHashTableTest() {
- TypeSerializer<?>[] fieldSerializers = { LongSerializer.INSTANCE, StringSerializer.INSTANCE };
- @SuppressWarnings("unchecked")
- Class<Tuple2<Long, String>> clazz = (Class<Tuple2<Long, String>>) (Class<?>) Tuple2.class;
- this.serializer = new TupleSerializer<Tuple2<Long, String>>(clazz, fieldSerializers);
-
- TypeComparator<?>[] comparators = { new LongComparator(true) };
- TypeSerializer<?>[] comparatorSerializers = { LongSerializer.INSTANCE };
-
- this.comparator = new TupleComparator<Tuple2<Long, String>>(new int[] {0}, comparators, comparatorSerializers);
-
+
this.probeComparator = new LongComparator(true);
this.pairComparator = new TypePairComparator<Long, Tuple2<Long, String>>() {
@@ -85,11 +75,20 @@ public class CompactingHashTableTest {
}
};
}
-
+
+ @Override
+ protected <T> AbstractMutableHashTable<T> getHashTable(TypeSerializer<T> serializer, TypeComparator<T> comparator, List<MemorySegment> memory) {
+ return new CompactingHashTable<T>(serializer, comparator, memory);
+ }
+
// ------------------------------------------------------------------------
// tests
// ------------------------------------------------------------------------
-
+
+ /**
+ * This has to be duplicated in InPlaceMutableHashTableTest and CompactingHashTableTest
+ * because of the different constructor calls.
+ */
@Test
public void testHashTableGrowthWithInsert() {
try {
@@ -101,7 +100,7 @@ public class CompactingHashTableTest {
// a lot of memory for the partition buffers, and start with a smaller hash table. that way
// we trigger a hash table growth early.
CompactingHashTable<Tuple2<Long, String>> table = new CompactingHashTable<Tuple2<Long, String>>(
- serializer, comparator, memory, 10000);
+ tuple2LongStringSerializer, tuple2LongStringComparator, memory, 10000);
table.open();
for (long i = 0; i < numElements; i++) {
@@ -143,6 +142,9 @@ public class CompactingHashTableTest {
/**
* This test validates that records are not lost via "insertOrReplace()" as in bug [FLINK-2361]
+ *
+ * This has to be duplicated in InPlaceMutableHashTableTest and CompactingHashTableTest
+ * because of the different constructor calls.
*/
@Test
public void testHashTableGrowthWithInsertOrReplace() {
@@ -154,8 +156,8 @@ public class CompactingHashTableTest {
// we create a hash table that thinks the records are super large. that makes it choose initially
// a lot of memory for the partition buffers, and start with a smaller hash table. that way
// we trigger a hash table growth early.
- CompactingHashTable<Tuple2<Long, String>> table = new CompactingHashTable<Tuple2<Long, String>>(
- serializer, comparator, memory, 10000);
+ CompactingHashTable<Tuple2<Long, String>> table = new CompactingHashTable<>(
+ tuple2LongStringSerializer, tuple2LongStringComparator, memory, 10000);
table.open();
for (long i = 0; i < numElements; i++) {
@@ -211,24 +213,33 @@ public class CompactingHashTableTest {
// we create a hash table that thinks the records are super large. that makes it choose initially
// a lot of memory for the partition buffers, and start with a smaller hash table. that way
// we trigger a hash table growth early.
- CompactingHashTable<Tuple2<Long, String>> table = new CompactingHashTable<Tuple2<Long, String>>(
- serializer, comparator, memory, 100);
+ CompactingHashTable<Tuple2<Long, String>> table = new CompactingHashTable<>(
+ tuple2LongStringSerializer, tuple2LongStringComparator, memory, 100);
table.open();
// first, we insert some elements
for (long i = 0; i < numElements; i++) {
- table.insertOrReplaceRecord(new Tuple2<Long, String>(i, longString));
+ table.insertOrReplaceRecord(Tuple2.of(i, longString));
}
// now, we replace the same elements, causing fragmentation
for (long i = 0; i < numElements; i++) {
- table.insertOrReplaceRecord(new Tuple2<Long, String>(i, longString));
+ table.insertOrReplaceRecord(Tuple2.of(i, longString));
}
// now we insert an additional set of elements. without compaction during this insertion,
// the memory will run out
for (long i = 0; i < numElements; i++) {
- table.insertOrReplaceRecord(new Tuple2<Long, String>(i + numElements, longString));
+ table.insertOrReplaceRecord(Tuple2.of(i + numElements, longString));
+ }
+
+ // check the results
+ CompactingHashTable<Tuple2<Long, String>>.HashTableProber<Tuple2<Long, String>> prober =
+ table.getProber(tuple2LongStringComparator, new SameTypePairComparator<>(tuple2LongStringComparator));
+ Tuple2<Long, String> reuse = new Tuple2<>();
+ for (long i = 0; i < numElements; i++) {
+ assertNotNull(prober.getMatchFor(Tuple2.of(i, longString), reuse));
+ assertNotNull(prober.getMatchFor(Tuple2.of(i + numElements, longString), reuse));
}
}
catch (Exception e) {
@@ -236,6 +247,233 @@ public class CompactingHashTableTest {
fail(e.getMessage());
}
}
+
+ @Test
+ public void testResize() {
+ // Only CompactingHashTable
+ try {
+ final int NUM_MEM_PAGES = 30 * NUM_PAIRS / PAGE_SIZE;
+ final Random rnd = new Random(RANDOM_SEED);
+ final IntPair[] pairs = getRandomizedIntPairs(NUM_PAIRS, rnd);
+
+ List<MemorySegment> memory = getMemory(NUM_MEM_PAGES);
+ CompactingHashTable<IntPair> table = new CompactingHashTable<IntPair>(intPairSerializer, intPairComparator, memory);
+ table.open();
+
+ for (int i = 0; i < NUM_PAIRS; i++) {
+ table.insert(pairs[i]);
+ }
+
+ AbstractHashTableProber<IntPair, IntPair> prober =
+ table.getProber(intPairComparator, new SameTypePairComparator<>(intPairComparator));
+ IntPair target = new IntPair();
+
+ for (int i = 0; i < NUM_PAIRS; i++) {
+ assertNotNull(prober.getMatchFor(pairs[i], target));
+ assertEquals(pairs[i].getValue(), target.getValue());
+ }
+
+ // make sure there is enough memory for resize
+ memory.addAll(getMemory(ADDITIONAL_MEM));
+ Boolean b = Whitebox.<Boolean>invokeMethod(table, "resizeHashTable");
+ assertTrue(b);
+
+ for (int i = 0; i < NUM_PAIRS; i++) {
+ assertNotNull(pairs[i].getKey() + " " + pairs[i].getValue(), prober.getMatchFor(pairs[i], target));
+ assertEquals(pairs[i].getValue(), target.getValue());
+ }
+
+ table.close();
+ assertEquals("Memory lost", NUM_MEM_PAGES + ADDITIONAL_MEM, table.getFreeMemory().size());
+ } catch (Exception e) {
+ e.printStackTrace();
+ fail("Error: " + e.getMessage());
+ }
+ }
+
+ @Test
+ public void testDoubleResize() {
+ // Only CompactingHashTable
+ try {
+ final int NUM_MEM_PAGES = 30 * NUM_PAIRS / PAGE_SIZE;
+ final Random rnd = new Random(RANDOM_SEED);
+ final IntPair[] pairs = getRandomizedIntPairs(NUM_PAIRS, rnd);
+
+ List<MemorySegment> memory = getMemory(NUM_MEM_PAGES);
+ CompactingHashTable<IntPair> table = new CompactingHashTable<IntPair>(intPairSerializer, intPairComparator, memory);
+ table.open();
+
+ for (int i = 0; i < NUM_PAIRS; i++) {
+ table.insert(pairs[i]);
+ }
+
+ AbstractHashTableProber<IntPair, IntPair> prober =
+ table.getProber(intPairComparator, new SameTypePairComparator<>(intPairComparator));
+ IntPair target = new IntPair();
+
+ for (int i = 0; i < NUM_PAIRS; i++) {
+ assertNotNull(prober.getMatchFor(pairs[i], target));
+ assertEquals(pairs[i].getValue(), target.getValue());
+ }
+
+ // make sure there is enough memory for resize
+ memory.addAll(getMemory(ADDITIONAL_MEM));
+ Boolean b = Whitebox.<Boolean>invokeMethod(table, "resizeHashTable");
+ assertTrue(b);
+
+ for (int i = 0; i < NUM_PAIRS; i++) {
+ assertNotNull(pairs[i].getKey() + " " + pairs[i].getValue(), prober.getMatchFor(pairs[i], target));
+ assertEquals(pairs[i].getValue(), target.getValue());
+ }
+
+ // make sure there is enough memory for resize
+ memory.addAll(getMemory(ADDITIONAL_MEM));
+ b = Whitebox.<Boolean>invokeMethod(table, "resizeHashTable");
+ assertTrue(b);
+
+ for (int i = 0; i < NUM_PAIRS; i++) {
+ assertNotNull(pairs[i].getKey() + " " + pairs[i].getValue(), prober.getMatchFor(pairs[i], target));
+ assertEquals(pairs[i].getValue(), target.getValue());
+ }
+
+ table.close();
+ assertEquals("Memory lost", NUM_MEM_PAGES + ADDITIONAL_MEM + ADDITIONAL_MEM, table.getFreeMemory().size());
+ } catch (Exception e) {
+ e.printStackTrace();
+ fail("Error: " + e.getMessage());
+ }
+ }
+
+ @Test
+ public void testTripleResize() {
+ // Only CompactingHashTable
+ try {
+ final int NUM_MEM_PAGES = 30 * NUM_PAIRS / PAGE_SIZE;
+ final Random rnd = new Random(RANDOM_SEED);
+ final IntPair[] pairs = getRandomizedIntPairs(NUM_PAIRS, rnd);
+
+ List<MemorySegment> memory = getMemory(NUM_MEM_PAGES);
+ CompactingHashTable<IntPair> table = new CompactingHashTable<IntPair>(intPairSerializer, intPairComparator, memory);
+ table.open();
+
+ for (int i = 0; i < NUM_PAIRS; i++) {
+ table.insert(pairs[i]);
+ }
+
+ AbstractHashTableProber<IntPair, IntPair> prober =
+ table.getProber(intPairComparator, new SameTypePairComparator<>(intPairComparator));
+ IntPair target = new IntPair();
+
+ for (int i = 0; i < NUM_PAIRS; i++) {
+ assertNotNull(prober.getMatchFor(pairs[i], target));
+ assertEquals(pairs[i].getValue(), target.getValue());
+ }
+
+ // make sure there is enough memory for resize
+ memory.addAll(getMemory(ADDITIONAL_MEM));
+ Boolean b = Whitebox.<Boolean>invokeMethod(table, "resizeHashTable");
+ assertTrue(b);
+
+ for (int i = 0; i < NUM_PAIRS; i++) {
+ assertNotNull(pairs[i].getKey() + " " + pairs[i].getValue(), prober.getMatchFor(pairs[i], target));
+ assertEquals(pairs[i].getValue(), target.getValue());
+ }
+
+ // make sure there is enough memory for resize
+ memory.addAll(getMemory(ADDITIONAL_MEM));
+ b = Whitebox.<Boolean>invokeMethod(table, "resizeHashTable");
+ assertTrue(b);
+
+ for (int i = 0; i < NUM_PAIRS; i++) {
+ assertNotNull(pairs[i].getKey() + " " + pairs[i].getValue(), prober.getMatchFor(pairs[i], target));
+ assertEquals(pairs[i].getValue(), target.getValue());
+ }
+
+ // make sure there is enough memory for resize
+ memory.addAll(getMemory(2*ADDITIONAL_MEM));
+ b = Whitebox.<Boolean>invokeMethod(table, "resizeHashTable");
+ assertTrue(b);
+
+ for (int i = 0; i < NUM_PAIRS; i++) {
+ assertNotNull(pairs[i].getKey() + " " + pairs[i].getValue(), prober.getMatchFor(pairs[i], target));
+ assertEquals(pairs[i].getValue(), target.getValue());
+ }
+
+ table.close();
+ assertEquals("Memory lost", NUM_MEM_PAGES + 4*ADDITIONAL_MEM, table.getFreeMemory().size());
+ } catch (Exception e) {
+ e.printStackTrace();
+ fail("Error: " + e.getMessage());
+ }
+ }
+
+ @Test
+ public void testResizeWithCompaction(){
+ // Only CompactingHashTable
+ try {
+ final int NUM_MEM_PAGES = (SIZE * NUM_LISTS / PAGE_SIZE);
+
+ final Random rnd = new Random(RANDOM_SEED);
+ final IntList[] lists = getRandomizedIntLists(NUM_LISTS, rnd);
+
+ List<MemorySegment> memory = getMemory(NUM_MEM_PAGES);
+ CompactingHashTable<IntList> table = new CompactingHashTable<IntList>(serializerV, comparatorV, memory);
+ table.open();
+
+ for (int i = 0; i < NUM_LISTS; i++) {
+ table.insert(lists[i]);
+ }
+
+ AbstractHashTableProber<IntList, IntList> prober = table.getProber(comparatorV, pairComparatorV);
+ IntList target = new IntList();
+
+ for (int i = 0; i < NUM_LISTS; i++) {
+ assertNotNull(prober.getMatchFor(lists[i], target));
+ assertArrayEquals(lists[i].getValue(), target.getValue());
+ }
+
+ // make sure there is enough memory for resize
+ memory.addAll(getMemory(ADDITIONAL_MEM));
+ Boolean b = Whitebox.<Boolean>invokeMethod(table, "resizeHashTable");
+ assertTrue(b);
+
+ for (int i = 0; i < NUM_LISTS; i++) {
+ assertNotNull(prober.getMatchFor(lists[i], target));
+ assertArrayEquals(lists[i].getValue(), target.getValue());
+ }
+
+ final IntList[] overwriteLists = getRandomizedIntLists(NUM_LISTS, rnd);
+
+ // test replacing
+ for (int i = 0; i < NUM_LISTS; i++) {
+ table.insertOrReplaceRecord(overwriteLists[i]);
+ }
+
+ Field list = Whitebox.getField(CompactingHashTable.class, "partitions");
+ @SuppressWarnings("unchecked")
+ ArrayList<InMemoryPartition<IntList>> partitions = (ArrayList<InMemoryPartition<IntList>>) list.get(table);
+ int numPartitions = partitions.size();
+ for(int i = 0; i < numPartitions; i++) {
+ Whitebox.invokeMethod(table, "compactPartition", i);
+ }
+
+ // make sure there is enough memory for resize
+ memory.addAll(getMemory(2*ADDITIONAL_MEM));
+ b = Whitebox.<Boolean>invokeMethod(table, "resizeHashTable");
+ assertTrue(b);
+
+ for (int i = 0; i < NUM_LISTS; i++) {
+ assertNotNull("" + i, prober.getMatchFor(overwriteLists[i], target));
+ assertArrayEquals(overwriteLists[i].getValue(), target.getValue());
+ }
+
+ table.close();
+ assertEquals("Memory lost", NUM_MEM_PAGES + 3*ADDITIONAL_MEM, table.getFreeMemory().size());
+ } catch (Exception e) {
+ e.printStackTrace();
+ fail("Error: " + e.getMessage());
+ }
+ }
private static List<MemorySegment> getMemory(int numSegments, int segmentSize) {
ArrayList<MemorySegment> list = new ArrayList<MemorySegment>(numSegments);
http://git-wip-us.apache.org/repos/asf/flink/blob/52e191a5/flink-runtime/src/test/java/org/apache/flink/runtime/operators/hash/HashTablePerformanceComparison.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/hash/HashTablePerformanceComparison.java b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/hash/HashTablePerformanceComparison.java
index 9cb3606..10bac1f 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/hash/HashTablePerformanceComparison.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/hash/HashTablePerformanceComparison.java
@@ -102,7 +102,7 @@ public class HashTablePerformanceComparison {
System.out.println("Starting update...");
start = System.currentTimeMillis();
while(updater.next(target) != null) {
- target.setValue(target.getValue()*-1);
+ target.setValue(target.getValue() + 1);
table.insertOrReplaceRecord(target);
}
end = System.currentTimeMillis();
@@ -112,7 +112,7 @@ public class HashTablePerformanceComparison {
start = System.currentTimeMillis();
while (updateTester.next(target) != null) {
assertNotNull(prober.getMatchFor(target, temp));
- assertEquals(target.getValue(), temp.getValue());
+ assertEquals(target.getValue() + 1, temp.getValue());
}
end = System.currentTimeMillis();
System.out.println("Probing done. Time: " + (end-start) + " ms");
@@ -175,7 +175,7 @@ public class HashTablePerformanceComparison {
System.out.println("Starting update...");
start = System.currentTimeMillis();
while(updater.next(compare) != null) {
- compare.setValue(compare.getValue()*-1);
+ compare.setValue(compare.getValue() + 1);
iter = table.getMatchesFor(compare);
iter.next(target);
iter.writeBack(compare);
@@ -187,7 +187,7 @@ public class HashTablePerformanceComparison {
System.out.println("Starting second probing run...");
start = System.currentTimeMillis();
while(updateTester.next(compare) != null) {
- compare.setValue(compare.getValue()*-1);
+ compare.setValue(compare.getValue() + 1);
iter = table.getMatchesFor(compare);
iter.next(target);
assertEquals(target.getKey(), compare.getKey());
@@ -209,6 +209,80 @@ public class HashTablePerformanceComparison {
fail("Error: " + e.getMessage());
}
}
+
+ @Test
+ public void testInPlaceMutableHashTablePerformance() {
+ try {
+ final int NUM_MEM_PAGES = SIZE * NUM_PAIRS / PAGE_SIZE;
+
+ MutableObjectIterator<IntPair> buildInput = new UniformIntPairGenerator(NUM_PAIRS, 1, false);
+
+ MutableObjectIterator<IntPair> probeTester = new UniformIntPairGenerator(NUM_PAIRS, 1, false);
+
+ MutableObjectIterator<IntPair> updater = new UniformIntPairGenerator(NUM_PAIRS, 1, false);
+
+ MutableObjectIterator<IntPair> updateTester = new UniformIntPairGenerator(NUM_PAIRS, 1, false);
+
+ long start;
+ long end;
+
+ long first = System.currentTimeMillis();
+
+ System.out.println("Creating and filling InPlaceMutableHashTable...");
+ start = System.currentTimeMillis();
+ InPlaceMutableHashTable<IntPair> table = new InPlaceMutableHashTable<>(serializer, comparator, getMemory(NUM_MEM_PAGES, PAGE_SIZE));
+ table.open();
+
+ IntPair target = new IntPair();
+ while(buildInput.next(target) != null) {
+ table.insert(target);
+ }
+ end = System.currentTimeMillis();
+ System.out.println("HashMap ready. Time: " + (end-start) + " ms");
+
+ System.out.println("Starting first probing run...");
+ start = System.currentTimeMillis();
+
+ AbstractHashTableProber<IntPair, IntPair> prober = table.getProber(comparator, pairComparator);
+ IntPair temp = new IntPair();
+ while(probeTester.next(target) != null) {
+ assertNotNull(prober.getMatchFor(target, temp));
+ assertEquals(temp.getValue(), target.getValue());
+ }
+ end = System.currentTimeMillis();
+ System.out.println("Probing done. Time: " + (end-start) + " ms");
+
+ System.out.println("Starting update...");
+ start = System.currentTimeMillis();
+ while(updater.next(target) != null) {
+ target.setValue(target.getValue() + 1);
+ table.insertOrReplaceRecord(target);
+ }
+ end = System.currentTimeMillis();
+ System.out.println("Update done. Time: " + (end-start) + " ms");
+
+ System.out.println("Starting second probing run...");
+ start = System.currentTimeMillis();
+ while (updateTester.next(target) != null) {
+ assertNotNull(prober.getMatchFor(target, temp));
+ assertEquals(target.getValue() + 1, temp.getValue());
+ }
+ end = System.currentTimeMillis();
+ System.out.println("Probing done. Time: " + (end-start) + " ms");
+
+ table.close();
+
+ end = System.currentTimeMillis();
+ System.out.println("Overall time: " + (end-first) + " ms");
+
+ assertEquals("Memory lost", NUM_MEM_PAGES, table.getFreeMemory().size());
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail("Error: " + e.getMessage());
+ }
+ }
+
private static List<MemorySegment> getMemory(int numPages, int pageSize) {
List<MemorySegment> memory = new ArrayList<MemorySegment>();
http://git-wip-us.apache.org/repos/asf/flink/blob/52e191a5/flink-runtime/src/test/java/org/apache/flink/runtime/operators/hash/InPlaceMutableHashTableTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/hash/InPlaceMutableHashTableTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/hash/InPlaceMutableHashTableTest.java
new file mode 100644
index 0000000..4db5ef8
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/hash/InPlaceMutableHashTableTest.java
@@ -0,0 +1,543 @@
+/*
+ * 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 com.google.common.collect.Ordering;
+import org.apache.flink.api.common.functions.ReduceFunction;
+import org.apache.flink.api.common.functions.util.CopyingListCollector;
+import org.apache.flink.api.common.typeutils.SameTypePairComparator;
+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.api.common.typeutils.base.LongComparator;
+import org.apache.flink.api.common.typeutils.base.LongSerializer;
+import org.apache.flink.api.common.typeutils.base.StringSerializer;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.runtime.TupleComparator;
+import org.apache.flink.api.java.typeutils.runtime.TupleSerializer;
+import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.core.memory.MemorySegmentFactory;
+import org.apache.flink.runtime.operators.testutils.UniformStringPairGenerator;
+import org.apache.flink.runtime.operators.testutils.types.*;
+import org.apache.flink.util.Collector;
+import org.apache.flink.util.MutableObjectIterator;
+import org.junit.Test;
+
+import java.io.EOFException;
+import java.util.*;
+
+import static org.junit.Assert.*;
+import static org.junit.Assert.fail;
+
+public class InPlaceMutableHashTableTest extends MutableHashTableTestBase {
+
+ private static final long RANDOM_SEED = 58723953465322L;
+
+ private static final int PAGE_SIZE = 16 * 1024;
+
+ private final TypeSerializer<Tuple2<Long, String>> serializer;
+ private final TypeComparator<Tuple2<Long, String>> comparator;
+
+ private final TypeComparator<Long> probeComparator;
+
+ private final TypePairComparator<Long, Tuple2<Long, String>> pairComparator;
+
+ @Override
+ protected <T> AbstractMutableHashTable<T> getHashTable(TypeSerializer<T> serializer, TypeComparator<T> comparator, List<MemorySegment> memory) {
+ return new InPlaceMutableHashTable<>(serializer, comparator, memory);
+ }
+
+ // ------------------ Note: This part was mostly copied from CompactingHashTableTest ------------------
+
+ public InPlaceMutableHashTableTest() {
+ TypeSerializer<?>[] fieldSerializers = { LongSerializer.INSTANCE, StringSerializer.INSTANCE };
+ @SuppressWarnings("unchecked")
+ Class<Tuple2<Long, String>> clazz = (Class<Tuple2<Long, String>>) (Class<?>) Tuple2.class;
+ this.serializer = new TupleSerializer<Tuple2<Long, String>>(clazz, fieldSerializers);
+
+ TypeComparator<?>[] comparators = { new LongComparator(true) };
+ TypeSerializer<?>[] comparatorSerializers = { LongSerializer.INSTANCE };
+
+ this.comparator = new TupleComparator<Tuple2<Long, String>>(new int[] {0}, comparators, comparatorSerializers);
+
+ this.probeComparator = new LongComparator(true);
+
+ this.pairComparator = new TypePairComparator<Long, Tuple2<Long, String>>() {
+
+ private long ref;
+
+ @Override
+ public void setReference(Long reference) {
+ ref = reference;
+ }
+
+ @Override
+ public boolean equalToReference(Tuple2<Long, String> candidate) {
+ //noinspection UnnecessaryUnboxing
+ return candidate.f0.longValue() == ref;
+ }
+
+ @Override
+ public int compareToReference(Tuple2<Long, String> candidate) {
+ long x = ref;
+ long y = candidate.f0;
+ return (x < y) ? -1 : ((x == y) ? 0 : 1);
+ }
+ };
+ }
+
+ /**
+ * This has to be duplicated in InPlaceMutableHashTableTest and CompactingHashTableTest
+ * because of the different constructor calls.
+ */
+ @Test
+ public void testHashTableGrowthWithInsert() {
+ try {
+ final int numElements = 1000000;
+
+ List<MemorySegment> memory = getMemory(10000, 32 * 1024);
+
+ InPlaceMutableHashTable<Tuple2<Long, String>> table = new InPlaceMutableHashTable<Tuple2<Long, String>>(
+ serializer, comparator, memory);
+ table.open();
+
+ for (long i = 0; i < numElements; i++) {
+ table.insert(new Tuple2<Long, String>(i, String.valueOf(i)));
+ }
+
+ // make sure that all elements are contained via the entry iterator
+ {
+ BitSet bitSet = new BitSet(numElements);
+ MutableObjectIterator<Tuple2<Long, String>> iter = table.getEntryIterator();
+ Tuple2<Long, String> next;
+ while ((next = iter.next()) != null) {
+ assertNotNull(next.f0);
+ assertNotNull(next.f1);
+ assertEquals(next.f0.longValue(), Long.parseLong(next.f1));
+
+ bitSet.set(next.f0.intValue());
+ }
+
+ assertEquals(numElements, bitSet.cardinality());
+ }
+
+ // make sure all entries are contained via the prober
+ {
+ InPlaceMutableHashTable<Tuple2<Long, String>>.HashTableProber<Long> proper =
+ table.getProber(probeComparator, pairComparator);
+
+ Tuple2<Long, String> reuse = new Tuple2<>();
+
+ for (long i = 0; i < numElements; i++) {
+ assertNotNull(proper.getMatchFor(i, reuse));
+ assertNull(proper.getMatchFor(i + numElements, reuse));
+ }
+ }
+
+ table.close();
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ /**
+ * This test validates that records are not lost via "insertOrReplace()" as in bug [FLINK-2361]
+ *
+ * This has to be duplicated in InPlaceMutableHashTableTest and CompactingHashTableTest
+ * because of the different constructor calls.
+ */
+ @Test
+ public void testHashTableGrowthWithInsertOrReplace() {
+ try {
+ final int numElements = 1000000;
+
+ List<MemorySegment> memory = getMemory(1000, 32 * 1024);
+
+ InPlaceMutableHashTable<Tuple2<Long, String>> table = new InPlaceMutableHashTable<Tuple2<Long, String>>(
+ serializer, comparator, memory);
+ table.open();
+
+ for (long i = 0; i < numElements; i++) {
+ table.insertOrReplaceRecord(Tuple2.of(i, String.valueOf(i)));
+ }
+
+ // make sure that all elements are contained via the entry iterator
+ {
+ BitSet bitSet = new BitSet(numElements);
+ MutableObjectIterator<Tuple2<Long, String>> iter = table.getEntryIterator();
+ Tuple2<Long, String> next;
+ while ((next = iter.next()) != null) {
+ assertNotNull(next.f0);
+ assertNotNull(next.f1);
+ assertEquals(next.f0.longValue(), Long.parseLong(next.f1));
+
+ bitSet.set(next.f0.intValue());
+ }
+
+ assertEquals(numElements, bitSet.cardinality());
+ }
+
+ // make sure all entries are contained via the prober
+ {
+ InPlaceMutableHashTable<Tuple2<Long, String>>.HashTableProber<Long> proper =
+ table.getProber(probeComparator, pairComparator);
+
+ Tuple2<Long, String> reuse = new Tuple2<>();
+
+ for (long i = 0; i < numElements; i++) {
+ assertNotNull(proper.getMatchFor(i, reuse));
+ assertNull(proper.getMatchFor(i + numElements, reuse));
+ }
+ }
+
+ table.close();
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ // ------------------ The following are the InPlaceMutableHashTable-specific tests ------------------
+
+ private class InPlaceMutableHashTableWithJavaHashMap<T, K> {
+
+ TypeSerializer<T> serializer;
+
+ TypeComparator<T> comparator;
+
+ ReduceFunction<T> reducer;
+
+ Collector<T> outputCollector;
+
+ HashMap<K, T> map = new HashMap<>();
+
+ public InPlaceMutableHashTableWithJavaHashMap(TypeSerializer<T> serializer, TypeComparator<T> comparator, ReduceFunction<T> reducer, Collector<T> outputCollector) {
+ this.serializer = serializer;
+ this.comparator = comparator;
+ this.reducer = reducer;
+ this.outputCollector = outputCollector;
+ }
+
+ public void updateTableEntryWithReduce(T record, K key) throws Exception {
+ record = serializer.copy(record);
+
+ if (!map.containsKey(key)) {
+ map.put(key, record);
+ } else {
+ T x = map.get(key);
+ x = reducer.reduce(x, record);
+ map.put(key, x);
+ }
+ }
+
+ public void emitAndReset() {
+ for (T record: map.values()) {
+ outputCollector.collect(record);
+ }
+ map.clear();
+ }
+ }
+
+ @Test
+ public void testWithIntPair() throws Exception {
+ Random rnd = new Random(RANDOM_SEED);
+
+ // varying the keyRange between 1000 and 1000000 can make a 5x speed difference
+ // (because of cache misses (also in the segment arrays))
+ final int keyRange = 1000000;
+ final int valueRange = 10;
+ final int numRecords = 1000000;
+
+ final IntPairSerializer serializer = new IntPairSerializer();
+ final TypeComparator<IntPair> comparator = new IntPairComparator();
+ final ReduceFunction<IntPair> reducer = new SumReducer();
+
+ // Create the InPlaceMutableHashTableWithJavaHashMap, which will provide the correct output.
+ List<IntPair> expectedOutput = new ArrayList<>();
+ InPlaceMutableHashTableWithJavaHashMap<IntPair, Integer> reference = new InPlaceMutableHashTableWithJavaHashMap<>(
+ serializer, comparator, reducer, new CopyingListCollector<>(expectedOutput, serializer));
+
+ // Create the InPlaceMutableHashTable to test
+ final int numMemPages = keyRange * 32 / PAGE_SIZE; // memory use is proportional to the number of different keys
+ List<IntPair> actualOutput = new ArrayList<>();
+
+ InPlaceMutableHashTable<IntPair> table = new InPlaceMutableHashTable<>(
+ serializer, comparator, getMemory(numMemPages, PAGE_SIZE));
+ InPlaceMutableHashTable<IntPair>.ReduceFacade reduceFacade = table.new ReduceFacade(reducer,
+ new CopyingListCollector<>(actualOutput, serializer), true);
+ table.open();
+
+ // Generate some input
+ final List<IntPair> input = new ArrayList<>();
+ for(int i = 0; i < numRecords; i++) {
+ input.add(new IntPair(rnd.nextInt(keyRange), rnd.nextInt(valueRange)));
+ }
+
+ //System.out.println("start");
+ //long start = System.currentTimeMillis();
+
+ // Process the generated input
+ final int numIntermingledEmits = 5;
+ for (IntPair record: input) {
+ reduceFacade.updateTableEntryWithReduce(serializer.copy(record));
+ reference.updateTableEntryWithReduce(serializer.copy(record), record.getKey());
+ if(rnd.nextDouble() < 1.0 / ((double)numRecords / numIntermingledEmits)) {
+ // this will fire approx. numIntermingledEmits times
+ reference.emitAndReset();
+ reduceFacade.emitAndReset();
+ }
+ }
+ reference.emitAndReset();
+ reduceFacade.emit();
+ table.close();
+
+ //long end = System.currentTimeMillis();
+ //System.out.println("stop, time: " + (end - start));
+
+ // Check results
+
+ assertEquals(expectedOutput.size(), actualOutput.size());
+
+ Integer[] expectedValues = new Integer[expectedOutput.size()];
+ for (int i = 0; i < expectedOutput.size(); i++) {
+ expectedValues[i] = expectedOutput.get(i).getValue();
+ }
+ Integer[] actualValues = new Integer[actualOutput.size()];
+ for (int i = 0; i < actualOutput.size(); i++) {
+ actualValues[i] = actualOutput.get(i).getValue();
+ }
+
+ Arrays.sort(expectedValues, Ordering.<Integer>natural());
+ Arrays.sort(actualValues, Ordering.<Integer>natural());
+ assertArrayEquals(expectedValues, actualValues);
+ }
+
+ class SumReducer implements ReduceFunction<IntPair> {
+ @Override
+ public IntPair reduce(IntPair a, IntPair b) throws Exception {
+ if (a.getKey() != b.getKey()) {
+ throw new RuntimeException("SumReducer was called with two records that have differing keys.");
+ }
+ a.setValue(a.getValue() + b.getValue());
+ return a;
+ }
+ }
+
+
+ @Test
+ public void testWithLengthChangingReduceFunction() throws Exception {
+ Random rnd = new Random(RANDOM_SEED);
+
+ final int numKeys = 10000;
+ final int numVals = 10;
+ final int numRecords = numKeys * numVals;
+
+ StringPairSerializer serializer = new StringPairSerializer();
+ StringPairComparator comparator = new StringPairComparator();
+ ReduceFunction<StringPair> reducer = new ConcatReducer();
+
+ // Create the InPlaceMutableHashTableWithJavaHashMap, which will provide the correct output.
+ List<StringPair> expectedOutput = new ArrayList<>();
+ InPlaceMutableHashTableWithJavaHashMap<StringPair, String> reference = new InPlaceMutableHashTableWithJavaHashMap<>(
+ serializer, comparator, reducer, new CopyingListCollector<>(expectedOutput, serializer));
+
+ // Create the InPlaceMutableHashTable to test
+ final int numMemPages = numRecords * 10 / PAGE_SIZE;
+
+ List<StringPair> actualOutput = new ArrayList<>();
+
+ InPlaceMutableHashTable<StringPair> table =
+ new InPlaceMutableHashTable<>(serializer, comparator, getMemory(numMemPages, PAGE_SIZE));
+ InPlaceMutableHashTable<StringPair>.ReduceFacade reduceFacade =
+ table.new ReduceFacade(reducer, new CopyingListCollector<>(actualOutput, serializer), true);
+
+ // The loop is for checking the feature that multiple open / close are possible.
+ for(int j = 0; j < 3; j++) {
+ table.open();
+
+ // Test emit when table is empty
+ reduceFacade.emit();
+
+ // Process some manual stuff
+ reference.updateTableEntryWithReduce(serializer.copy(new StringPair("foo", "bar")), "foo");
+ reference.updateTableEntryWithReduce(serializer.copy(new StringPair("foo", "baz")), "foo");
+ reference.updateTableEntryWithReduce(serializer.copy(new StringPair("alma", "xyz")), "alma");
+ reduceFacade.updateTableEntryWithReduce(serializer.copy(new StringPair("foo", "bar")));
+ reduceFacade.updateTableEntryWithReduce(serializer.copy(new StringPair("foo", "baz")));
+ reduceFacade.updateTableEntryWithReduce(serializer.copy(new StringPair("alma", "xyz")));
+ for (int i = 0; i < 5; i++) {
+ reduceFacade.updateTableEntryWithReduce(serializer.copy(new StringPair("korte", "abc")));
+ reference.updateTableEntryWithReduce(serializer.copy(new StringPair("korte", "abc")), "korte");
+ }
+ reference.emitAndReset();
+ reduceFacade.emitAndReset();
+
+ // Generate some input
+ UniformStringPairGenerator gen = new UniformStringPairGenerator(numKeys, numVals, true);
+ List<StringPair> input = new ArrayList<>();
+ StringPair cur = new StringPair();
+ while (gen.next(cur) != null) {
+ input.add(serializer.copy(cur));
+ }
+ Collections.shuffle(input, rnd);
+
+ // Process the generated input
+ final int numIntermingledEmits = 5;
+ for (StringPair record : input) {
+ reference.updateTableEntryWithReduce(serializer.copy(record), record.getKey());
+ reduceFacade.updateTableEntryWithReduce(serializer.copy(record));
+ if (rnd.nextDouble() < 1.0 / ((double) numRecords / numIntermingledEmits)) {
+ // this will fire approx. numIntermingledEmits times
+ reference.emitAndReset();
+ reduceFacade.emitAndReset();
+ }
+ }
+ reference.emitAndReset();
+ reduceFacade.emit();
+ table.close();
+
+ // Check results
+
+ assertEquals(expectedOutput.size(), actualOutput.size());
+
+ String[] expectedValues = new String[expectedOutput.size()];
+ for (int i = 0; i < expectedOutput.size(); i++) {
+ expectedValues[i] = expectedOutput.get(i).getValue();
+ }
+ String[] actualValues = new String[actualOutput.size()];
+ for (int i = 0; i < actualOutput.size(); i++) {
+ actualValues[i] = actualOutput.get(i).getValue();
+ }
+
+ Arrays.sort(expectedValues, Ordering.<String>natural());
+ Arrays.sort(actualValues, Ordering.<String>natural());
+ assertArrayEquals(expectedValues, actualValues);
+
+ expectedOutput.clear();
+ actualOutput.clear();
+ }
+ }
+
+ // Warning: Generally, reduce wouldn't give deterministic results with non-commutative ReduceFunction,
+ // but InPlaceMutableHashTable and ReduceHashTableWithJavaHashMap calls it in the same order.
+ class ConcatReducer implements ReduceFunction<StringPair> {
+ @Override
+ public StringPair reduce(StringPair a, StringPair b) throws Exception {
+ if (a.getKey().compareTo(b.getKey()) != 0) {
+ throw new RuntimeException("ConcatReducer was called with two records that have differing keys.");
+ }
+ return new StringPair(a.getKey(), a.getValue().concat(b.getValue()));
+ }
+ }
+
+
+ private static List<MemorySegment> getMemory(int numPages, int pageSize) {
+ List<MemorySegment> memory = new ArrayList<>();
+
+ for (int i = 0; i < numPages; i++) {
+ memory.add(MemorySegmentFactory.allocateUnpooledSegment(pageSize));
+ }
+
+ return memory;
+ }
+
+
+ /**
+ * The records are larger than one segment. Additionally, there is just barely enough memory,
+ * so lots of compactions will happen.
+ */
+ @Test
+ public void testLargeRecordsWithManyCompactions() {
+ try {
+ final int numElements = 1000;
+
+ final String longString1 = getLongString(100000), longString2 = getLongString(110000);
+ List<MemorySegment> memory = getMemory(3800, 32 * 1024);
+
+ InPlaceMutableHashTable<Tuple2<Long, String>> table =
+ new InPlaceMutableHashTable<>(serializer, comparator, memory);
+ table.open();
+
+ // first, we insert some elements
+ for (long i = 0; i < numElements; i++) {
+ table.insertOrReplaceRecord(Tuple2.of(i, longString1));
+ }
+
+ // now, we replace the same elements with larger ones, causing fragmentation
+ for (long i = 0; i < numElements; i++) {
+ table.insertOrReplaceRecord(Tuple2.of(i, longString2));
+ }
+
+ // check the results
+ InPlaceMutableHashTable<Tuple2<Long, String>>.HashTableProber<Tuple2<Long, String>> prober =
+ table.getProber(comparator, new SameTypePairComparator<>(comparator));
+ Tuple2<Long, String> reuse = new Tuple2<>();
+ for (long i = 0; i < numElements; i++) {
+ assertNotNull(prober.getMatchFor(Tuple2.of(i, longString2), reuse));
+ }
+
+ table.close();
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ private static String getLongString(int length) {
+ StringBuilder bld = new StringBuilder(length);
+ for (int i = 0; i < length; i++) {
+ bld.append('a');
+ }
+ return bld.toString();
+ }
+
+
+ @Test
+ public void testOutOfMemory() {
+ try {
+ List<MemorySegment> memory = getMemory(100, 1024);
+
+ InPlaceMutableHashTable<Tuple2<Long, String>> table =
+ new InPlaceMutableHashTable<>(serializer, comparator, memory);
+
+ try {
+ final int numElements = 100000;
+
+ table.open();
+
+ // Insert too many elements
+ for (long i = 0; i < numElements; i++) {
+ table.insertOrReplaceRecord(Tuple2.of(i, "alma"));
+ }
+
+ fail("We should have got out of memory (EOFException)");
+ } catch (EOFException e) {
+ // OK
+ table.close();
+ }
+ } catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/52e191a5/flink-runtime/src/test/java/org/apache/flink/runtime/operators/hash/MemoryHashTableTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/hash/MemoryHashTableTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/hash/MemoryHashTableTest.java
deleted file mode 100644
index a71f028..0000000
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/hash/MemoryHashTableTest.java
+++ /dev/null
@@ -1,746 +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.lang.reflect.Field;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Random;
-
-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.core.memory.MemorySegmentFactory;
-import org.apache.flink.runtime.operators.testutils.UniformStringPairGenerator;
-import org.apache.flink.runtime.operators.testutils.types.IntList;
-import org.apache.flink.runtime.operators.testutils.types.IntListComparator;
-import org.apache.flink.runtime.operators.testutils.types.IntListPairComparator;
-import org.apache.flink.runtime.operators.testutils.types.IntListSerializer;
-import org.apache.flink.runtime.operators.testutils.types.IntPair;
-import org.apache.flink.runtime.operators.testutils.types.IntPairComparator;
-import org.apache.flink.runtime.operators.testutils.types.IntPairListPairComparator;
-import org.apache.flink.runtime.operators.testutils.types.IntPairPairComparator;
-import org.apache.flink.runtime.operators.testutils.types.IntPairSerializer;
-import org.apache.flink.runtime.operators.testutils.types.StringPair;
-import org.apache.flink.runtime.operators.testutils.types.StringPairComparator;
-import org.apache.flink.runtime.operators.testutils.types.StringPairPairComparator;
-import org.apache.flink.runtime.operators.testutils.types.StringPairSerializer;
-import org.apache.flink.util.MutableObjectIterator;
-
-import org.junit.Test;
-
-import org.powermock.reflect.Whitebox;
-
-import static org.junit.Assert.*;
-
-
-public class MemoryHashTableTest {
-
- private static final long RANDOM_SEED = 76518743207143L;
-
- private static final int KEY_VALUE_DIFF = 1021;
-
- private static final int PAGE_SIZE = 16 * 1024;
-
-
- private final Random rnd = new Random(RANDOM_SEED);
-
- private final TypeSerializer<IntPair> serializer = new IntPairSerializer();
-
- private final TypeComparator<IntPair> comparator = new IntPairComparator();
-
- private final TypePairComparator<IntPair, IntPair> pairComparator = new IntPairPairComparator();
-
-
- private static final int MAX_LIST_SIZE = 8;
-
- private final TypeSerializer<IntList> serializerV = new IntListSerializer();
-
- private final TypeComparator<IntList> comparatorV = new IntListComparator();
-
- private final TypePairComparator<IntList, IntList> pairComparatorV = new IntListPairComparator();
-
- private final TypePairComparator<IntPair, IntList> pairComparatorPL =new IntPairListPairComparator();
-
- private final int SIZE = 75;
-
- private final int NUM_PAIRS = 100000;
-
- private final int NUM_LISTS = 100000;
-
- private final int ADDITIONAL_MEM = 100;
-
- private final int NUM_REWRITES = 10;
-
-
- private final TypeSerializer<StringPair> serializerS = new StringPairSerializer();
-
- private final TypeComparator<StringPair> comparatorS = new StringPairComparator();
-
- private final TypePairComparator<StringPair, StringPair> pairComparatorS = new StringPairPairComparator();
-
-
- @Test
- public void testDifferentProbers() {
- final int NUM_MEM_PAGES = 32 * NUM_PAIRS / PAGE_SIZE;
-
- AbstractMutableHashTable<IntPair> table = new CompactingHashTable<IntPair>(serializer, comparator, getMemory(NUM_MEM_PAGES, PAGE_SIZE));
-
- AbstractHashTableProber<IntPair, IntPair> prober1 = table.getProber(comparator, pairComparator);
- AbstractHashTableProber<IntPair, IntPair> prober2 = table.getProber(comparator, pairComparator);
-
- assertFalse(prober1 == prober2);
-
- table.close();
- assertEquals("Memory lost", NUM_MEM_PAGES, table.getFreeMemory().size());
- }
-
- @Test
- public void testBuildAndRetrieve() {
- try {
- final int NUM_MEM_PAGES = 32 * NUM_PAIRS / PAGE_SIZE;
-
- final IntPair[] pairs = getRandomizedIntPairs(NUM_PAIRS, rnd);
-
- AbstractMutableHashTable<IntPair> table = new CompactingHashTable<IntPair>(serializer, comparator, getMemory(NUM_MEM_PAGES, PAGE_SIZE));
- table.open();
-
- for (int i = 0; i < NUM_PAIRS; i++) {
- table.insert(pairs[i]);
- }
-
- AbstractHashTableProber<IntPair, IntPair> prober = table.getProber(comparator, pairComparator);
- IntPair target = new IntPair();
-
- for (int i = 0; i < NUM_PAIRS; i++) {
- assertNotNull(prober.getMatchFor(pairs[i], target));
- assertEquals(pairs[i].getValue(), target.getValue());
- }
-
-
- table.close();
- assertEquals("Memory lost", NUM_MEM_PAGES, table.getFreeMemory().size());
- } catch (Exception e) {
- e.printStackTrace();
- fail("Error: " + e.getMessage());
- }
- }
-
- @Test
- public void testEntryIterator() {
- try {
- final int NUM_MEM_PAGES = SIZE * NUM_LISTS / PAGE_SIZE;
- final IntList[] lists = getRandomizedIntLists(NUM_LISTS, rnd);
-
- AbstractMutableHashTable<IntList> table = new CompactingHashTable<IntList>(serializerV, comparatorV, getMemory(NUM_MEM_PAGES, PAGE_SIZE));
- table.open();
- int result = 0;
- for (int i = 0; i < NUM_LISTS; i++) {
- table.insert(lists[i]);
- result += lists[i].getKey();
- }
-
- MutableObjectIterator<IntList> iter = table.getEntryIterator();
- IntList target = new IntList();
-
- int sum = 0;
- while((target = iter.next(target)) != null) {
- sum += target.getKey();
- }
- table.close();
-
- assertTrue(sum == result);
- assertEquals("Memory lost", NUM_MEM_PAGES, table.getFreeMemory().size());
- } catch (Exception e) {
- e.printStackTrace();
- fail("Error: " + e.getMessage());
- }
- }
-
- @Test
- public void testMultipleProbers() {
- try {
- final int NUM_MEM_PAGES = SIZE * NUM_LISTS / PAGE_SIZE;
- final IntList[] lists = getRandomizedIntLists(NUM_LISTS, rnd);
- final IntPair[] pairs = getRandomizedIntPairs(NUM_LISTS, rnd);
-
- AbstractMutableHashTable<IntList> table = new CompactingHashTable<IntList>(serializerV, comparatorV, getMemory(NUM_MEM_PAGES, PAGE_SIZE));
- table.open();
- for (int i = 0; i < NUM_LISTS; i++) {
- table.insert(lists[i]);
- }
-
- AbstractHashTableProber<IntList, IntList> listProber = table.getProber(comparatorV, pairComparatorV);
-
- AbstractHashTableProber<IntPair, IntList> pairProber = table.getProber(comparator, pairComparatorPL);
-
- IntList target = new IntList();
- for (int i = 0; i < NUM_LISTS; i++) {
- assertNotNull(pairProber.getMatchFor(pairs[i], target));
- assertNotNull(listProber.getMatchFor(lists[i], target));
- assertArrayEquals(lists[i].getValue(), target.getValue());
- }
- table.close();
- assertEquals("Memory lost", NUM_MEM_PAGES, table.getFreeMemory().size());
- } catch (Exception e) {
- e.printStackTrace();
- fail("Error: " + e.getMessage());
- }
- }
-
- @Test
- public void testVariableLengthBuildAndRetrieve() {
- try {
- final int NUM_MEM_PAGES = SIZE * NUM_LISTS / PAGE_SIZE;
-
- final IntList[] lists = getRandomizedIntLists(NUM_LISTS, rnd);
-
- AbstractMutableHashTable<IntList> table = new CompactingHashTable<IntList>(serializerV, comparatorV, getMemory(NUM_MEM_PAGES, PAGE_SIZE));
- table.open();
-
- for (int i = 0; i < NUM_LISTS; i++) {
- try {
- table.insert(lists[i]);
- } catch (Exception e) {
- throw e;
- }
- }
-
-
- AbstractHashTableProber<IntList, IntList> prober = table.getProber(comparatorV, pairComparatorV);
- IntList target = new IntList();
-
- for (int i = 0; i < NUM_LISTS; i++) {
- assertNotNull(prober.getMatchFor(lists[i], target));
- assertArrayEquals(lists[i].getValue(), target.getValue());
- }
-
- final IntList[] overwriteLists = getRandomizedIntLists(NUM_LISTS, rnd);
-
- // test replacing
- for (int i = 0; i < NUM_LISTS; i++) {
- table.insertOrReplaceRecord(overwriteLists[i]);
- }
-
- for (int i = 0; i < NUM_LISTS; i++) {
- assertNotNull("" + i, prober.getMatchFor(overwriteLists[i], target));
- assertArrayEquals(overwriteLists[i].getValue(), target.getValue());
- }
-
- table.close();
- assertEquals("Memory lost", NUM_MEM_PAGES, table.getFreeMemory().size());
- } catch (Exception e) {
- e.printStackTrace();
- fail("Error: " + e.getMessage());
- }
- }
-
- @Test
- public void testVariableLengthBuildAndRetrieveMajorityUpdated() {
- try {
- final int NUM_MEM_PAGES = SIZE * NUM_LISTS / PAGE_SIZE;
-
- final IntList[] lists = getRandomizedIntLists(NUM_LISTS, rnd);
-
- AbstractMutableHashTable<IntList> table = new CompactingHashTable<IntList>(serializerV, comparatorV, getMemory(NUM_MEM_PAGES, PAGE_SIZE));
- table.open();
-
- for (int i = 0; i < NUM_LISTS; i++) {
- table.insert(lists[i]);
- }
-
- AbstractHashTableProber<IntList, IntList> prober = table.getProber(comparatorV, pairComparatorV);
- IntList target = new IntList();
-
- for (int i = 0; i < NUM_LISTS; i++) {
- assertNotNull(prober.getMatchFor(lists[i], target));
- assertArrayEquals(lists[i].getValue(), target.getValue());
- }
-
- final IntList[] overwriteLists = getRandomizedIntLists(NUM_LISTS, rnd);
-
- // test replacing
- for (int i = 0; i < NUM_LISTS; i++) {
- if( i % 100 != 0) {
- table.insertOrReplaceRecord(overwriteLists[i]);
- lists[i] = overwriteLists[i];
- }
- }
-
- for (int i = 0; i < NUM_LISTS; i++) {
- assertNotNull("" + i, prober.getMatchFor(lists[i], target));
- assertArrayEquals(lists[i].getValue(), target.getValue());
- }
-
- table.close();
- assertEquals("Memory lost", NUM_MEM_PAGES, table.getFreeMemory().size());
- } catch (Exception e) {
- e.printStackTrace();
- fail("Error: " + e.getMessage());
- }
- }
-
- @Test
- public void testVariableLengthBuildAndRetrieveMinorityUpdated() {
- try {
- final int NUM_LISTS = 20000;
- final int NUM_MEM_PAGES = SIZE * NUM_LISTS / PAGE_SIZE;
-
- final int STEP_SIZE = 100;
-
- final IntList[] lists = getRandomizedIntLists(NUM_LISTS, rnd);
-
- AbstractMutableHashTable<IntList> table = new CompactingHashTable<IntList>(serializerV, comparatorV, getMemory(NUM_MEM_PAGES, PAGE_SIZE));
- table.open();
-
- for (int i = 0; i < NUM_LISTS; i++) {
- table.insert(lists[i]);
- }
-
- AbstractHashTableProber<IntList, IntList> prober = table.getProber(comparatorV, pairComparatorV);
- IntList target = new IntList();
-
- for (int i = 0; i < NUM_LISTS; i++) {
- assertNotNull(prober.getMatchFor(lists[i], target));
- assertArrayEquals(lists[i].getValue(), target.getValue());
- }
-
- final IntList[] overwriteLists = getRandomizedIntLists(NUM_LISTS/STEP_SIZE, rnd);
-
- // test replacing
- for (int i = 0; i < NUM_LISTS; i += STEP_SIZE) {
- overwriteLists[i/STEP_SIZE].setKey(overwriteLists[i/STEP_SIZE].getKey()*STEP_SIZE);
- table.insertOrReplaceRecord(overwriteLists[i/STEP_SIZE]);
- lists[i] = overwriteLists[i/STEP_SIZE];
- }
-
- for (int i = 0; i < NUM_LISTS; i++) {
- assertNotNull(prober.getMatchFor(lists[i], target));
- assertArrayEquals(lists[i].getValue(), target.getValue());
- }
-
- table.close();
- assertEquals("Memory lost", NUM_MEM_PAGES, table.getFreeMemory().size());
- } catch (Exception e) {
- e.printStackTrace();
- fail("Error: " + e.getMessage());
- }
- }
-
- @Test
- public void testRepeatedBuildAndRetrieve() {
- try {
- final int NUM_MEM_PAGES = SIZE * NUM_LISTS / PAGE_SIZE;
-
- final IntList[] lists = getRandomizedIntLists(NUM_LISTS, rnd);
-
- AbstractMutableHashTable<IntList> table = new CompactingHashTable<IntList>(serializerV, comparatorV, getMemory(NUM_MEM_PAGES, PAGE_SIZE));
- table.open();
-
- for (int i = 0; i < NUM_LISTS; i++) {
- try {
- table.insert(lists[i]);
- } catch (Exception e) {
- throw e;
- }
- }
-
-
- AbstractHashTableProber<IntList, IntList> prober = table.getProber(comparatorV, pairComparatorV);
- IntList target = new IntList();
-
- for (int i = 0; i < NUM_LISTS; i++) {
- assertNotNull(prober.getMatchFor(lists[i], target));
- assertArrayEquals(lists[i].getValue(), target.getValue());
- }
-
- IntList[] overwriteLists;
-
- for(int k = 0; k < NUM_REWRITES; k++) {
- overwriteLists = getRandomizedIntLists(NUM_LISTS, rnd);
- // test replacing
- for (int i = 0; i < NUM_LISTS; i++) {
- table.insertOrReplaceRecord(overwriteLists[i]);
- }
-
- for (int i = 0; i < NUM_LISTS; i++) {
- assertNotNull("" + i, prober.getMatchFor(overwriteLists[i], target));
- assertArrayEquals(overwriteLists[i].getValue(), target.getValue());
- }
- }
-
- table.close();
- assertEquals("Memory lost", NUM_MEM_PAGES, table.getFreeMemory().size());
- } catch (Exception e) {
- e.printStackTrace();
- fail("Error: " + e.getMessage());
- }
- }
-
- @Test
- public void testProberUpdate() {
- try {
- final int NUM_MEM_PAGES = SIZE * NUM_LISTS / PAGE_SIZE;
-
- final IntList[] lists = getRandomizedIntLists(NUM_LISTS, rnd);
-
- AbstractMutableHashTable<IntList> table = new CompactingHashTable<IntList>(serializerV, comparatorV, getMemory(NUM_MEM_PAGES, PAGE_SIZE));
- table.open();
-
- for (int i = 0; i < NUM_LISTS; i++) {
- table.insert(lists[i]);
- }
-
- final IntList[] overwriteLists = getRandomizedIntLists(NUM_LISTS, rnd);
-
- AbstractHashTableProber<IntList, IntList> prober = table.getProber(comparatorV, pairComparatorV);
- IntList target = new IntList();
-
- for (int i = 0; i < NUM_LISTS; i++) {
- assertNotNull(""+i,prober.getMatchFor(lists[i], target));
- assertArrayEquals(lists[i].getValue(), target.getValue());
- prober.updateMatch(overwriteLists[i]);
- }
-
- for (int i = 0; i < NUM_LISTS; i++) {
- assertNotNull("" + i, prober.getMatchFor(overwriteLists[i], target));
- assertArrayEquals(overwriteLists[i].getValue(), target.getValue());
- }
-
- table.close();
- assertEquals("Memory lost", NUM_MEM_PAGES, table.getFreeMemory().size());
- } catch (Exception e) {
- e.printStackTrace();
- fail("Error: " + e.getMessage());
- }
- }
-
- @Test
- public void testResize() {
- try {
- final int NUM_MEM_PAGES = 30 * NUM_PAIRS / PAGE_SIZE;
- final IntPair[] pairs = getRandomizedIntPairs(NUM_PAIRS, rnd);
-
- List<MemorySegment> memory = getMemory(NUM_MEM_PAGES, PAGE_SIZE);
- CompactingHashTable<IntPair> table = new CompactingHashTable<IntPair>(serializer, comparator, memory);
- table.open();
-
- for (int i = 0; i < NUM_PAIRS; i++) {
- table.insert(pairs[i]);
- }
-
- AbstractHashTableProber<IntPair, IntPair> prober = table.getProber(comparator, pairComparator);
- IntPair target = new IntPair();
-
- for (int i = 0; i < NUM_PAIRS; i++) {
- assertNotNull(prober.getMatchFor(pairs[i], target));
- assertEquals(pairs[i].getValue(), target.getValue());
- }
-
- // make sure there is enough memory for resize
- memory.addAll(getMemory(ADDITIONAL_MEM, PAGE_SIZE));
- Boolean b = Whitebox.<Boolean>invokeMethod(table, "resizeHashTable");
- assertTrue(b.booleanValue());
-
- for (int i = 0; i < NUM_PAIRS; i++) {
- assertNotNull(pairs[i].getKey() + " " + pairs[i].getValue(), prober.getMatchFor(pairs[i], target));
- assertEquals(pairs[i].getValue(), target.getValue());
- }
-
- table.close();
- assertEquals("Memory lost", NUM_MEM_PAGES + ADDITIONAL_MEM, table.getFreeMemory().size());
- } catch (Exception e) {
- e.printStackTrace();
- fail("Error: " + e.getMessage());
- }
- }
-
- @Test
- public void testDoubleResize() {
- try {
- final int NUM_MEM_PAGES = 30 * NUM_PAIRS / PAGE_SIZE;
- final IntPair[] pairs = getRandomizedIntPairs(NUM_PAIRS, rnd);
-
- List<MemorySegment> memory = getMemory(NUM_MEM_PAGES, PAGE_SIZE);
- CompactingHashTable<IntPair> table = new CompactingHashTable<IntPair>(serializer, comparator, memory);
- table.open();
-
- for (int i = 0; i < NUM_PAIRS; i++) {
- table.insert(pairs[i]);
- }
-
- AbstractHashTableProber<IntPair, IntPair> prober = table.getProber(comparator, pairComparator);
- IntPair target = new IntPair();
-
- for (int i = 0; i < NUM_PAIRS; i++) {
- assertNotNull(prober.getMatchFor(pairs[i], target));
- assertEquals(pairs[i].getValue(), target.getValue());
- }
-
- // make sure there is enough memory for resize
- memory.addAll(getMemory(ADDITIONAL_MEM, PAGE_SIZE));
- Boolean b = Whitebox.<Boolean>invokeMethod(table, "resizeHashTable");
- assertTrue(b.booleanValue());
-
- for (int i = 0; i < NUM_PAIRS; i++) {
- assertNotNull(pairs[i].getKey() + " " + pairs[i].getValue(), prober.getMatchFor(pairs[i], target));
- assertEquals(pairs[i].getValue(), target.getValue());
- }
-
- // make sure there is enough memory for resize
- memory.addAll(getMemory(ADDITIONAL_MEM, PAGE_SIZE));
- b = Whitebox.<Boolean>invokeMethod(table, "resizeHashTable");
- assertTrue(b.booleanValue());
-
- for (int i = 0; i < NUM_PAIRS; i++) {
- assertNotNull(pairs[i].getKey() + " " + pairs[i].getValue(), prober.getMatchFor(pairs[i], target));
- assertEquals(pairs[i].getValue(), target.getValue());
- }
-
- table.close();
- assertEquals("Memory lost", NUM_MEM_PAGES + ADDITIONAL_MEM + ADDITIONAL_MEM, table.getFreeMemory().size());
- } catch (Exception e) {
- e.printStackTrace();
- fail("Error: " + e.getMessage());
- }
- }
-
- @Test
- public void testTripleResize() {
- try {
- final int NUM_MEM_PAGES = 30 * NUM_PAIRS / PAGE_SIZE;
- final IntPair[] pairs = getRandomizedIntPairs(NUM_PAIRS, rnd);
-
- List<MemorySegment> memory = getMemory(NUM_MEM_PAGES, PAGE_SIZE);
- CompactingHashTable<IntPair> table = new CompactingHashTable<IntPair>(serializer, comparator, memory);
- table.open();
-
- for (int i = 0; i < NUM_PAIRS; i++) {
- table.insert(pairs[i]);
- }
-
- AbstractHashTableProber<IntPair, IntPair> prober = table.getProber(comparator, pairComparator);
- IntPair target = new IntPair();
-
- for (int i = 0; i < NUM_PAIRS; i++) {
- assertNotNull(prober.getMatchFor(pairs[i], target));
- assertEquals(pairs[i].getValue(), target.getValue());
- }
-
- // make sure there is enough memory for resize
- memory.addAll(getMemory(ADDITIONAL_MEM, PAGE_SIZE));
- Boolean b = Whitebox.<Boolean>invokeMethod(table, "resizeHashTable");
- assertTrue(b.booleanValue());
-
- for (int i = 0; i < NUM_PAIRS; i++) {
- assertNotNull(pairs[i].getKey() + " " + pairs[i].getValue(), prober.getMatchFor(pairs[i], target));
- assertEquals(pairs[i].getValue(), target.getValue());
- }
-
- // make sure there is enough memory for resize
- memory.addAll(getMemory(ADDITIONAL_MEM, PAGE_SIZE));
- b = Whitebox.<Boolean>invokeMethod(table, "resizeHashTable");
- assertTrue(b.booleanValue());
-
- for (int i = 0; i < NUM_PAIRS; i++) {
- assertNotNull(pairs[i].getKey() + " " + pairs[i].getValue(), prober.getMatchFor(pairs[i], target));
- assertEquals(pairs[i].getValue(), target.getValue());
- }
-
- // make sure there is enough memory for resize
- memory.addAll(getMemory(2*ADDITIONAL_MEM, PAGE_SIZE));
- b = Whitebox.<Boolean>invokeMethod(table, "resizeHashTable");
- assertTrue(b.booleanValue());
-
- for (int i = 0; i < NUM_PAIRS; i++) {
- assertNotNull(pairs[i].getKey() + " " + pairs[i].getValue(), prober.getMatchFor(pairs[i], target));
- assertEquals(pairs[i].getValue(), target.getValue());
- }
-
- table.close();
- assertEquals("Memory lost", NUM_MEM_PAGES + 4*ADDITIONAL_MEM, table.getFreeMemory().size());
- } catch (Exception e) {
- e.printStackTrace();
- fail("Error: " + e.getMessage());
- }
- }
-
- @Test
- public void testResizeWithCompaction(){
- try {
- final int NUM_MEM_PAGES = (SIZE * NUM_LISTS / PAGE_SIZE);
-
- final IntList[] lists = getRandomizedIntLists(NUM_LISTS, rnd);
-
- List<MemorySegment> memory = getMemory(NUM_MEM_PAGES, PAGE_SIZE);
- CompactingHashTable<IntList> table = new CompactingHashTable<IntList>(serializerV, comparatorV, memory);
- table.open();
-
- for (int i = 0; i < NUM_LISTS; i++) {
- try {
- table.insert(lists[i]);
- } catch (Exception e) {
- throw e;
- }
- }
-
- AbstractHashTableProber<IntList, IntList> prober = table.getProber(comparatorV, pairComparatorV);
- IntList target = new IntList();
-
- for (int i = 0; i < NUM_LISTS; i++) {
- assertNotNull(prober.getMatchFor(lists[i], target));
- assertArrayEquals(lists[i].getValue(), target.getValue());
- }
-
- // make sure there is enough memory for resize
- memory.addAll(getMemory(ADDITIONAL_MEM, PAGE_SIZE));
- Boolean b = Whitebox.<Boolean>invokeMethod(table, "resizeHashTable");
- assertTrue(b.booleanValue());
-
- for (int i = 0; i < NUM_LISTS; i++) {
- assertNotNull(prober.getMatchFor(lists[i], target));
- assertArrayEquals(lists[i].getValue(), target.getValue());
- }
-
- final IntList[] overwriteLists = getRandomizedIntLists(NUM_LISTS, rnd);
-
- // test replacing
- for (int i = 0; i < NUM_LISTS; i++) {
- table.insertOrReplaceRecord(overwriteLists[i]);
- }
-
- Field list = Whitebox.getField(CompactingHashTable.class, "partitions");
- @SuppressWarnings("unchecked")
- ArrayList<InMemoryPartition<IntList>> partitions = (ArrayList<InMemoryPartition<IntList>>) list.get(table);
- int numPartitions = partitions.size();
- for(int i = 0; i < numPartitions; i++) {
- Whitebox.invokeMethod(table, "compactPartition", i);
- }
-
- // make sure there is enough memory for resize
- memory.addAll(getMemory(2*ADDITIONAL_MEM, PAGE_SIZE));
- b = Whitebox.<Boolean>invokeMethod(table, "resizeHashTable");
- assertTrue(b.booleanValue());
-
- for (int i = 0; i < NUM_LISTS; i++) {
- assertNotNull("" + i, prober.getMatchFor(overwriteLists[i], target));
- assertArrayEquals(overwriteLists[i].getValue(), target.getValue());
- }
-
- table.close();
- assertEquals("Memory lost", NUM_MEM_PAGES + 3*ADDITIONAL_MEM, table.getFreeMemory().size());
- } catch (Exception e) {
- e.printStackTrace();
- fail("Error: " + e.getMessage());
- }
- }
-
- @Test
- public void testVariableLengthStringBuildAndRetrieve() {
- try {
- final int NUM_MEM_PAGES = 40 * NUM_PAIRS / PAGE_SIZE;
-
- MutableObjectIterator<StringPair> buildInput = new UniformStringPairGenerator(NUM_PAIRS, 1, false);
-
- MutableObjectIterator<StringPair> probeTester = new UniformStringPairGenerator(NUM_PAIRS, 1, false);
-
- MutableObjectIterator<StringPair> updater = new UniformStringPairGenerator(NUM_PAIRS, 1, false);
-
- MutableObjectIterator<StringPair> updateTester = new UniformStringPairGenerator(NUM_PAIRS, 1, false);
-
- AbstractMutableHashTable<StringPair> table = new CompactingHashTable<StringPair>(serializerS, comparatorS, getMemory(NUM_MEM_PAGES, PAGE_SIZE));
- table.open();
-
- StringPair target = new StringPair();
- while(buildInput.next(target) != null) {
- table.insert(target);
- }
-
- AbstractHashTableProber<StringPair, StringPair> prober = table.getProber(comparatorS, pairComparatorS);
- StringPair temp = new StringPair();
- while(probeTester.next(target) != null) {
- assertNotNull("" + target.getKey(), prober.getMatchFor(target, temp));
- assertEquals(temp.getValue(), target.getValue());
- }
-
- while(updater.next(target) != null) {
- target.setValue(target.getValue());
- table.insertOrReplaceRecord(target);
- }
-
- while (updateTester.next(target) != null) {
- assertNotNull(prober.getMatchFor(target, temp));
- assertEquals(target.getValue(), temp.getValue());
- }
-
- table.close();
- assertEquals("Memory lost", NUM_MEM_PAGES, table.getFreeMemory().size());
- } catch (Exception e) {
- e.printStackTrace();
- fail("Error: " + e.getMessage());
- }
- }
-
-
- private static IntPair[] getRandomizedIntPairs(int num, Random rnd) {
- IntPair[] pairs = new IntPair[num];
-
- // create all the pairs, dense
- for (int i = 0; i < num; i++) {
- pairs[i] = new IntPair(i, i + KEY_VALUE_DIFF);
- }
-
- // randomly swap them
- for (int i = 0; i < 2 * num; i++) {
- int pos1 = rnd.nextInt(num);
- int pos2 = rnd.nextInt(num);
-
- IntPair tmp = pairs[pos1];
- pairs[pos1] = pairs[pos2];
- pairs[pos2] = tmp;
- }
-
- return pairs;
- }
-
- private static IntList[] getRandomizedIntLists(int num, Random rnd) {
- IntList[] lists = new IntList[num];
- for (int i = 0; i < num; i++) {
- int[] value = new int[rnd.nextInt(MAX_LIST_SIZE)+1];
- //int[] value = new int[MAX_LIST_SIZE-1];
- for (int j = 0; j < value.length; j++) {
- value[j] = -rnd.nextInt(Integer.MAX_VALUE);
- }
- lists[i] = new IntList(i, value);
- }
-
- return lists;
- }
-
- private static List<MemorySegment> getMemory(int numPages, int pageSize) {
- List<MemorySegment> memory = new ArrayList<MemorySegment>();
-
- for (int i = 0; i < numPages; i++) {
- memory.add(MemorySegmentFactory.allocateUnpooledSegment(pageSize));
- }
-
- return memory;
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/52e191a5/flink-runtime/src/test/java/org/apache/flink/runtime/operators/hash/MutableHashTableTestBase.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/hash/MutableHashTableTestBase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/hash/MutableHashTableTestBase.java
new file mode 100644
index 0000000..337fb0b
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/hash/MutableHashTableTestBase.java
@@ -0,0 +1,510 @@
+/*
+ * 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.ArrayList;
+import java.util.List;
+import java.util.Random;
+
+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.api.common.typeutils.base.LongComparator;
+import org.apache.flink.api.common.typeutils.base.LongSerializer;
+import org.apache.flink.api.common.typeutils.base.StringSerializer;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.runtime.TupleComparator;
+import org.apache.flink.api.java.typeutils.runtime.TupleSerializer;
+import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.core.memory.MemorySegmentFactory;
+import org.apache.flink.runtime.operators.testutils.UniformStringPairGenerator;
+import org.apache.flink.runtime.operators.testutils.types.IntList;
+import org.apache.flink.runtime.operators.testutils.types.IntListComparator;
+import org.apache.flink.runtime.operators.testutils.types.IntListPairComparator;
+import org.apache.flink.runtime.operators.testutils.types.IntListSerializer;
+import org.apache.flink.runtime.operators.testutils.types.IntPair;
+import org.apache.flink.runtime.operators.testutils.types.IntPairComparator;
+import org.apache.flink.runtime.operators.testutils.types.IntPairListPairComparator;
+import org.apache.flink.runtime.operators.testutils.types.IntPairPairComparator;
+import org.apache.flink.runtime.operators.testutils.types.IntPairSerializer;
+import org.apache.flink.runtime.operators.testutils.types.StringPair;
+import org.apache.flink.runtime.operators.testutils.types.StringPairComparator;
+import org.apache.flink.runtime.operators.testutils.types.StringPairPairComparator;
+import org.apache.flink.runtime.operators.testutils.types.StringPairSerializer;
+import org.apache.flink.util.MutableObjectIterator;
+
+import org.junit.Test;
+
+import static org.junit.Assert.*;
+
+
+public abstract class MutableHashTableTestBase {
+
+ protected static final long RANDOM_SEED = 76518743207143L;
+
+ private static final int KEY_VALUE_DIFF = 1021;
+
+ public final int PAGE_SIZE = 16 * 1024;
+
+
+ public final TypeSerializer<IntPair> intPairSerializer = new IntPairSerializer();
+
+ public final TypeComparator<IntPair> intPairComparator = new IntPairComparator();
+
+ public final TypePairComparator<IntPair, IntPair> pairComparator = new IntPairPairComparator();
+
+
+ private static final int MAX_LIST_SIZE = 8;
+
+ public final TypeSerializer<IntList> serializerV = new IntListSerializer();
+
+ public final TypeComparator<IntList> comparatorV = new IntListComparator();
+
+ public final TypePairComparator<IntList, IntList> pairComparatorV = new IntListPairComparator();
+
+ public final TypePairComparator<IntPair, IntList> pairComparatorPL = new IntPairListPairComparator();
+
+ @SuppressWarnings("unchecked")
+ protected TupleSerializer<Tuple2<Long, String>> tuple2LongStringSerializer =
+ new TupleSerializer<>(
+ (Class<Tuple2<Long, String>>) (Class<?>) Tuple2.class,
+ new TypeSerializer<?>[] { LongSerializer.INSTANCE, StringSerializer.INSTANCE });
+
+ protected TupleComparator<Tuple2<Long, String>> tuple2LongStringComparator =
+ new TupleComparator<>(
+ new int[] {0},
+ new TypeComparator<?>[] { new LongComparator(true) },
+ new TypeSerializer<?>[] { LongSerializer.INSTANCE });
+
+ public final int SIZE = 75;
+
+ public final int NUM_PAIRS = 100000;
+
+ public final int NUM_LISTS = 100000;
+
+ protected final int ADDITIONAL_MEM = 100;
+
+ private final int NUM_REWRITES = 10;
+
+
+ public final TypeSerializer<StringPair> serializerS = new StringPairSerializer();
+
+ public final TypeComparator<StringPair> comparatorS = new StringPairComparator();
+
+ private final TypePairComparator<StringPair, StringPair> pairComparatorS = new StringPairPairComparator();
+
+
+ abstract protected <T> AbstractMutableHashTable<T> getHashTable(
+ TypeSerializer<T> serializer, TypeComparator<T> comparator, List<MemorySegment> memory);
+
+ @Test
+ public void testDifferentProbers() {
+ final int NUM_MEM_PAGES = 32 * NUM_PAIRS / PAGE_SIZE;
+ AbstractMutableHashTable<IntPair> table = getHashTable(intPairSerializer, intPairComparator, getMemory(NUM_MEM_PAGES));
+
+ AbstractHashTableProber<IntPair, IntPair> prober1 = table.getProber(intPairComparator, pairComparator);
+ AbstractHashTableProber<IntPair, IntPair> prober2 = table.getProber(intPairComparator, pairComparator);
+
+ assertFalse(prober1 == prober2);
+
+ table.close(); // (This also tests calling close without calling open first.)
+ assertEquals("Memory lost", NUM_MEM_PAGES, table.getFreeMemory().size());
+ }
+
+ @Test
+ public void testBuildAndRetrieve() throws Exception {
+ final int NUM_MEM_PAGES = 32 * NUM_PAIRS / PAGE_SIZE;
+ AbstractMutableHashTable<IntPair> table = getHashTable(intPairSerializer, intPairComparator, getMemory(NUM_MEM_PAGES));
+
+ final Random rnd = new Random(RANDOM_SEED);
+ final IntPair[] pairs = getRandomizedIntPairs(NUM_PAIRS, rnd);
+
+ table.open();
+
+ for (int i = 0; i < NUM_PAIRS; i++) {
+ table.insert(pairs[i]);
+ }
+
+ AbstractHashTableProber<IntPair, IntPair> prober = table.getProber(intPairComparator, pairComparator);
+ IntPair target = new IntPair();
+
+ for (int i = 0; i < NUM_PAIRS; i++) {
+ assertNotNull(prober.getMatchFor(pairs[i], target));
+ assertEquals(pairs[i].getValue(), target.getValue());
+ }
+
+
+ table.close();
+ assertEquals("Memory lost", NUM_MEM_PAGES, table.getFreeMemory().size());
+ }
+
+ @Test
+ public void testEntryIterator() throws Exception {
+ final int NUM_MEM_PAGES = SIZE * NUM_LISTS / PAGE_SIZE;
+ AbstractMutableHashTable<IntList> table = getHashTable(serializerV, comparatorV, getMemory(NUM_MEM_PAGES));
+
+ final Random rnd = new Random(RANDOM_SEED);
+ final IntList[] lists = getRandomizedIntLists(NUM_LISTS, rnd);
+
+ table.open();
+ int result = 0;
+ for (int i = 0; i < NUM_LISTS; i++) {
+ table.insert(lists[i]);
+ result += lists[i].getKey();
+ }
+
+ MutableObjectIterator<IntList> iter = table.getEntryIterator();
+ IntList target = new IntList();
+
+ int sum = 0;
+ while((target = iter.next(target)) != null) {
+ sum += target.getKey();
+ }
+ table.close();
+
+ assertTrue(sum == result);
+ assertEquals("Memory lost", NUM_MEM_PAGES, table.getFreeMemory().size());
+ }
+
+ @Test
+ public void testMultipleProbers() throws Exception {
+ final int NUM_MEM_PAGES = SIZE * NUM_LISTS / PAGE_SIZE;
+ AbstractMutableHashTable<IntList> table = getHashTable(serializerV, comparatorV, getMemory(NUM_MEM_PAGES));
+
+ final Random rnd = new Random(RANDOM_SEED);
+ final IntList[] lists = getRandomizedIntLists(NUM_LISTS, rnd);
+ final IntPair[] pairs = getRandomizedIntPairs(NUM_LISTS, rnd);
+
+ table.open();
+ for (int i = 0; i < NUM_LISTS; i++) {
+ table.insert(lists[i]);
+ }
+
+ AbstractHashTableProber<IntList, IntList> listProber = table.getProber(comparatorV, pairComparatorV);
+
+ AbstractHashTableProber<IntPair, IntList> pairProber = table.getProber(intPairComparator, pairComparatorPL);
+
+ IntList target = new IntList();
+ for (int i = 0; i < NUM_LISTS; i++) {
+ assertNotNull(pairProber.getMatchFor(pairs[i], target));
+ assertNotNull(listProber.getMatchFor(lists[i], target));
+ assertArrayEquals(lists[i].getValue(), target.getValue());
+ }
+ table.close();
+ assertEquals("Memory lost", NUM_MEM_PAGES, table.getFreeMemory().size());
+ }
+
+ @Test
+ public void testVariableLengthBuildAndRetrieve() throws Exception {
+ final int NUM_MEM_PAGES = SIZE * NUM_LISTS / PAGE_SIZE;
+ AbstractMutableHashTable<IntList> table = getHashTable(serializerV, comparatorV, getMemory(NUM_MEM_PAGES));
+
+ final Random rnd = new Random(RANDOM_SEED);
+ final IntList[] lists = getRandomizedIntLists(NUM_LISTS, rnd);
+
+ table.open();
+
+ for (int i = 0; i < NUM_LISTS; i++) {
+ try {
+ table.insert(lists[i]);
+ } catch (Exception e) {
+ throw e;
+ }
+ }
+
+
+ AbstractHashTableProber<IntList, IntList> prober = table.getProber(comparatorV, pairComparatorV);
+ IntList target = new IntList();
+
+ for (int i = 0; i < NUM_LISTS; i++) {
+ assertNotNull(prober.getMatchFor(lists[i], target));
+ assertArrayEquals(lists[i].getValue(), target.getValue());
+ }
+
+ final IntList[] overwriteLists = getRandomizedIntLists(NUM_LISTS, rnd);
+
+ // test replacing
+ for (int i = 0; i < NUM_LISTS; i++) {
+ table.insertOrReplaceRecord(overwriteLists[i]);
+ }
+
+ for (int i = 0; i < NUM_LISTS; i++) {
+ assertNotNull("" + i, prober.getMatchFor(overwriteLists[i], target));
+ assertArrayEquals(overwriteLists[i].getValue(), target.getValue());
+ }
+
+ table.close();
+ assertEquals("Memory lost", NUM_MEM_PAGES, table.getFreeMemory().size());
+ }
+
+ @Test
+ public void testVariableLengthBuildAndRetrieveMajorityUpdated() throws Exception {
+ final int NUM_MEM_PAGES = SIZE * NUM_LISTS / PAGE_SIZE;
+ AbstractMutableHashTable<IntList> table = getHashTable(serializerV, comparatorV, getMemory(NUM_MEM_PAGES));
+
+ final Random rnd = new Random(RANDOM_SEED);
+ final IntList[] lists = getRandomizedIntLists(NUM_LISTS, rnd);
+
+ table.open();
+
+ for (int i = 0; i < NUM_LISTS; i++) {
+ table.insert(lists[i]);
+ }
+
+ AbstractHashTableProber<IntList, IntList> prober = table.getProber(comparatorV, pairComparatorV);
+ IntList target = new IntList();
+
+ for (int i = 0; i < NUM_LISTS; i++) {
+ assertNotNull(prober.getMatchFor(lists[i], target));
+ assertArrayEquals(lists[i].getValue(), target.getValue());
+ }
+
+ final IntList[] overwriteLists = getRandomizedIntLists(NUM_LISTS, rnd);
+
+ // test replacing
+ for (int i = 0; i < NUM_LISTS; i++) {
+ if( i % 100 != 0) {
+ table.insertOrReplaceRecord(overwriteLists[i]);
+ lists[i] = overwriteLists[i];
+ }
+ }
+
+ for (int i = 0; i < NUM_LISTS; i++) {
+ assertNotNull("" + i, prober.getMatchFor(lists[i], target));
+ assertArrayEquals(lists[i].getValue(), target.getValue());
+ }
+
+ table.close();
+ assertEquals("Memory lost", NUM_MEM_PAGES, table.getFreeMemory().size());
+ }
+
+ @Test
+ public void testVariableLengthBuildAndRetrieveMinorityUpdated() throws Exception {
+ final int NUM_LISTS = 20000;
+ final int NUM_MEM_PAGES = SIZE * NUM_LISTS / PAGE_SIZE;
+ AbstractMutableHashTable<IntList> table =
+ getHashTable(serializerV, comparatorV, getMemory(NUM_MEM_PAGES));
+
+ final int STEP_SIZE = 100;
+
+ final Random rnd = new Random(RANDOM_SEED);
+ final IntList[] lists = getRandomizedIntLists(NUM_LISTS, rnd);
+
+ table.open();
+
+ for (int i = 0; i < NUM_LISTS; i++) {
+ table.insert(lists[i]);
+ }
+
+ AbstractHashTableProber<IntList, IntList> prober = table.getProber(comparatorV, pairComparatorV);
+ IntList target = new IntList();
+
+ for (int i = 0; i < NUM_LISTS; i++) {
+ assertNotNull(prober.getMatchFor(lists[i], target));
+ assertArrayEquals(lists[i].getValue(), target.getValue());
+ }
+
+ final IntList[] overwriteLists = getRandomizedIntLists(NUM_LISTS/STEP_SIZE, rnd);
+
+ // test replacing
+ for (int i = 0; i < NUM_LISTS; i += STEP_SIZE) {
+ overwriteLists[i/STEP_SIZE].setKey(overwriteLists[i/STEP_SIZE].getKey()*STEP_SIZE);
+ table.insertOrReplaceRecord(overwriteLists[i/STEP_SIZE]);
+ lists[i] = overwriteLists[i/STEP_SIZE];
+ }
+
+ for (int i = 0; i < NUM_LISTS; i++) {
+ assertNotNull(prober.getMatchFor(lists[i], target));
+ assertArrayEquals(lists[i].getValue(), target.getValue());
+ }
+
+ table.close();
+ assertEquals("Memory lost", NUM_MEM_PAGES, table.getFreeMemory().size());
+ }
+
+ @Test
+ public void testRepeatedBuildAndRetrieve() throws Exception {
+ final int NUM_MEM_PAGES = SIZE * NUM_LISTS / PAGE_SIZE;
+ AbstractMutableHashTable<IntList> table = getHashTable(serializerV, comparatorV, getMemory(NUM_MEM_PAGES));
+
+ final Random rnd = new Random(RANDOM_SEED);
+ final IntList[] lists = getRandomizedIntLists(NUM_LISTS, rnd);
+
+ table.open();
+
+ for (int i = 0; i < NUM_LISTS; i++) {
+ try {
+ table.insert(lists[i]);
+ } catch (Exception e) {
+ throw e;
+ }
+ }
+
+
+ AbstractHashTableProber<IntList, IntList> prober = table.getProber(comparatorV, pairComparatorV);
+ IntList target = new IntList();
+
+ for (int i = 0; i < NUM_LISTS; i++) {
+ assertNotNull(prober.getMatchFor(lists[i], target));
+ assertArrayEquals(lists[i].getValue(), target.getValue());
+ }
+
+ IntList[] overwriteLists;
+
+ for(int k = 0; k < NUM_REWRITES; k++) {
+ overwriteLists = getRandomizedIntLists(NUM_LISTS, rnd);
+ // test replacing
+ for (int i = 0; i < NUM_LISTS; i++) {
+ table.insertOrReplaceRecord(overwriteLists[i]);
+ }
+
+ for (int i = 0; i < NUM_LISTS; i++) {
+ assertNotNull("" + i, prober.getMatchFor(overwriteLists[i], target));
+ assertArrayEquals(overwriteLists[i].getValue(), target.getValue());
+ }
+ }
+
+ table.close();
+ assertEquals("Memory lost", NUM_MEM_PAGES, table.getFreeMemory().size());
+ }
+
+ @Test
+ public void testProberUpdate() throws Exception {
+ final int NUM_MEM_PAGES = SIZE * NUM_LISTS / PAGE_SIZE;
+ AbstractMutableHashTable<IntList> table = getHashTable(serializerV, comparatorV, getMemory(NUM_MEM_PAGES));
+
+ final Random rnd = new Random(RANDOM_SEED);
+ final IntList[] lists = getRandomizedIntLists(NUM_LISTS, rnd);
+
+ table.open();
+
+ for (int i = 0; i < NUM_LISTS; i++) {
+ table.insert(lists[i]);
+ }
+
+ final IntList[] overwriteLists = getRandomizedIntLists(NUM_LISTS, rnd);
+
+ AbstractHashTableProber<IntList, IntList> prober = table.getProber(comparatorV, pairComparatorV);
+ IntList target = new IntList();
+
+ for (int i = 0; i < NUM_LISTS; i++) {
+ assertNotNull(""+i,prober.getMatchFor(lists[i], target));
+ assertArrayEquals(lists[i].getValue(), target.getValue());
+ prober.updateMatch(overwriteLists[i]);
+ }
+
+ for (int i = 0; i < NUM_LISTS; i++) {
+ assertNotNull("" + i, prober.getMatchFor(overwriteLists[i], target));
+ assertArrayEquals(overwriteLists[i].getValue(), target.getValue());
+ }
+
+ table.close();
+ assertEquals("Memory lost", NUM_MEM_PAGES, table.getFreeMemory().size());
+ }
+
+ @Test
+ public void testVariableLengthStringBuildAndRetrieve() throws IOException {
+ final int NUM_MEM_PAGES = 40 * NUM_PAIRS / PAGE_SIZE;
+ AbstractMutableHashTable<StringPair> table = getHashTable(serializerS, comparatorS, getMemory(NUM_MEM_PAGES));
+
+ MutableObjectIterator<StringPair> buildInput = new UniformStringPairGenerator(NUM_PAIRS, 1, false);
+
+ MutableObjectIterator<StringPair> probeTester = new UniformStringPairGenerator(NUM_PAIRS, 1, false);
+
+ MutableObjectIterator<StringPair> updater = new UniformStringPairGenerator(NUM_PAIRS, 1, false);
+
+ MutableObjectIterator<StringPair> updateTester = new UniformStringPairGenerator(NUM_PAIRS, 1, false);
+
+ table.open();
+
+ StringPair target = new StringPair();
+ while(buildInput.next(target) != null) {
+ table.insert(target);
+ }
+
+ AbstractHashTableProber<StringPair, StringPair> prober = table.getProber(comparatorS, pairComparatorS);
+ StringPair temp = new StringPair();
+ while(probeTester.next(target) != null) {
+ assertNotNull("" + target.getKey(), prober.getMatchFor(target, temp));
+ assertEquals(temp.getValue(), target.getValue());
+ }
+
+ while(updater.next(target) != null) {
+ target.setValue(target.getValue());
+ table.insertOrReplaceRecord(target);
+ }
+
+ while (updateTester.next(target) != null) {
+ assertNotNull(prober.getMatchFor(target, temp));
+ assertEquals(target.getValue(), temp.getValue());
+ }
+
+ table.close();
+ assertEquals("Memory lost", NUM_MEM_PAGES, table.getFreeMemory().size());
+ }
+
+
+ protected static IntPair[] getRandomizedIntPairs(int num, Random rnd) {
+ IntPair[] pairs = new IntPair[num];
+
+ // create all the pairs, dense
+ for (int i = 0; i < num; i++) {
+ pairs[i] = new IntPair(i, i + KEY_VALUE_DIFF);
+ }
+
+ // randomly swap them
+ for (int i = 0; i < 2 * num; i++) {
+ int pos1 = rnd.nextInt(num);
+ int pos2 = rnd.nextInt(num);
+
+ IntPair tmp = pairs[pos1];
+ pairs[pos1] = pairs[pos2];
+ pairs[pos2] = tmp;
+ }
+
+ return pairs;
+ }
+
+ protected static IntList[] getRandomizedIntLists(int num, Random rnd) {
+ IntList[] lists = new IntList[num];
+ for (int i = 0; i < num; i++) {
+ int[] value = new int[rnd.nextInt(MAX_LIST_SIZE)+1];
+ //int[] value = new int[MAX_LIST_SIZE-1];
+ for (int j = 0; j < value.length; j++) {
+ value[j] = -rnd.nextInt(Integer.MAX_VALUE);
+ }
+ lists[i] = new IntList(i, value);
+ }
+
+ return lists;
+ }
+
+ public List<MemorySegment> getMemory(int numPages) {
+ return getMemory(numPages, PAGE_SIZE);
+ }
+
+ private static List<MemorySegment> getMemory(int numSegments, int segmentSize) {
+ ArrayList<MemorySegment> list = new ArrayList<MemorySegment>(numSegments);
+ for (int i = 0; i < numSegments; i++) {
+ list.add(MemorySegmentFactory.allocateUnpooledSegment(segmentSize));
+ }
+ return list;
+ }
+}