You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by se...@apache.org on 2015/04/09 04:12:23 UTC
svn commit: r1672229 - in /hive/trunk:
ql/src/java/org/apache/hadoop/hive/ql/exec/
ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/
ql/src/test/org/apache/hadoop/hive/ql/exec/persistence/
serde/src/java/org/apache/hadoop/hive/serde2/
Author: sershe
Date: Thu Apr 9 02:12:23 2015
New Revision: 1672229
URL: http://svn.apache.org/r1672229
Log:
HIVE-10131 : LLAP: BytesBytesMultiHashMap and mapjoin container should reuse refs (Matt McCline, reviewed by Sergey Shelukhin)
Modified:
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/CommonJoinOperator.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/CommonMergeJoinOperator.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/SMBMapJoinOperator.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/AbstractRowContainer.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/BytesBytesMultiHashMap.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/FlatRowContainer.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/HybridHashTableContainer.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/MapJoinBytesTableContainer.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/MapJoinEagerRowContainer.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/RowContainer.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/UnwrapRowContainer.java
hive/trunk/ql/src/test/org/apache/hadoop/hive/ql/exec/persistence/TestBytesBytesMultiHashMap.java
hive/trunk/serde/src/java/org/apache/hadoop/hive/serde2/ByteStream.java
hive/trunk/serde/src/java/org/apache/hadoop/hive/serde2/WriteBuffers.java
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/CommonJoinOperator.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/CommonJoinOperator.java?rev=1672229&r1=1672228&r2=1672229&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/CommonJoinOperator.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/CommonJoinOperator.java Thu Apr 9 02:12:23 2015
@@ -691,11 +691,11 @@ public abstract class CommonJoinOperator
Byte alias = order[i];
AbstractRowContainer<List<Object>> alw = storage[alias];
- if (alw.rowCount() != 1) {
+ if (!alw.isSingleRow()) {
allOne = false;
}
- if (alw.rowCount() == 0) {
+ if (!alw.hasRows()) {
alw.addRow(dummyObj[i]);
hasNulls = true;
} else if (condn[i].getPreserved()) {
@@ -721,16 +721,16 @@ public abstract class CommonJoinOperator
AbstractRowContainer<List<Object>> alw = storage[alias];
if (noOuterJoin) {
- if (alw.rowCount() == 0) {
+ if (!alw.hasRows()) {
return;
- } else if (alw.rowCount() > 1) {
+ } else if (!alw.isSingleRow()) {
mayHasMoreThanOne = true;
}
} else {
- if (alw.rowCount() == 0) {
+ if (!alw.hasRows()) {
hasEmpty = true;
alw.addRow(dummyObj[i]);
- } else if (!hasEmpty && alw.rowCount() == 1) {
+ } else if (!hasEmpty && alw.isSingleRow()) {
if (hasAnyFiltered(alias, alw.rowIter().first())) {
hasEmpty = true;
}
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/CommonMergeJoinOperator.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/CommonMergeJoinOperator.java?rev=1672229&r1=1672228&r2=1672229&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/CommonMergeJoinOperator.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/CommonMergeJoinOperator.java Thu Apr 9 02:12:23 2015
@@ -391,7 +391,7 @@ public class CommonMergeJoinOperator ext
if (candidateStorage[pos] == null) {
continue;
}
- if (this.candidateStorage[pos].rowCount() > 0) {
+ if (this.candidateStorage[pos].hasRows()) {
dataInCache = true;
break;
}
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/SMBMapJoinOperator.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/SMBMapJoinOperator.java?rev=1672229&r1=1672228&r2=1672229&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/SMBMapJoinOperator.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/SMBMapJoinOperator.java Thu Apr 9 02:12:23 2015
@@ -342,7 +342,7 @@ public class SMBMapJoinOperator extends
joinOneGroup();
dataInCache = false;
for (byte pos = 0; pos < order.length; pos++) {
- if (this.candidateStorage[pos].rowCount() > 0) {
+ if (this.candidateStorage[pos].hasRows()) {
dataInCache = true;
break;
}
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/AbstractRowContainer.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/AbstractRowContainer.java?rev=1672229&r1=1672228&r2=1672229&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/AbstractRowContainer.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/AbstractRowContainer.java Thu Apr 9 02:12:23 2015
@@ -37,6 +37,17 @@ public interface AbstractRowContainer<RO
public void addRow(ROW t) throws HiveException;
/**
+ * @return whether the row container has at least 1 row.
+ * NOTE: Originally we named this isEmpty, but that name conflicted with another interface.
+ */
+ public boolean hasRows() throws HiveException;
+
+ /**
+ * @return whether the row container has 1 row.
+ */
+ public boolean isSingleRow() throws HiveException;
+
+ /**
* @return number of elements in the RowContainer
*/
public int rowCount() throws HiveException;
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/BytesBytesMultiHashMap.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/BytesBytesMultiHashMap.java?rev=1672229&r1=1672228&r2=1672229&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/BytesBytesMultiHashMap.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/BytesBytesMultiHashMap.java Thu Apr 9 02:12:23 2015
@@ -186,8 +186,9 @@ public final class BytesBytesMultiHashMa
public class ThreadSafeGetter {
private WriteBuffers.Position position = new WriteBuffers.Position();
- public byte getValueRefs(byte[] key, int length, List<WriteBuffers.ByteSegmentRef> result) {
- return BytesBytesMultiHashMap.this.getValueRefs(key, length, result, position);
+ public byte getValueResult(byte[] key, int offset, int length,
+ BytesBytesMultiHashMap.Result hashMapResult) {
+ return BytesBytesMultiHashMap.this.getValueResult(key, offset, length, hashMapResult, position);
}
public void populateValue(WriteBuffers.ByteSegmentRef valueRef) {
@@ -196,6 +197,227 @@ public final class BytesBytesMultiHashMa
}
}
+ /**
+ * The result of looking up a key in the multi-hash map.
+ *
+ * This object can read through the 0, 1, or more values found for the key.
+ */
+ public static class Result {
+
+ // Whether there are more than 0 rows.
+ private boolean hasRows;
+
+ // We need a pointer to the hash map since this class must be static to support having
+ // multiple hash tables with Hybrid Grace partitioning.
+ private BytesBytesMultiHashMap hashMap;
+
+ // And, a mutable read position for thread safety when sharing a hash map.
+ private WriteBuffers.Position readPos;
+
+ // These values come from setValueResult when it finds a key. These values allow this
+ // class to read (and re-read) the values.
+ private long firstOffset;
+ private boolean hasList;
+ private long offsetAfterListRecordKeyLen;
+
+ // When we have multiple values, we save the next value record's offset here.
+ private long nextTailOffset;
+
+ // 0-based index of which row we are on.
+ private long readIndex;
+
+ // A reference to the current row.
+ private WriteBuffers.ByteSegmentRef byteSegmentRef;
+
+ public Result() {
+ hasRows = false;
+ byteSegmentRef = new WriteBuffers.ByteSegmentRef();
+ }
+
+ /**
+ * @return Whether there are 1 or more values.
+ */
+ public boolean hasRows() {
+ // NOTE: Originally we named this isEmpty, but that name conflicted with another interface.
+ return hasRows;
+ }
+
+ /**
+ * @return Whether there is just 1 value row.
+ */
+ public boolean isSingleRow() {
+ return !hasList;
+ }
+
+ /**
+ * Set internal values for reading the values after finding a key.
+ *
+ * @param hashMap
+ * The hash map we found the key in.
+ * @param firstOffset
+ * The absolute offset of the first record in the write buffers.
+ * @param hasList
+ * Whether there are multiple values (true) or just a single value (false).
+ * @param offsetAfterListRecordKeyLen
+ * The offset of just after the key length in the list record. Or, 0 when single row.
+ * @param readPos
+ * Holds mutable read position for thread safety.
+ */
+ public void set(BytesBytesMultiHashMap hashMap, long firstOffset, boolean hasList,
+ long offsetAfterListRecordKeyLen, WriteBuffers.Position readPos) {
+
+ this.hashMap = hashMap;
+ this.readPos = readPos;
+
+ this.firstOffset = firstOffset;
+ this.hasList = hasList;
+ this.offsetAfterListRecordKeyLen = offsetAfterListRecordKeyLen;
+
+ // Position at first row.
+ readIndex = 0;
+ nextTailOffset = -1;
+
+ hasRows = true;
+ }
+
+ public WriteBuffers.ByteSegmentRef first() {
+ if (!hasRows) {
+ return null;
+ }
+
+ // Position at first row.
+ readIndex = 0;
+ nextTailOffset = -1;
+
+ return internalRead();
+ }
+
+ public WriteBuffers.ByteSegmentRef next() {
+ if (!hasRows) {
+ return null;
+ }
+
+ return internalRead();
+ }
+
+ /**
+ * Read the current value.
+ *
+ * @return
+ * The ByteSegmentRef to the current value read.
+ */
+ private WriteBuffers.ByteSegmentRef internalRead() {
+
+ if (!hasList) {
+
+ /*
+ * Single value.
+ */
+
+ if (readIndex > 0) {
+ return null;
+ }
+
+ // For a non-list (i.e. single value), the offset is for the variable length long (VLong)
+ // holding the value length (followed by the key length).
+ hashMap.writeBuffers.setReadPoint(firstOffset, readPos);
+ int valueLength = (int) hashMap.writeBuffers.readVLong(readPos);
+
+ // The value is before the offset. Make byte segment reference absolute.
+ byteSegmentRef.reset(firstOffset - valueLength, valueLength);
+ hashMap.writeBuffers.populateValue(byteSegmentRef);
+
+ readIndex++;
+ return byteSegmentRef;
+ }
+
+ /*
+ * Multiple values.
+ */
+
+ if (readIndex == 0) {
+ // For a list, the value and key lengths of 1st record were overwritten with the
+ // relative offset to a new list record.
+ long relativeOffset = hashMap.writeBuffers.readNByteLong(firstOffset, 5, readPos);
+
+ // At the beginning of the list record will be the value length.
+ hashMap.writeBuffers.setReadPoint(firstOffset + relativeOffset, readPos);
+ int valueLength = (int) hashMap.writeBuffers.readVLong(readPos);
+
+ // The value is before the list record offset. Make byte segment reference absolute.
+ byteSegmentRef.reset(firstOffset - valueLength, valueLength);
+ hashMap.writeBuffers.populateValue(byteSegmentRef);
+
+ readIndex++;
+ return byteSegmentRef;
+ }
+
+ if (readIndex == 1) {
+ // We remembered the offset of just after the key length in the list record.
+ // Read the absolute offset to the 2nd value.
+ nextTailOffset = hashMap.writeBuffers.readNByteLong(offsetAfterListRecordKeyLen, 5, readPos);
+ if (nextTailOffset <= 0) {
+ throw new Error("Expecting a second value");
+ }
+ } else if (nextTailOffset <= 0) {
+ return null;
+ }
+
+ hashMap.writeBuffers.setReadPoint(nextTailOffset, readPos);
+
+ // Get the value length.
+ int valueLength = (int) hashMap.writeBuffers.readVLong(readPos);
+
+ // Now read the relative offset to next record. Next record is always before the
+ // previous record in the write buffers (see writeBuffers javadoc).
+ long delta = hashMap.writeBuffers.readVLong(readPos);
+ long newTailOffset = delta == 0 ? 0 : (nextTailOffset - delta);
+
+ // The value is before the value record offset. Make byte segment reference absolute.
+ byteSegmentRef.reset(nextTailOffset - valueLength, valueLength);
+ hashMap.writeBuffers.populateValue(byteSegmentRef);
+
+ nextTailOffset = newTailOffset;
+ readIndex++;
+ return byteSegmentRef;
+ }
+
+ /**
+ * @return Whether we have read all the values or not.
+ */
+ public boolean isEof() {
+ // LOG.info("BytesBytesMultiHashMap isEof hasRows " + hasRows + " hasList " + hasList + " readIndex " + readIndex + " nextTailOffset " + nextTailOffset);
+ if (!hasRows) {
+ return true;
+ }
+
+ if (!hasList) {
+ return (readIndex > 0);
+ } else {
+ // Multiple values.
+ if (readIndex <= 1) {
+ // Careful: We have not read the list record and 2nd value yet, so nextTailOffset
+ // is not valid yet.
+ return false;
+ } else {
+ return (nextTailOffset <= 0);
+ }
+ }
+ }
+
+ /**
+ * Lets go of any references to a hash map.
+ */
+ public void forget() {
+ hashMap = null;
+ readPos = null;
+ byteSegmentRef.reset(0, 0);
+ hasRows = false;
+ readIndex = 0;
+ nextTailOffset = -1;
+ }
+ }
+
/** The source of keys and values to put into hashtable; avoids byte copying. */
public static interface KvSource {
/** Write key into output. */
@@ -264,55 +486,41 @@ public final class BytesBytesMultiHashMa
}
/** Not thread-safe! Use createGetterForThread. */
- public byte getValueRefs(byte[] key, int length, List<WriteBuffers.ByteSegmentRef> result) {
- return getValueRefs(key, length, result, writeBuffers.getReadPosition());
+ public byte getValueResult(byte[] key, int offset, int length, Result hashMapResult) {
+ return getValueResult(key, offset, length, hashMapResult, writeBuffers.getReadPosition());
}
/**
- * Gets "lazy" values for a key (as a set of byte segments in underlying buffer).
+ * Finds a key. Values can be read with the supplied result object.
+ *
* @param key Key buffer.
- * @param length Length of the key in buffer.
- * @param result The list to use to store the results.
- * @return the state byte for the key (see class description).
+ * @param offset the offset to the key in the buffer
+ * @param hashMapResult The object to fill in that can read the values.
+ * @param readPos Holds mutable read position for thread safety.
+ * @return The state byte.
*/
- private byte getValueRefs(byte[] key, int length,
- List<WriteBuffers.ByteSegmentRef> result, WriteBuffers.Position readPos) {
+ private byte getValueResult(byte[] key, int offset, int length, Result hashMapResult,
+ WriteBuffers.Position readPos) {
+
+ hashMapResult.forget();
+
// First, find first record for the key.
- result.clear();
- long ref = findKeyRefToRead(key, length, readPos);
+ long ref = findKeyRefToRead(key, offset, length, readPos);
if (ref == 0) {
return 0;
}
+
boolean hasList = Ref.hasList(ref);
// This relies on findKeyRefToRead doing key equality check and leaving read ptr where needed.
- long lrPtrOffset = hasList ? writeBuffers.getReadPoint(readPos) : 0;
+ long offsetAfterListRecordKeyLen = hasList ? writeBuffers.getReadPoint(readPos) : 0;
- writeBuffers.setReadPoint(getFirstRecordLengthsOffset(ref, readPos), readPos);
- int valueLength = (int)writeBuffers.readVLong(readPos);
- result.add(new WriteBuffers.ByteSegmentRef(Ref.getOffset(ref) - valueLength, valueLength));
- byte stateByte = Ref.getStateByte(ref);
- if (!hasList) {
- return stateByte;
- }
+ hashMapResult.set(this, Ref.getOffset(ref), hasList, offsetAfterListRecordKeyLen,
+ readPos);
- // There're multiple records for the key; get the offset of the next one.
- long nextTailOffset = writeBuffers.readNByteLong(lrPtrOffset, 5, readPos);
- // LOG.info("Next tail offset " + nextTailOffset);
-
- while (nextTailOffset > 0) {
- writeBuffers.setReadPoint(nextTailOffset, readPos);
- valueLength = (int)writeBuffers.readVLong(readPos);
- result.add(new WriteBuffers.ByteSegmentRef(nextTailOffset - valueLength, valueLength));
- // Now read the relative offset to next record. Next record is always before the
- // previous record in the write buffers (see writeBuffers javadoc).
- long delta = writeBuffers.readVLong(readPos);
- nextTailOffset = delta == 0 ? 0 : (nextTailOffset - delta);
- }
- return stateByte;
+ return Ref.getStateByte(ref);
}
-
/**
* Take the segment reference from {@link #getValueRefs(byte[], int, List)}
* result and makes it self-contained - adds byte array where the value is stored, and
@@ -418,9 +626,10 @@ public final class BytesBytesMultiHashMa
* @param length Read key length.
* @return The ref to use for reading.
*/
- private long findKeyRefToRead(byte[] key, int length, WriteBuffers.Position readPos) {
+ private long findKeyRefToRead(byte[] key, int offset, int length,
+ WriteBuffers.Position readPos) {
final int bucketMask = (refs.length - 1);
- int hashCode = writeBuffers.hashCode(key, 0, length);
+ int hashCode = writeBuffers.hashCode(key, offset, length);
int slot = hashCode & bucketMask;
// LOG.info("Read hash code for " + Utils.toStringBinary(key, 0, length)
// + " is " + Integer.toBinaryString(hashCode) + " - " + slot);
@@ -432,7 +641,7 @@ public final class BytesBytesMultiHashMa
if (ref == 0) {
return 0;
}
- if (isSameKey(key, length, ref, hashCode, readPos)) {
+ if (isSameKey(key, offset, length, ref, hashCode, readPos)) {
return ref;
}
++metricGetConflict;
@@ -502,7 +711,7 @@ public final class BytesBytesMultiHashMa
/**
* Same as {@link #isSameKey(long, int, long, int)} but for externally stored key.
*/
- private boolean isSameKey(byte[] key, int length, long ref, int hashCode,
+ private boolean isSameKey(byte[] key, int offset, int length, long ref, int hashCode,
WriteBuffers.Position readPos) {
if (!compareHashBits(ref, hashCode)) {
return false; // Hash bits don't match.
@@ -512,7 +721,11 @@ public final class BytesBytesMultiHashMa
keyLength = (int)writeBuffers.readVLong(readPos);
long keyOffset = Ref.getOffset(ref) - (valueLength + keyLength);
// See the comment in the other isSameKey
- return writeBuffers.isEqual(key, length, keyOffset, keyLength);
+ if (offset == 0) {
+ return writeBuffers.isEqual(key, length, keyOffset, keyLength);
+ } else {
+ return writeBuffers.isEqual(key, offset, length, keyOffset, keyLength);
+ }
}
private boolean compareHashBits(long ref, int hashCode) {
@@ -673,7 +886,6 @@ public final class BytesBytesMultiHashMa
public void debugDumpTable() {
StringBuilder dump = new StringBuilder(keysAssigned + " keys\n");
TreeMap<Long, Integer> byteIntervals = new TreeMap<Long, Integer>();
- List<WriteBuffers.ByteSegmentRef> results = new ArrayList<WriteBuffers.ByteSegmentRef>();
int examined = 0;
for (int slot = 0; slot < refs.length; ++slot) {
long ref = refs[slot];
@@ -696,9 +908,17 @@ public final class BytesBytesMultiHashMa
byteIntervals.put(keyOffset - 4, keyLength + 4);
writeBuffers.populateValue(fakeRef);
System.arraycopy(fakeRef.getBytes(), (int)fakeRef.getOffset(), key, 0, keyLength);
- getValueRefs(key, key.length, results);
dump.append(Utils.toStringBinary(key, 0, key.length)).append(" ref [").append(dumpRef(ref))
- .append("]: ").append(results.size()).append(" rows\n");
+ .append("]: ");
+ Result hashMapResult = new Result();
+ getValueResult(key, 0, key.length, hashMapResult);
+ List<WriteBuffers.ByteSegmentRef> results = new ArrayList<WriteBuffers.ByteSegmentRef>();
+ WriteBuffers.ByteSegmentRef byteSegmentRef = hashMapResult.first();
+ while (byteSegmentRef != null) {
+ results.add(hashMapResult.byteSegmentRef);
+ byteSegmentRef = hashMapResult.next();
+ }
+ dump.append(results.size()).append(" rows\n");
for (int i = 0; i < results.size(); ++i) {
WriteBuffers.ByteSegmentRef segment = results.get(i);
byteIntervals.put(segment.getOffset(),
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/FlatRowContainer.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/FlatRowContainer.java?rev=1672229&r1=1672228&r2=1672229&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/FlatRowContainer.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/FlatRowContainer.java Thu Apr 9 02:12:23 2015
@@ -208,6 +208,16 @@ public class FlatRowContainer extends Ab
}
@Override
+ public boolean hasRows() throws HiveException {
+ return rowCount() > 0;
+ }
+
+ @Override
+ public boolean isSingleRow() throws HiveException {
+ return rowCount() == 1;
+ }
+
+ @Override
public int rowCount() throws HiveException {
return rowLength > 0 ? (array.length / rowLength) : -rowLength; // see rowLength javadoc
}
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/HybridHashTableContainer.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/HybridHashTableContainer.java?rev=1672229&r1=1672228&r2=1672229&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/HybridHashTableContainer.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/HybridHashTableContainer.java Thu Apr 9 02:12:23 2015
@@ -20,6 +20,7 @@ package org.apache.hadoop.hive.ql.exec.p
import com.esotericsoftware.kryo.Kryo;
+
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
@@ -566,7 +567,7 @@ public class HybridHashTableContainer im
@Override
public MapJoinRowContainer getCurrentRows() {
- return currentValue.isEmpty() ? null : currentValue;
+ return !currentValue.hasRows() ? null : currentValue;
}
@Override
@@ -579,8 +580,8 @@ public class HybridHashTableContainer im
private class ReusableRowContainer
implements MapJoinRowContainer, AbstractRowContainer.RowIterator<List<Object>> {
private byte aliasFilter;
- private List<WriteBuffers.ByteSegmentRef> refs;
- private int currentRow;
+ private BytesBytesMultiHashMap.Result hashMapResult;
+
/**
* Sometimes, when container is empty in multi-table mapjoin, we need to add a dummy row.
* This container does not normally support adding rows; this is for the dummy row.
@@ -600,6 +601,7 @@ public class HybridHashTableContainer im
valueStruct = null; // No rows?
}
uselessIndirection = new ByteArrayRef();
+ hashMapResult = new BytesBytesMultiHashMap.Result();
clearRows();
}
@@ -611,57 +613,58 @@ public class HybridHashTableContainer im
* the evaluation for this big table row will be postponed.
*/
public JoinUtil.JoinResult setFromOutput(Output output) throws HiveException {
- if (refs == null) {
- refs = new ArrayList<WriteBuffers.ByteSegmentRef>(0);
- }
-
int keyHash = WriteBuffers.murmurHash(output.getData(), 0, output.getLength());
partitionId = keyHash & (hashPartitions.length - 1);
// If the target hash table is on disk, spill this row to disk as well to be processed later
if (isOnDisk(partitionId)) {
toSpillPartitionId = partitionId;
- refs.clear();
+ hashMapResult.forget();
return JoinUtil.JoinResult.SPILL;
}
else {
- byte aliasFilter = hashPartitions[partitionId].hashMap.getValueRefs(
- output.getData(), output.getLength(), refs);
- this.aliasFilter = refs.isEmpty() ? (byte) 0xff : aliasFilter;
- this.dummyRow = null;
- if (refs.isEmpty()) {
- return JoinUtil.JoinResult.NOMATCH;
- }
- else {
+ aliasFilter = hashPartitions[partitionId].hashMap.getValueResult(output.getData(), 0, output.getLength(), hashMapResult);
+ dummyRow = null;
+ if (hashMapResult.hasRows()) {
return JoinUtil.JoinResult.MATCH;
+ } else {
+ aliasFilter = (byte) 0xff;
+ return JoinUtil.JoinResult.NOMATCH;
}
}
}
- public boolean isEmpty() {
- return refs.isEmpty() && (dummyRow == null);
+ @Override
+ public boolean hasRows() {
+ return hashMapResult.hasRows() || (dummyRow != null);
+ }
+
+ @Override
+ public boolean isSingleRow() {
+ if (!hashMapResult.hasRows()) {
+ return (dummyRow != null);
+ }
+ return hashMapResult.isSingleRow();
}
// Implementation of row container
@Override
- public RowIterator<List<Object>> rowIter() throws HiveException {
- currentRow = -1;
+ public AbstractRowContainer.RowIterator<List<Object>> rowIter() throws HiveException {
return this;
}
@Override
public int rowCount() throws HiveException {
- return dummyRow != null ? 1 : refs.size();
+ // For performance reasons we do not want to chase the values to the end to determine
+ // the count. Use hasRows and isSingleRow instead.
+ throw new UnsupportedOperationException("Getting the row count not supported");
}
@Override
public void clearRows() {
// Doesn't clear underlying hashtable
- if (refs != null) {
- refs.clear();
- }
+ hashMapResult.forget();
dummyRow = null;
- currentRow = -1;
aliasFilter = (byte) 0xff;
}
@@ -678,36 +681,47 @@ public class HybridHashTableContainer im
// Implementation of row iterator
@Override
public List<Object> first() throws HiveException {
- currentRow = 0;
- return next();
- }
-
- @Override
- public List<Object> next() throws HiveException {
+ // A little strange that we forget the dummy row on read.
if (dummyRow != null) {
List<Object> result = dummyRow;
dummyRow = null;
return result;
}
- if (currentRow < 0 || refs.size() < currentRow) throw new HiveException("No rows");
- if (refs.size() == currentRow) return null;
- WriteBuffers.ByteSegmentRef ref = refs.get(currentRow++);
+
+ WriteBuffers.ByteSegmentRef byteSegmentRef = hashMapResult.first();
+ if (byteSegmentRef == null) {
+ return null;
+ } else {
+ return uppack(byteSegmentRef);
+ }
+
+ }
+
+ @Override
+ public List<Object> next() throws HiveException {
+
+ WriteBuffers.ByteSegmentRef byteSegmentRef = hashMapResult.next();
+ if (byteSegmentRef == null) {
+ return null;
+ } else {
+ return uppack(byteSegmentRef);
+ }
+
+ }
+
+ private List<Object> uppack(WriteBuffers.ByteSegmentRef ref) throws HiveException {
if (ref.getLength() == 0) {
return EMPTY_LIST; // shortcut, 0 length means no fields
}
- if (ref.getBytes() == null) {
- // partitionId is derived from previously calculated value in setFromOutput()
- hashPartitions[partitionId].hashMap.populateValue(ref);
- }
uselessIndirection.setData(ref.getBytes());
valueStruct.init(uselessIndirection, (int)ref.getOffset(), ref.getLength());
- return valueStruct.getFieldsAsList();
+ return valueStruct.getFieldsAsList(); // TODO: should we unset bytes after that?
}
@Override
public void addRow(List<Object> t) {
- if (dummyRow != null || !refs.isEmpty()) {
+ if (dummyRow != null || hashMapResult.hasRows()) {
throw new RuntimeException("Cannot add rows when not empty");
}
dummyRow = t;
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/MapJoinBytesTableContainer.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/MapJoinBytesTableContainer.java?rev=1672229&r1=1672228&r2=1672229&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/MapJoinBytesTableContainer.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/MapJoinBytesTableContainer.java Thu Apr 9 02:12:23 2015
@@ -417,7 +417,7 @@ public class MapJoinBytesTableContainer
@Override
public MapJoinRowContainer getCurrentRows() {
- return currentValue.isEmpty() ? null : currentValue;
+ return !currentValue.hasRows() ? null : currentValue;
}
@Override
@@ -430,10 +430,11 @@ public class MapJoinBytesTableContainer
private class ReusableRowContainer
implements MapJoinRowContainer, AbstractRowContainer.RowIterator<List<Object>> {
private byte aliasFilter;
+
/** Hash table wrapper specific to the container. */
private final BytesBytesMultiHashMap.ThreadSafeGetter threadSafeHashMapGetter;
- private List<WriteBuffers.ByteSegmentRef> refs;
- private int currentRow;
+ private BytesBytesMultiHashMap.Result hashMapResult;
+
/**
* Sometimes, when container is empty in multi-table mapjoin, we need to add a dummy row.
* This container does not normally support adding rows; this is for the dummy row.
@@ -452,48 +453,55 @@ public class MapJoinBytesTableContainer
}
uselessIndirection = new ByteArrayRef();
threadSafeHashMapGetter = hashMap.createGetterForThread();
+ hashMapResult = new BytesBytesMultiHashMap.Result();
clearRows();
}
public JoinUtil.JoinResult setFromOutput(Output output) {
- if (refs == null) {
- refs = new ArrayList<WriteBuffers.ByteSegmentRef>();
- }
- byte aliasFilter = threadSafeHashMapGetter.getValueRefs(
- output.getData(), output.getLength(), refs);
- this.aliasFilter = refs.isEmpty() ? (byte) 0xff : aliasFilter;
- this.dummyRow = null;
- if (refs.isEmpty()) {
- return JoinUtil.JoinResult.NOMATCH;
- } else {
+
+ aliasFilter = threadSafeHashMapGetter.getValueResult(
+ output.getData(), 0, output.getLength(), hashMapResult);
+ dummyRow = null;
+ if (hashMapResult.hasRows()) {
return JoinUtil.JoinResult.MATCH;
+ } else {
+ aliasFilter = (byte) 0xff;
+ return JoinUtil.JoinResult.NOMATCH;
}
+
+ }
+
+ @Override
+ public boolean hasRows() {
+ return hashMapResult.hasRows() || (dummyRow != null);
}
- public boolean isEmpty() {
- return refs.isEmpty() && (dummyRow == null);
+ @Override
+ public boolean isSingleRow() {
+ if (!hashMapResult.hasRows()) {
+ return (dummyRow != null);
+ }
+ return hashMapResult.isSingleRow();
}
// Implementation of row container
@Override
public AbstractRowContainer.RowIterator<List<Object>> rowIter() throws HiveException {
- currentRow = -1;
return this;
}
@Override
public int rowCount() throws HiveException {
- return dummyRow != null ? 1 : refs.size();
+ // For performance reasons we do not want to chase the values to the end to determine
+ // the count. Use hasRows and isSingleRow instead.
+ throw new UnsupportedOperationException("Getting the row count not supported");
}
@Override
public void clearRows() {
// Doesn't clear underlying hashtable
- if (refs != null) {
- refs.clear();
- }
+ hashMapResult.forget();
dummyRow = null;
- currentRow = -1;
aliasFilter = (byte) 0xff;
}
@@ -510,30 +518,39 @@ public class MapJoinBytesTableContainer
// Implementation of row iterator
@Override
public List<Object> first() throws HiveException {
- currentRow = 0;
- return nextInternal();
- }
- @Override
- public List<Object> next() throws HiveException {
- return nextInternal();
- }
-
- private List<Object> nextInternal() throws HiveException {
+ // A little strange that we forget the dummy row on read.
if (dummyRow != null) {
List<Object> result = dummyRow;
dummyRow = null;
return result;
}
- if (currentRow < 0 || refs.size() < currentRow) throw new HiveException("No rows");
- if (refs.size() == currentRow) return null;
- WriteBuffers.ByteSegmentRef ref = refs.get(currentRow++);
+
+ WriteBuffers.ByteSegmentRef byteSegmentRef = hashMapResult.first();
+ if (byteSegmentRef == null) {
+ return null;
+ } else {
+ return uppack(byteSegmentRef);
+ }
+
+ }
+
+ @Override
+ public List<Object> next() throws HiveException {
+
+ WriteBuffers.ByteSegmentRef byteSegmentRef = hashMapResult.next();
+ if (byteSegmentRef == null) {
+ return null;
+ } else {
+ return uppack(byteSegmentRef);
+ }
+
+ }
+
+ private List<Object> uppack(WriteBuffers.ByteSegmentRef ref) throws HiveException {
if (ref.getLength() == 0) {
return EMPTY_LIST; // shortcut, 0 length means no fields
}
- if (ref.getBytes() == null) {
- threadSafeHashMapGetter.populateValue(ref);
- }
uselessIndirection.setData(ref.getBytes());
valueStruct.init(uselessIndirection, (int)ref.getOffset(), ref.getLength());
return valueStruct.getFieldsAsList(); // TODO: should we unset bytes after that?
@@ -541,7 +558,7 @@ public class MapJoinBytesTableContainer
@Override
public void addRow(List<Object> t) {
- if (dummyRow != null || !refs.isEmpty()) {
+ if (dummyRow != null || hashMapResult.hasRows()) {
throw new RuntimeException("Cannot add rows when not empty");
}
dummyRow = t;
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/MapJoinEagerRowContainer.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/MapJoinEagerRowContainer.java?rev=1672229&r1=1672228&r2=1672229&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/MapJoinEagerRowContainer.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/MapJoinEagerRowContainer.java Thu Apr 9 02:12:23 2015
@@ -81,6 +81,16 @@ public class MapJoinEagerRowContainer
return null;
}
+ @Override
+ public boolean hasRows() {
+ return list.size() > 0;
+ }
+
+ @Override
+ public boolean isSingleRow() {
+ return list.size() == 1;
+ }
+
/**
* Get the number of elements in the RowContainer.
*
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/RowContainer.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/RowContainer.java?rev=1672229&r1=1672228&r2=1672229&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/RowContainer.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/RowContainer.java Thu Apr 9 02:12:23 2015
@@ -331,6 +331,17 @@ public class RowContainer<ROW extends Li
}
}
+
+ @Override
+ public boolean hasRows() {
+ return size > 0;
+ }
+
+ @Override
+ public boolean isSingleRow() {
+ return size == 1;
+ }
+
/**
* Get the number of elements in the RowContainer.
*
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/UnwrapRowContainer.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/UnwrapRowContainer.java?rev=1672229&r1=1672228&r2=1672229&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/UnwrapRowContainer.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/UnwrapRowContainer.java Thu Apr 9 02:12:23 2015
@@ -95,6 +95,17 @@ public class UnwrapRowContainer
internal.addRow(t);
}
+
+ @Override
+ public boolean hasRows() throws HiveException {
+ return internal.hasRows();
+ }
+
+ @Override
+ public boolean isSingleRow() throws HiveException {
+ return internal.isSingleRow();
+ }
+
@Override
public int rowCount() throws HiveException {
return internal.rowCount();
Modified: hive/trunk/ql/src/test/org/apache/hadoop/hive/ql/exec/persistence/TestBytesBytesMultiHashMap.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/test/org/apache/hadoop/hive/ql/exec/persistence/TestBytesBytesMultiHashMap.java?rev=1672229&r1=1672228&r2=1672229&view=diff
==============================================================================
--- hive/trunk/ql/src/test/org/apache/hadoop/hive/ql/exec/persistence/TestBytesBytesMultiHashMap.java (original)
+++ hive/trunk/ql/src/test/org/apache/hadoop/hive/ql/exec/persistence/TestBytesBytesMultiHashMap.java Thu Apr 9 02:12:23 2015
@@ -50,10 +50,10 @@ public class TestBytesBytesMultiHashMap
BytesBytesMultiHashMap map = new BytesBytesMultiHashMap(CAPACITY, LOAD_FACTOR, WB_SIZE);
RandomKvSource kv = new RandomKvSource(0, 0);
map.put(kv, -1);
- verifyResults(map, kv.getLastKey(), kv.getLastValue());
+ verifyHashMapResult(map, kv.getLastKey(), kv.getLastValue());
kv = new RandomKvSource(10, 100);
map.put(kv, -1);
- verifyResults(map, kv.getLastKey(), kv.getLastValue());
+ verifyHashMapResult(map, kv.getLastKey(), kv.getLastValue());
}
@Test
@@ -61,12 +61,12 @@ public class TestBytesBytesMultiHashMap
BytesBytesMultiHashMap map = new BytesBytesMultiHashMap(CAPACITY, LOAD_FACTOR, WB_SIZE);
RandomKvSource kv = new RandomKvSource(0, 100);
map.put(kv, -1);
- verifyResults(map, kv.getLastKey(), kv.getLastValue());
+ verifyHashMapResult(map, kv.getLastKey(), kv.getLastValue());
FixedKeyKvSource kv2 = new FixedKeyKvSource(kv.getLastKey(), 0, 100);
kv2.values.add(kv.getLastValue());
for (int i = 0; i < 3; ++i) {
map.put(kv2, -1);
- verifyResults(map, kv2.key, kv2.values.toArray(new byte[kv2.values.size()][]));
+ verifyHashMapResult(map, kv2.key, kv2.values.toArray(new byte[kv2.values.size()][]));
}
}
@@ -80,11 +80,11 @@ public class TestBytesBytesMultiHashMap
FixedKeyKvSource kv2 = new FixedKeyKvSource(kv.getLastKey(), 0, 100);
map.put(kv2, -1);
key[0] = (byte)(key[0] + 1);
- List<WriteBuffers.ByteSegmentRef> results = new ArrayList<WriteBuffers.ByteSegmentRef>(0);
- map.getValueRefs(key, key.length, results);
- assertTrue(results.isEmpty());
- map.getValueRefs(key, 0, results);
- assertTrue(results.isEmpty());
+ BytesBytesMultiHashMap.Result hashMapResult = new BytesBytesMultiHashMap.Result();
+ map.getValueResult(key, 0, key.length, hashMapResult);
+ assertTrue(!hashMapResult.hasRows());
+ map.getValueResult(key, 0, 0, hashMapResult);
+ assertTrue(!hashMapResult.hasRows());
}
@Test
@@ -96,13 +96,12 @@ public class TestBytesBytesMultiHashMap
map.put(kv, -1);
}
for (int i = 0; i < kv.keys.size(); ++i) {
- verifyResults(map, kv.keys.get(i), kv.values.get(i));
+ verifyHashMapResult(map, kv.keys.get(i), kv.values.get(i));
}
assertEquals(CAPACITY, map.getCapacity());
// Get of non-existent key should terminate..
- List<WriteBuffers.ByteSegmentRef> results = new ArrayList<WriteBuffers.ByteSegmentRef>(0);
- map.getValueRefs(new byte[0], 0, results);
- assertTrue(results.isEmpty());
+ BytesBytesMultiHashMap.Result hashMapResult = new BytesBytesMultiHashMap.Result();
+ map.getValueResult(new byte[0], 0, 0, hashMapResult);
}
@Test
@@ -113,23 +112,29 @@ public class TestBytesBytesMultiHashMap
for (int i = 0; i < 18; ++i) {
map.put(kv, -1);
for (int j = 0; j <= i; ++j) {
- verifyResults(map, kv.keys.get(j), kv.values.get(j));
+ verifyHashMapResult(map, kv.keys.get(j), kv.values.get(j));
}
}
assertEquals(1 << 18, map.getCapacity());
}
- private void verifyResults(BytesBytesMultiHashMap map, byte[] key, byte[]... values) {
- List<WriteBuffers.ByteSegmentRef> results = new ArrayList<WriteBuffers.ByteSegmentRef>(0);
- byte state = map.getValueRefs(key, key.length, results);
- assertEquals(state, results.size());
- assertEquals(values.length, results.size());
+ private void verifyHashMapResult(BytesBytesMultiHashMap map, byte[] key, byte[]... values) {
+ BytesBytesMultiHashMap.Result hashMapResult = new BytesBytesMultiHashMap.Result();
+ byte state = map.getValueResult(key, 0, key.length, hashMapResult);
HashSet<ByteBuffer> hs = new HashSet<ByteBuffer>();
- for (int i = 0; i < results.size(); ++i) {
- WriteBuffers.ByteSegmentRef result = results.get(i);
- map.populateValue(result);
- hs.add(result.copy());
+ int count = 0;
+ if (hashMapResult.hasRows()) {
+ WriteBuffers.ByteSegmentRef ref = hashMapResult.first();
+ while (ref != null) {
+ count++;
+ hs.add(ref.copy());
+ ref = hashMapResult.next();
+ }
+ } else {
+ assertTrue(hashMapResult.isEof());
}
+ assertEquals(state, count);
+ assertEquals(values.length, count);
for (int i = 0; i < values.length; ++i) {
assertTrue(hs.contains(ByteBuffer.wrap(values[i])));
}
Modified: hive/trunk/serde/src/java/org/apache/hadoop/hive/serde2/ByteStream.java
URL: http://svn.apache.org/viewvc/hive/trunk/serde/src/java/org/apache/hadoop/hive/serde2/ByteStream.java?rev=1672229&r1=1672228&r2=1672229&view=diff
==============================================================================
--- hive/trunk/serde/src/java/org/apache/hadoop/hive/serde2/ByteStream.java (original)
+++ hive/trunk/serde/src/java/org/apache/hadoop/hive/serde2/ByteStream.java Thu Apr 9 02:12:23 2015
@@ -19,11 +19,13 @@
package org.apache.hadoop.hive.serde2;
import java.io.IOException;
+import java.util.Arrays;
import org.apache.hadoop.hive.common.io.NonSyncByteArrayInputStream;
import org.apache.hadoop.hive.common.io.NonSyncByteArrayOutputStream;
import org.apache.hadoop.hive.serde2.binarysortable.BinarySortableSerDe;
import org.apache.hadoop.hive.serde2.objectinspector.PrimitiveObjectInspector.PrimitiveCategory;
+import org.apache.hadoop.hive.serde2.ByteStream.Output;
/**
* Extensions to bytearrayinput/output streams.
@@ -90,14 +92,32 @@ public class ByteStream {
}
@Override
+ public void writeByte(long offset, byte value) {
+ getData()[(int) offset] = value;
+ }
+
+ @Override
public void reserve(int byteCount) {
for (int i = 0; i < byteCount; ++i) {
write(0);
}
}
+
+ public boolean arraysEquals(Output output) {
+ if (count != output.count) {
+ return false;
+ }
+ for (int i = 0; i < count; i++) {
+ if (buf[i] != output.buf[i]) {
+ return false;
+ }
+ }
+ return true;
+ }
}
public static interface RandomAccessOutput {
+ public void writeByte(long offset, byte value);
public void writeInt(long offset, int value);
public void reserve(int byteCount);
public void write(int b);
Modified: hive/trunk/serde/src/java/org/apache/hadoop/hive/serde2/WriteBuffers.java
URL: http://svn.apache.org/viewvc/hive/trunk/serde/src/java/org/apache/hadoop/hive/serde2/WriteBuffers.java?rev=1672229&r1=1672228&r2=1672229&view=diff
==============================================================================
--- hive/trunk/serde/src/java/org/apache/hadoop/hive/serde2/WriteBuffers.java (original)
+++ hive/trunk/serde/src/java/org/apache/hadoop/hive/serde2/WriteBuffers.java Thu Apr 9 02:12:23 2015
@@ -303,6 +303,38 @@ public final class WriteBuffers implemen
return true;
}
+ /**
+ * Compares part of the buffer with a part of an external byte array.
+ * Does not modify readPoint.
+ */
+ public boolean isEqual(byte[] left, int leftOffset, int leftLength, long rightOffset, int rightLength) {
+ if (rightLength != leftLength) {
+ return false;
+ }
+ int rightIndex = getBufferIndex(rightOffset), rightFrom = getOffset(rightOffset);
+ byte[] rightBuffer = writeBuffers.get(rightIndex);
+ if (rightFrom + rightLength <= wbSize) {
+ // TODO: allow using unsafe optionally.
+ for (int i = 0; i < leftLength; ++i) {
+ if (left[leftOffset + i] != rightBuffer[rightFrom + i]) {
+ return false;
+ }
+ }
+ return true;
+ }
+ for (int i = 0; i < rightLength; ++i) {
+ if (rightFrom == wbSize) {
+ ++rightIndex;
+ rightBuffer = writeBuffers.get(rightIndex);
+ rightFrom = 0;
+ }
+ if (left[leftOffset + i] != rightBuffer[rightFrom++]) {
+ return false;
+ }
+ }
+ return true;
+ }
+
public void clear() {
writeBuffers.clear();
clearState();
@@ -520,6 +552,19 @@ public final class WriteBuffers implemen
writePos.bufferIndex = prevIndex;
writePos.buffer = writeBuffers.get(writePos.bufferIndex);
writePos.offset = prevOffset;
+ }
+
+
+ @Override
+ public void writeByte(long offset, byte value) {
+ int prevIndex = writePos.bufferIndex, prevOffset = writePos.offset;
+ setWritePoint(offset);
+ // One byte is always available for writing.
+ writePos.buffer[writePos.offset] = value;
+
+ writePos.bufferIndex = prevIndex;
+ writePos.buffer = writeBuffers.get(writePos.bufferIndex);
+ writePos.offset = prevOffset;
}
// Lifted from org.apache.hadoop.util.hash.MurmurHash... but supports offset.