diff --git ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/BytesBytesMultiHashMap.java ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/BytesBytesMultiHashMap.java index dd5c621..ef84ee6 100644 --- ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/BytesBytesMultiHashMap.java +++ ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/BytesBytesMultiHashMap.java @@ -486,7 +486,7 @@ public ThreadSafeGetter createGetterForThread() { } /** Not thread-safe! Use createGetterForThread. */ - public byte getValueResult(byte[] key, int offset, int length, Result hashMapResult) { + private byte getValueResult(byte[] key, int offset, int length, Result hashMapResult) { return getValueResult(key, offset, length, hashMapResult, writeBuffers.getReadPosition()); } diff --git ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/HybridHashTableContainer.java ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/HybridHashTableContainer.java index c2abba2..26b4508 100644 --- ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/HybridHashTableContainer.java +++ ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/HybridHashTableContainer.java @@ -28,6 +28,7 @@ import org.apache.hadoop.hive.ql.exec.ExprNodeEvaluator; import org.apache.hadoop.hive.ql.exec.JoinUtil; import org.apache.hadoop.hive.ql.exec.Utilities; +import org.apache.hadoop.hive.ql.exec.persistence.BytesBytesMultiHashMap.ThreadSafeGetter; import org.apache.hadoop.hive.ql.exec.vector.VectorHashKeyWrapper; import org.apache.hadoop.hive.ql.exec.vector.VectorHashKeyWrapperBatch; import org.apache.hadoop.hive.ql.exec.vector.expressions.VectorExpressionWriter; @@ -90,7 +91,8 @@ * The triplet: hashmap (either in memory or on disk), small table container, big table container */ public static class HashPartition { - BytesBytesMultiHashMap hashMap; // In memory hashMap + final BytesBytesMultiHashMap hashMap; // In memory hashMap + final ThreadSafeGetter hashMapGetter; // concurrent access getter KeyValueContainer sidefileKVContainer; // Stores small table key/value pairs ObjectContainer matchfileObjContainer; // Stores big table rows Path hashMapLocalPath; // Local file system path for spilled hashMap @@ -107,9 +109,12 @@ public HashPartition(int threshold, float loadFactor, int wbSize, long memUsage, boolean createHashMap) { if (createHashMap) { hashMap = new BytesBytesMultiHashMap(threshold, loadFactor, wbSize, memUsage); + hashMapGetter = hashMap.createGetterForThread(); } else { hashMapSpilledOnCreation = true; hashMapOnDisk = true; + hashMap = null; + hashMapGetter = null; } this.threshold = threshold; this.loadFactor = loadFactor; @@ -623,7 +628,7 @@ public ReusableRowContainer() { return JoinUtil.JoinResult.SPILL; } else { - aliasFilter = hashPartitions[partitionId].hashMap.getValueResult(output.getData(), 0, output.getLength(), hashMapResult); + aliasFilter = hashPartitions[partitionId].hashMapGetter.getValueResult(output.getData(), 0, output.getLength(), hashMapResult); dummyRow = null; if (hashMapResult.hasRows()) { return JoinUtil.JoinResult.MATCH; diff --git ql/src/test/org/apache/hadoop/hive/ql/exec/persistence/TestBytesBytesMultiHashMap.java ql/src/test/org/apache/hadoop/hive/ql/exec/persistence/TestBytesBytesMultiHashMap.java index c417b6f..979f7dd 100644 --- ql/src/test/org/apache/hadoop/hive/ql/exec/persistence/TestBytesBytesMultiHashMap.java +++ ql/src/test/org/apache/hadoop/hive/ql/exec/persistence/TestBytesBytesMultiHashMap.java @@ -81,9 +81,9 @@ public void testGetNonExistent() throws Exception { map.put(kv2, -1); key[0] = (byte)(key[0] + 1); BytesBytesMultiHashMap.Result hashMapResult = new BytesBytesMultiHashMap.Result(); - map.getValueResult(key, 0, key.length, hashMapResult); + map.createGetterForThread().getValueResult(key, 0, key.length, hashMapResult); assertTrue(!hashMapResult.hasRows()); - map.getValueResult(key, 0, 0, hashMapResult); + map.createGetterForThread().getValueResult(key, 0, 0, hashMapResult); assertTrue(!hashMapResult.hasRows()); } @@ -101,7 +101,7 @@ public void testPutWithFullMap() throws Exception { assertEquals(CAPACITY, map.getCapacity()); // Get of non-existent key should terminate.. BytesBytesMultiHashMap.Result hashMapResult = new BytesBytesMultiHashMap.Result(); - map.getValueResult(new byte[0], 0, 0, hashMapResult); + map.createGetterForThread().getValueResult(new byte[0], 0, 0, hashMapResult); } @Test @@ -120,7 +120,7 @@ public void testExpand() throws Exception { private void verifyHashMapResult(BytesBytesMultiHashMap map, byte[] key, byte[]... values) { BytesBytesMultiHashMap.Result hashMapResult = new BytesBytesMultiHashMap.Result(); - byte state = map.getValueResult(key, 0, key.length, hashMapResult); + byte state = map.createGetterForThread().getValueResult(key, 0, key.length, hashMapResult); HashSet hs = new HashSet(); int count = 0; if (hashMapResult.hasRows()) {