svn commit: r1594713 [2/3] - in /hive/trunk: common/src/java/org/apache/hadoop/hive/conf/ conf/ hbase-handler/src/java/org/apache/hadoop/hive/hbase/ hbase-handler/src/test/org/apache/hadoop/hive/hbase/ ql/src/java/org/apache/hadoop/hive/ql/ ql/src/java...

[se...@apache.org]

Added: 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=1594713&view=auto
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/MapJoinBytesTableContainer.java (added)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/MapJoinBytesTableContainer.java Wed May 14 20:53:30 2014
@@ -0,0 +1,495 @@
+package org.apache.hadoop.hive.ql.exec.persistence;
+
+
+import java.io.ObjectOutputStream;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.debug.Utils;
+import org.apache.hadoop.hive.ql.exec.ExprNodeEvaluator;
+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;
+import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.serde2.SerDe;
+import org.apache.hadoop.hive.serde2.SerDeException;
+import org.apache.hadoop.hive.serde2.ByteStream.Output;
+import org.apache.hadoop.hive.serde2.ByteStream.RandomAccessOutput;
+import org.apache.hadoop.hive.serde2.WriteBuffers;
+import org.apache.hadoop.hive.serde2.binarysortable.BinarySortableSerDe;
+import org.apache.hadoop.hive.serde2.io.ShortWritable;
+import org.apache.hadoop.hive.serde2.lazy.ByteArrayRef;
+import org.apache.hadoop.hive.serde2.lazybinary.LazyBinaryFactory;
+import org.apache.hadoop.hive.serde2.lazybinary.LazyBinarySerDe;
+import org.apache.hadoop.hive.serde2.lazybinary.LazyBinaryStruct;
+import org.apache.hadoop.hive.serde2.lazybinary.LazyBinaryUtils;
+import org.apache.hadoop.hive.serde2.lazybinary.objectinspector.LazyBinaryObjectInspectorFactory;
+import org.apache.hadoop.hive.serde2.lazybinary.objectinspector.LazyBinaryStructObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.StructField;
+import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.primitive.ShortObjectInspector;
+import org.apache.hadoop.hive.serde2.typeinfo.TypeInfo;
+import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoUtils;
+import org.apache.hadoop.io.BinaryComparable;
+import org.apache.hadoop.io.Writable;
+
+/**
+ * Table container that serializes keys and values using LazyBinarySerDe into
+ * BytesBytesMultiHashMap, with very low memory overhead. However,
+ * there may be some perf overhead when retrieving rows.
+ */
+public class MapJoinBytesTableContainer implements MapJoinTableContainer {
+  private static final Log LOG = LogFactory.getLog(MapJoinTableContainer.class);
+
+  private final BytesBytesMultiHashMap hashMap;
+  /** The OI used to deserialize values. We never deserialize keys. */
+  private LazyBinaryStructObjectInspector internalValueOi;
+  /**
+   * This is kind of hacky. Currently we get BinarySortableSerDe-serialized keys; we could
+   * re-serialize them into LazyBinarySerDe, but instead we just reuse the bytes. However, to
+   * compare the large table keys correctly when we do, we need to serialize them with correct
+   * ordering. Hence, remember the ordering here; it is null if we do use LazyBinarySerDe.
+   */
+  private boolean[] sortableSortOrders;
+  private KeyValueHelper writeHelper;
+
+  private List<Object> EMPTY_LIST = new ArrayList<Object>(0);
+
+  public MapJoinBytesTableContainer(Configuration hconf, MapJoinObjectSerDeContext valCtx)
+      throws SerDeException {
+    this(HiveConf.getIntVar(hconf, HiveConf.ConfVars.HIVEHASHTABLETHRESHOLD),
+        HiveConf.getFloatVar(hconf, HiveConf.ConfVars.HIVEHASHTABLELOADFACTOR),
+        HiveConf.getIntVar(hconf, HiveConf.ConfVars.HIVEHASHTABLEWBSIZE), valCtx);
+  }
+
+  private MapJoinBytesTableContainer(int threshold, float loadFactor, int wbSize,
+      MapJoinObjectSerDeContext valCtx) throws SerDeException {
+    hashMap = new BytesBytesMultiHashMap(threshold, loadFactor, wbSize);
+  }
+
+  private LazyBinaryStructObjectInspector createInternalOi(
+      MapJoinObjectSerDeContext valCtx) throws SerDeException {
+    // We are going to use LBSerDe to serialize values; create OI for retrieval.
+    List<? extends StructField> fields = ((StructObjectInspector)
+        valCtx.getSerDe().getObjectInspector()).getAllStructFieldRefs();
+    List<String> colNames = new ArrayList<String>(fields.size());
+    List<ObjectInspector> colOis = new ArrayList<ObjectInspector>(fields.size());
+    for (int i = 0; i < fields.size(); ++i) {
+      StructField field = fields.get(i);
+      colNames.add(field.getFieldName());
+      // It would be nice if OI could return typeInfo...
+      TypeInfo typeInfo = TypeInfoUtils.getTypeInfoFromTypeString(
+          field.getFieldObjectInspector().getTypeName());
+      colOis.add(LazyBinaryUtils.getLazyBinaryObjectInspectorFromTypeInfo(typeInfo));
+    }
+
+    return LazyBinaryObjectInspectorFactory
+        .getLazyBinaryStructObjectInspector(colNames, colOis);
+  }
+
+  private static interface KeyValueHelper extends BytesBytesMultiHashMap.KvSource {
+    void setKeyValue(Writable key, Writable val) throws SerDeException;
+  }
+
+  private static class KeyValueWriter implements KeyValueHelper {
+    private final SerDe keySerDe, valSerDe;
+    private final StructObjectInspector keySoi, valSoi;
+    private final List<ObjectInspector> keyOis, valOis;
+    private final Object[] keyObjs, valObjs;
+    private final boolean hasFilterTag;
+
+    public KeyValueWriter(
+        SerDe keySerDe, SerDe valSerDe, boolean hasFilterTag) throws SerDeException {
+      this.keySerDe = keySerDe;
+      this.valSerDe = valSerDe;
+      keySoi = (StructObjectInspector)keySerDe.getObjectInspector();
+      valSoi = (StructObjectInspector)valSerDe.getObjectInspector();
+      List<? extends StructField> keyFields = keySoi.getAllStructFieldRefs(),
+          valFields = valSoi.getAllStructFieldRefs();
+      keyOis = new ArrayList<ObjectInspector>(keyFields.size());
+      valOis = new ArrayList<ObjectInspector>(valFields.size());
+      for (int i = 0; i < keyFields.size(); ++i) {
+        keyOis.add(keyFields.get(i).getFieldObjectInspector());
+      }
+      for (int i = 0; i < valFields.size(); ++i) {
+        valOis.add(valFields.get(i).getFieldObjectInspector());
+      }
+      keyObjs = new Object[keyOis.size()];
+      valObjs = new Object[valOis.size()];
+      this.hasFilterTag = hasFilterTag;
+    }
+
+    @Override
+    public void writeKey(RandomAccessOutput dest) throws SerDeException {
+      LazyBinarySerDe.serializeStruct(dest, keyObjs, keyOis);
+    }
+
+    @Override
+    public void writeValue(RandomAccessOutput dest) throws SerDeException {
+      LazyBinarySerDe.serializeStruct(dest, valObjs, valOis);
+    }
+
+    @Override
+    public void setKeyValue(Writable key, Writable val) throws SerDeException {
+      Object keyObj = keySerDe.deserialize(key), valObj = valSerDe.deserialize(val);
+      List<? extends StructField> keyFields = keySoi.getAllStructFieldRefs(),
+          valFields = valSoi.getAllStructFieldRefs();
+      for (int i = 0; i < keyFields.size(); ++i) {
+        keyObjs[i] = keySoi.getStructFieldData(keyObj, keyFields.get(i));
+      }
+      for (int i = 0; i < valFields.size(); ++i) {
+        valObjs[i] = valSoi.getStructFieldData(valObj, valFields.get(i));
+      }
+    }
+
+    @Override
+    public byte updateStateByte(Byte previousValue) {
+      if (!hasFilterTag) return (byte)0xff;
+      byte aliasFilter = (previousValue == null) ? (byte)0xff : previousValue.byteValue();
+      aliasFilter &= ((ShortWritable)valObjs[valObjs.length - 1]).get();
+      return aliasFilter;
+    }
+  }
+
+  private static class LazyBinaryKvWriter implements KeyValueHelper {
+    private final LazyBinaryStruct.SingleFieldGetter filterGetter;
+    private Writable key, value;
+    private final SerDe keySerDe;
+    private Boolean hasTag = null; // sanity check - we should not receive keys with tags
+
+    public LazyBinaryKvWriter(SerDe keySerDe, LazyBinaryStructObjectInspector valSoi,
+        boolean hasFilterTag) throws SerDeException {
+      this.keySerDe = keySerDe;
+      if (hasFilterTag) {
+        List<? extends StructField> fields = valSoi.getAllStructFieldRefs();
+        int ix = fields.size() - 1;
+        if (!(fields.get(ix).getFieldObjectInspector() instanceof ShortObjectInspector)) {
+          throw new SerDeException("Has filter tag, but corresponding OI is " +
+              fields.get(ix).getFieldObjectInspector());
+        }
+        filterGetter = new LazyBinaryStruct.SingleFieldGetter(valSoi, fields.size() - 1);
+      } else {
+        filterGetter = null;
+      }
+    }
+
+    @Override
+    public void writeKey(RandomAccessOutput dest) throws SerDeException {
+      if (!(key instanceof BinaryComparable)) {
+        throw new SerDeException("Unexpected type " + key.getClass().getCanonicalName());
+      }
+      sanityCheckKeyForTag();
+      BinaryComparable b = (BinaryComparable)key;
+      dest.write(b.getBytes(), 0, b.getLength() - (hasTag ? 1 : 0));
+    }
+
+    /**
+     * If we received data with tags from ReduceSinkOperators, no keys will match. This should
+     * not happen, but is important enough that we want to find out and work around it if some
+     * optimized change causes RSO to pass on tags.
+     */
+    private void sanityCheckKeyForTag() throws SerDeException {
+      if (hasTag != null) return;
+      BinaryComparable b = (BinaryComparable)key;
+      Object o = keySerDe.deserialize(key);
+      StructObjectInspector soi = (StructObjectInspector)keySerDe.getObjectInspector();
+      List<? extends StructField> fields = soi.getAllStructFieldRefs();
+      Object[] data = new Object[fields.size()];
+      List<ObjectInspector> fois = new ArrayList<ObjectInspector>(fields.size());
+      for (int i = 0; i < fields.size(); i++) {
+        data[i] = soi.getStructFieldData(o, fields.get(i));
+        fois.add(fields.get(i).getFieldObjectInspector());
+      }
+      Output output = new Output();
+      BinarySortableSerDe.serializeStruct(output, data, fois, new boolean[fields.size()]);
+      hasTag = (output.getLength() != b.getLength());
+      if (hasTag) {
+        LOG.error("Tag found in keys and will be removed. This should not happen.");
+        if (output.getLength() != (b.getLength() - 1)) {
+          throw new SerDeException(
+              "Unexpected tag: " + b.getLength() + " reserialized to " + output.getLength());
+        }
+      }
+    }
+
+    @Override
+    public void writeValue(RandomAccessOutput dest) throws SerDeException {
+      if (!(value instanceof BinaryComparable)) {
+        throw new SerDeException("Unexpected type " + value.getClass().getCanonicalName());
+      }
+      BinaryComparable b = (BinaryComparable)value;
+      dest.write(b.getBytes(), 0, b.getLength());
+    }
+
+    @Override
+    public void setKeyValue(Writable key, Writable val) {
+      this.key = key;
+      this.value = val;
+    }
+
+    @Override
+    public byte updateStateByte(Byte previousValue) {
+      if (filterGetter == null) return (byte)0xff;
+      byte aliasFilter = (previousValue == null) ? (byte)0xff : previousValue.byteValue();
+      filterGetter.init((BinaryComparable)value);
+      aliasFilter &= filterGetter.getShort();
+      return aliasFilter;
+    }
+  }
+
+  @SuppressWarnings("deprecation")
+  @Override
+  public MapJoinKey putRow(MapJoinObjectSerDeContext keyContext, Writable currentKey,
+      MapJoinObjectSerDeContext valueContext, Writable currentValue) throws SerDeException {
+    SerDe keySerde = keyContext.getSerDe(), valSerde = valueContext.getSerDe();
+    if (writeHelper == null) {
+      LOG.info("Initializing container with "
+          + keySerde.getClass().getName() + " and " + valSerde.getClass().getName());
+      if (keySerde instanceof BinarySortableSerDe && valSerde instanceof LazyBinarySerDe) {
+        LazyBinaryStructObjectInspector valSoi =
+            (LazyBinaryStructObjectInspector)valSerde.getObjectInspector();
+        writeHelper = new LazyBinaryKvWriter(keySerde, valSoi, valueContext.hasFilterTag());
+        internalValueOi = valSoi;
+        sortableSortOrders = ((BinarySortableSerDe)keySerde).getSortOrders();
+      } else {
+        writeHelper = new KeyValueWriter(keySerde, valSerde, valueContext.hasFilterTag());
+        internalValueOi = createInternalOi(valueContext);
+        sortableSortOrders = null;
+      }
+    }
+    writeHelper.setKeyValue(currentKey, currentValue);
+    hashMap.put(writeHelper);
+    return null; // there's no key to return
+  }
+
+  @Override
+  public void clear() {
+    hashMap.clear();
+  }
+
+  @Override
+  public MapJoinKey getAnyKey() {
+    return null; // This table has no keys.
+  }
+
+  @Override
+  public ReusableGetAdaptor createGetter(MapJoinKey keyTypeFromLoader) {
+    if (keyTypeFromLoader != null) {
+      throw new AssertionError("No key expected from loader but got " + keyTypeFromLoader);
+    }
+    return new GetAdaptor();
+  }
+
+  @Override
+  public void seal() {
+    hashMap.seal();
+  }
+
+  /** Implementation of ReusableGetAdaptor that has Output for key serialization; row
+   * container is also created once and reused for every row. */
+  private class GetAdaptor implements ReusableGetAdaptor {
+    private ReusableRowContainer currentValue = new ReusableRowContainer();
+    private Output output;
+    private boolean[] nulls;
+
+    @Override
+    public void setFromVector(VectorHashKeyWrapper kw,
+        VectorExpressionWriter[] keyOutputWriters,
+        VectorHashKeyWrapperBatch keyWrapperBatch) throws HiveException {
+      if (nulls == null) {
+        nulls = new boolean[keyOutputWriters.length];
+      } else {
+        assert nulls.length == keyOutputWriters.length;
+      }
+      try {
+        output = MapJoinKey.serializeVector(
+            output, kw, keyOutputWriters, keyWrapperBatch, nulls, sortableSortOrders);
+      } catch (SerDeException e) {
+        throw new HiveException(e);
+      }
+      this.currentValue.setFromOutput(output);
+    }
+
+    @Override
+    public void setFromRow(Object row, List<ExprNodeEvaluator> fields,
+        List<ObjectInspector> ois) throws HiveException {
+      if (nulls == null) {
+        nulls = new boolean[fields.size()];
+      }
+      Object[] fieldObjs = new Object[fields.size()];
+      for (int keyIndex = 0; keyIndex < fields.size(); ++keyIndex) {
+        fieldObjs[keyIndex] = fields.get(keyIndex).evaluate(row);
+        nulls[keyIndex] = (fieldObjs[keyIndex] == null);
+      }
+      try {
+        output = MapJoinKey.serializeRow(output, fieldObjs, ois, sortableSortOrders);
+      } catch (SerDeException e) {
+        throw new HiveException(e);
+      }
+      this.currentValue.setFromOutput(output);
+    }
+
+
+    @Override
+    public void setFromOther(ReusableGetAdaptor other) {
+      assert other instanceof GetAdaptor;
+      GetAdaptor other2 = (GetAdaptor)other;
+      output = other2.output;
+      nulls = other2.nulls;
+      this.currentValue.setFromOutput(output);
+    }
+
+    @Override
+    public boolean hasAnyNulls(int fieldCount, boolean[] nullsafes) {
+      if (nulls == null || nulls.length == 0) return false;
+      for (int i = 0; i < nulls.length; i++) {
+        if (nulls[i] == true && (nullsafes == null || !nullsafes[i])) {
+          return true;
+        }
+      }
+      return false;
+    }
+
+    @Override
+    public MapJoinRowContainer getCurrentRows() {
+      return currentValue.isEmpty() ? null : currentValue;
+    }
+  }
+
+  /** Row container that gets and deserializes the rows on demand from bytes provided. */
+  private class ReusableRowContainer
+    implements MapJoinRowContainer, AbstractRowContainer.RowIterator<List<Object>> {
+    private byte aliasFilter;
+    private List<WriteBuffers.ByteSegmentRef> refs;
+    private int currentRow;
+    /**
+     * 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.
+     */
+    private List<Object> dummyRow = null;
+
+    private final ByteArrayRef uselessIndirection; // LBStruct needs ByteArrayRef
+    private final LazyBinaryStruct valueStruct;
+
+    public ReusableRowContainer() {
+      if (internalValueOi != null) {
+        valueStruct = (LazyBinaryStruct)
+            LazyBinaryFactory.createLazyBinaryObject(internalValueOi);
+      } else {
+        valueStruct = null; // No rows?
+      }
+      uselessIndirection = new ByteArrayRef();
+      clearRows();
+    }
+
+    public void setFromOutput(Output output) {
+      if (refs == null) {
+        refs = new ArrayList<WriteBuffers.ByteSegmentRef>(0);
+      }
+      byte aliasFilter = hashMap.getValueRefs(output.getData(), output.getLength(), refs);
+      this.aliasFilter = refs.isEmpty() ? (byte) 0xff : aliasFilter;
+      this.dummyRow = null;
+    }
+
+    public boolean isEmpty() {
+      return refs.isEmpty() && (dummyRow == null);
+    }
+
+    // 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();
+    }
+
+    @Override
+    public void clearRows() {
+      // Doesn't clear underlying hashtable
+      if (refs != null) {
+        refs.clear();
+      }
+      dummyRow = null;
+      currentRow = -1;
+      aliasFilter = (byte) 0xff;
+    }
+
+    @Override
+    public byte getAliasFilter() throws HiveException {
+      return aliasFilter;
+    }
+
+    @Override
+    public MapJoinRowContainer copy() throws HiveException {
+      return this; // Independent of hashtable and can be modified, no need to copy.
+    }
+
+    // Implementation of row iterator
+    @Override
+    public List<Object> first() throws HiveException {
+      currentRow = 0;
+      return next();
+    }
+
+
+    @Override
+    public List<Object> next() throws HiveException {
+      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++);
+      if (ref.getLength() == 0) {
+        return EMPTY_LIST; // shortcut, 0 length means no fields
+      }
+      if (ref.getBytes() == null) {
+        hashMap.populateValue(ref);
+      }
+      uselessIndirection.setData(ref.getBytes());
+      valueStruct.init(uselessIndirection, (int)ref.getOffset(), ref.getLength());
+      return valueStruct.getFieldsAsList(); // TODO: should we unset bytes after that?
+    }
+
+    public void addRow(List<Object> t) {
+      if (dummyRow != null || !refs.isEmpty()) {
+        throw new RuntimeException("Cannot add rows when not empty");
+      }
+      dummyRow = t;
+    }
+
+    // Various unsupported methods.
+    public void addRow(Object[] value) {
+      throw new RuntimeException(this.getClass().getCanonicalName() + " cannot add arrays");
+    }
+    public void write(MapJoinObjectSerDeContext valueContext, ObjectOutputStream out) {
+      throw new RuntimeException(this.getClass().getCanonicalName() + " cannot be serialized");
+    }
+  }
+
+  public static boolean isSupportedKey(ObjectInspector keyOi) {
+    List<? extends StructField> keyFields = ((StructObjectInspector)keyOi).getAllStructFieldRefs();
+    for (StructField field : keyFields) {
+      if (!MapJoinKey.isSupportedField(field.getFieldObjectInspector())) {
+        return false;
+      }
+    }
+    return true;
+  }
+
+  @Override
+  public void dumpMetrics() {
+    hashMap.debugDumpMetrics();
+  }
+}

Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/MapJoinKey.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/MapJoinKey.java?rev=1594713&r1=1594712&r2=1594713&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/MapJoinKey.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/MapJoinKey.java Wed May 14 20:53:30 2014
@@ -25,7 +25,10 @@ import java.util.Arrays;
 import java.util.HashSet;
 import java.util.List;
 
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.hive.common.type.Decimal128;
+import org.apache.hadoop.hive.ql.debug.Utils;
 import org.apache.hadoop.hive.ql.exec.ExprNodeEvaluator;
 import org.apache.hadoop.hive.ql.exec.vector.VectorHashKeyWrapper;
 import org.apache.hadoop.hive.ql.exec.vector.VectorHashKeyWrapperBatch;
@@ -34,6 +37,7 @@ import org.apache.hadoop.hive.ql.metadat
 import org.apache.hadoop.hive.serde2.ByteStream.Output;
 import org.apache.hadoop.hive.serde2.SerDe;
 import org.apache.hadoop.hive.serde2.SerDeException;
+import org.apache.hadoop.hive.serde2.binarysortable.BinarySortableSerDe;
 import org.apache.hadoop.hive.serde2.io.HiveDecimalWritable;
 import org.apache.hadoop.hive.serde2.lazybinary.LazyBinarySerDe;
 import org.apache.hadoop.hive.serde2.lazybinary.LazyBinaryUtils;
@@ -48,20 +52,26 @@ import org.apache.hadoop.hive.serde2.obj
 import org.apache.hadoop.io.Writable;
 
 /**
- * The base class for MapJoinKey; also acts as a factory for creating and reading
- * the keys, choosing whether size-optimized byte-array based MJKBytes can be used.
+ * The base class for MapJoinKey.
+ * Ideally, this should now be removed, some table wrappers have no key object.
  */
 public abstract class MapJoinKey {
   private static final byte[] EMPTY_BYTE_ARRAY = new byte[0];
 
+  public abstract void write(MapJoinObjectSerDeContext context, ObjectOutputStream out)
+      throws IOException, SerDeException;
+
+  public abstract boolean hasAnyNulls(int fieldCount, boolean[] nullsafes);
+
   @SuppressWarnings("deprecation")
-  public static MapJoinKey read(Output output, MapJoinKey key, MapJoinObjectSerDeContext context,
-      Writable writable, boolean mayReuseKey) throws SerDeException {
+  public static MapJoinKey read(Output output, MapJoinKey key,
+      MapJoinObjectSerDeContext context, Writable writable, boolean mayReuseKey)
+          throws SerDeException {
     SerDe serde = context.getSerDe();
     Object obj = serde.deserialize(writable);
     boolean useOptimized = useOptimizedKeyBasedOnPrev(key);
     if (useOptimized || key == null) {
-      byte[] structBytes = serializeKey(output, obj, serde.getObjectInspector(), !useOptimized);
+      byte[] structBytes = serialize(output, obj, serde.getObjectInspector(), !useOptimized);
       if (structBytes != null) {
         return MapJoinKeyBytes.fromBytes(key, mayReuseKey, structBytes);
       } else if (useOptimized) {
@@ -95,7 +105,7 @@ public abstract class MapJoinKey {
     SUPPORTED_PRIMITIVES.add(PrimitiveCategory.CHAR);
   }
 
-  private static byte[] serializeKey(Output byteStream,
+  private static byte[] serialize(Output byteStream,
       Object obj, ObjectInspector oi, boolean checkTypes) throws SerDeException {
     if (null == obj || !(oi instanceof StructObjectInspector)) {
       return null; // not supported
@@ -113,16 +123,22 @@ public abstract class MapJoinKey {
     for (int i = 0; i < size; ++i) {
       StructField field = fields.get(i);
       ObjectInspector foi = field.getFieldObjectInspector();
-      if (checkTypes) {
-        if (foi.getCategory() != Category.PRIMITIVE) return null; // not supported
-        PrimitiveCategory pc = ((PrimitiveObjectInspector)foi).getPrimitiveCategory();
-        if (!SUPPORTED_PRIMITIVES.contains(pc)) return null; // not supported
+      if (checkTypes && !isSupportedField(foi)) {
+        return null;
       }
       fieldData[i] = soi.getStructFieldData(obj, field);
       fieldOis.add(foi);
     }
 
-    return serializeRowCommon(byteStream, fieldData, fieldOis);
+    byteStream = serializeRow(byteStream, fieldData, fieldOis, null);
+    return Arrays.copyOf(byteStream.getData(), byteStream.getLength());
+  }
+
+  public static boolean isSupportedField(ObjectInspector foi) {
+    if (foi.getCategory() != Category.PRIMITIVE) return false; // not supported
+    PrimitiveCategory pc = ((PrimitiveObjectInspector)foi).getPrimitiveCategory();
+    if (!SUPPORTED_PRIMITIVES.contains(pc)) return false; // not supported
+    return true;
   }
 
   public static MapJoinKey readFromVector(Output output, MapJoinKey key, VectorHashKeyWrapper kw,
@@ -132,7 +148,10 @@ public abstract class MapJoinKey {
     if (useOptimized || key == null) {
       byte[] structBytes = null;
       try {
-        structBytes = serializeVector(output, kw, keyOutputWriters, keyWrapperBatch);
+        if (keyOutputWriters.length <= 8) {
+          output = serializeVector(output, kw, keyOutputWriters, keyWrapperBatch, null, null);
+          structBytes = Arrays.copyOf(output.getData(), output.getLength());
+        }
       } catch (SerDeException e) {
         throw new HiveException(e);
       }
@@ -148,19 +167,26 @@ public abstract class MapJoinKey {
     return result;
   }
 
-  private static byte[] serializeVector(Output byteStream, VectorHashKeyWrapper kw,
-      VectorExpressionWriter[] keyOutputWriters, VectorHashKeyWrapperBatch keyWrapperBatch)
-          throws HiveException, SerDeException {
+  /**
+   * Serializes row to output for vectorized path.
+   * @param byteStream Output to reuse. Can be null, in that case a new one would be created.
+   */
+  public static Output serializeVector(Output byteStream, VectorHashKeyWrapper kw,
+      VectorExpressionWriter[] keyOutputWriters, VectorHashKeyWrapperBatch keyWrapperBatch,
+      boolean[] nulls, boolean[] sortableSortOrders) throws HiveException, SerDeException {
     Object[] fieldData = new Object[keyOutputWriters.length];
     List<ObjectInspector> fieldOis = new ArrayList<ObjectInspector>();
     for (int i = 0; i < keyOutputWriters.length; ++i) {
       VectorExpressionWriter writer = keyOutputWriters[i];
       fieldOis.add(writer.getObjectInspector());
-      // This is rather convoluted... to simplify for per, we could call getRawKeyValue
+      // This is rather convoluted... to simplify for perf, we could call getRawKeyValue
       // instead of writable, and serialize based on Java type as opposed to OI.
       fieldData[i] = keyWrapperBatch.getWritableKeyValue(kw, i, writer);
+      if (nulls != null) {
+        nulls[i] = (fieldData[i] == null);
+      }
     }
-    return serializeRowCommon(byteStream, fieldData, fieldOis);
+    return serializeRow(byteStream, fieldData, fieldOis, sortableSortOrders);
   }
 
   public static MapJoinKey readFromRow(Output output, MapJoinKey key, Object row,
@@ -173,7 +199,15 @@ public abstract class MapJoinKey {
     boolean useOptimized = useOptimizedKeyBasedOnPrev(key);
     if (useOptimized || key == null) {
       try {
-        byte[] structBytes = serializeRow(output, fieldObjs, keyFieldsOI);
+        byte[] structBytes = null;
+        if (fieldObjs.length <= 8) {
+          if (fieldObjs.length == 0) {
+            structBytes = EMPTY_BYTE_ARRAY; // shortcut for null keys
+          } else {
+            output = serializeRow(output, fieldObjs, keyFieldsOI, null);
+            structBytes = Arrays.copyOf(output.getData(), output.getLength());
+          }
+        }
         if (structBytes != null) {
           return MapJoinKeyBytes.fromBytes(key, mayReuseKey, structBytes);
         } else if (useOptimized) {
@@ -189,34 +223,31 @@ public abstract class MapJoinKey {
     return result;
   }
 
-  private static byte[] serializeRow(Output byteStream,
-      Object[] fieldData, List<ObjectInspector> fieldOis) throws SerDeException {
-    if (fieldData.length > 8) {
-      return null; // not supported
-    } else if (fieldData.length == 0) {
-      return EMPTY_BYTE_ARRAY; // shortcut for null keys
-    }
-    assert fieldData.length == fieldOis.size();
-    return serializeRowCommon(byteStream, fieldData, fieldOis);
-  }
+  private static final Log LOG = LogFactory.getLog(MapJoinKey.class);
 
-  private static byte[] serializeRowCommon(Output byteStream,
-      Object[] fieldData, List<ObjectInspector> fieldOis) throws SerDeException {
+
+  /**
+   * Serializes row to output.
+   * @param byteStream Output to reuse. Can be null, in that case a new one would be created.
+   */
+  public static Output serializeRow(Output byteStream, Object[] fieldData,
+      List<ObjectInspector> fieldOis, boolean[] sortableSortOrders) throws SerDeException {
     if (byteStream == null) {
       byteStream = new Output();
     } else {
       byteStream.reset();
     }
-    LazyBinarySerDe.serializeStruct(byteStream, fieldData, fieldOis);
-    return Arrays.copyOf(byteStream.getData(), byteStream.getCount());
+    if (fieldData.length == 0) {
+      byteStream.reset();
+    } else if (sortableSortOrders == null) {
+      LazyBinarySerDe.serializeStruct(byteStream, fieldData, fieldOis);
+    } else {
+      BinarySortableSerDe.serializeStruct(byteStream, fieldData, fieldOis, sortableSortOrders);
+    }
+    return byteStream;
   }
 
   private static boolean useOptimizedKeyBasedOnPrev(MapJoinKey key) {
     return (key != null) && (key instanceof MapJoinKeyBytes);
   }
-
-  public abstract void write(MapJoinObjectSerDeContext context, ObjectOutputStream out)
-      throws IOException, SerDeException;
-
-  public abstract boolean hasAnyNulls(int fieldCount, boolean[] nullsafes);
-}
\ No newline at end of file
+}

Added: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/MapJoinPersistableTableContainer.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/MapJoinPersistableTableContainer.java?rev=1594713&view=auto
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/MapJoinPersistableTableContainer.java (added)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/MapJoinPersistableTableContainer.java Wed May 14 20:53:30 2014
@@ -0,0 +1,30 @@
+/**
+ * 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.hadoop.hive.ql.exec.persistence;
+
+import java.util.Map;
+import java.util.Set;
+
+public interface MapJoinPersistableTableContainer extends MapJoinTableContainer {
+  void put(MapJoinKey key, MapJoinRowContainer value);
+  Set<Map.Entry<MapJoinKey, MapJoinRowContainer>> entrySet();
+  int size();
+  MapJoinRowContainer get(MapJoinKey key);
+  Map<String, String> getMetaData();
+}

Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/MapJoinRowContainer.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/MapJoinRowContainer.java?rev=1594713&r1=1594712&r2=1594713&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/MapJoinRowContainer.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/MapJoinRowContainer.java Wed May 14 20:53:30 2014
@@ -27,12 +27,12 @@ import org.apache.hadoop.hive.serde2.Ser
 
 public interface MapJoinRowContainer extends AbstractRowContainer<List<Object>> {
 
-  public byte getAliasFilter() throws HiveException;
+  byte getAliasFilter() throws HiveException;
 
-  public MapJoinRowContainer copy() throws HiveException;
+  MapJoinRowContainer copy() throws HiveException;
 
-  public void addRow(Object[] value) throws HiveException;
+  void addRow(Object[] value) throws HiveException;
 
-  public void write(MapJoinObjectSerDeContext valueContext, ObjectOutputStream out)
+  void write(MapJoinObjectSerDeContext valueContext, ObjectOutputStream out)
       throws IOException, SerDeException;
 }
\ No newline at end of file

Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/MapJoinTableContainer.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/MapJoinTableContainer.java?rev=1594713&r1=1594712&r2=1594713&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/MapJoinTableContainer.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/MapJoinTableContainer.java Wed May 14 20:53:30 2014
@@ -18,23 +18,79 @@
 
 package org.apache.hadoop.hive.ql.exec.persistence;
 
-import java.util.Map;
-import java.util.Set;
+import java.util.List;
 
-public interface MapJoinTableContainer {
-
-  public int size();
-
-  public MapJoinRowContainer get(MapJoinKey key);
+import org.apache.hadoop.hive.ql.exec.ExprNodeEvaluator;
+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;
+import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.serde2.SerDeException;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
+import org.apache.hadoop.io.Writable;
 
-  public void put(MapJoinKey key, MapJoinRowContainer value);
-
-  public Set<Map.Entry<MapJoinKey, MapJoinRowContainer>> entrySet();
-
-  public Map<String, String> getMetaData();
+public interface MapJoinTableContainer {
+  /**
+   * Retrieve rows from hashtable key by key, one key at a time, w/o copying the structures
+   * for each key. "Old" HashMapWrapper will still create/retrieve new objects for java HashMap;
+   * but the optimized one doesn't have to.
+   */
+  public interface ReusableGetAdaptor {
+    /**
+     * Changes current rows to which adaptor is referring to the rows corresponding to
+     * the key represented by a VHKW object, and writers and batch used to interpret it.
+     */
+    void setFromVector(VectorHashKeyWrapper kw, VectorExpressionWriter[] keyOutputWriters,
+        VectorHashKeyWrapperBatch keyWrapperBatch) throws HiveException;
+
+    /**
+     * Changes current rows to which adaptor is referring to the rows corresponding to
+     * the key represented by a row object, and fields and ois used to interpret it.
+     */
+    void setFromRow(Object row, List<ExprNodeEvaluator> fields, List<ObjectInspector> ois)
+        throws HiveException;
+
+    /**
+     * Changes current rows to which adaptor is referring to the rows corresponding to
+     * the key that another adaptor has already deserialized via setFromVector/setFromRow.
+     */
+    void setFromOther(ReusableGetAdaptor other);
+
+    /**
+     * Checks whether the current key has any nulls.
+     */
+    boolean hasAnyNulls(int fieldCount, boolean[] nullsafes);
+
+    /**
+     * @return The container w/the rows corresponding to a key set via a previous set... call.
+     */
+    MapJoinRowContainer getCurrentRows();
+  }
+
+  /**
+   * Adds row from input to the table.
+   */
+  MapJoinKey putRow(MapJoinObjectSerDeContext keyContext, Writable currentKey,
+      MapJoinObjectSerDeContext valueContext, Writable currentValue)
+          throws SerDeException, HiveException;
+
+  /**
+   * Indicates to the container that the puts have ended; table is now r/o.
+   */
+  void seal();
+
+  /**
+   * Creates reusable get adaptor that can be used to retrieve rows from the table
+   * based on either vectorized or non-vectorized input rows to MapJoinOperator.
+   * @param keyTypeFromLoader Last key from hash table loader, to determine key type used
+   *                          when loading hashtable (if it can vary).
+   */
+  ReusableGetAdaptor createGetter(MapJoinKey keyTypeFromLoader);
 
-  public void clear();
+  /** Clears the contents of the table. */
+  void clear();
 
-  public MapJoinKey getAnyKey();
+  MapJoinKey getAnyKey();
 
+  void dumpMetrics();
 }

Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/MapJoinTableContainerSerDe.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/MapJoinTableContainerSerDe.java?rev=1594713&r1=1594712&r2=1594713&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/MapJoinTableContainerSerDe.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/MapJoinTableContainerSerDe.java Wed May 14 20:53:30 2014
@@ -54,11 +54,11 @@ public class MapJoinTableContainerSerDe 
    * @param in Input stream.
    * @return Loaded table.
    */
-  public MapJoinTableContainer load(ObjectInputStream in) 
+  public MapJoinPersistableTableContainer load(ObjectInputStream in)
       throws HiveException {
     SerDe keySerDe = keyContext.getSerDe();
     SerDe valueSerDe = valueContext.getSerDe();
-    MapJoinTableContainer tableContainer;
+    MapJoinPersistableTableContainer tableContainer;
     try {
       String name = in.readUTF();
       Map<String, String> metaData = (Map<String, String>) in.readObject();
@@ -86,7 +86,7 @@ public class MapJoinTableContainerSerDe 
       throw new HiveException("Error while trying to create table container", e);
     }
   }
-  public void persist(ObjectOutputStream out, MapJoinTableContainer tableContainer)
+  public void persist(ObjectOutputStream out, MapJoinPersistableTableContainer tableContainer)
       throws HiveException {
     int numKeys = tableContainer.size();
     try { 
@@ -110,17 +110,20 @@ public class MapJoinTableContainerSerDe 
   }
   
   public static void persistDummyTable(ObjectOutputStream out) throws IOException {
-    MapJoinTableContainer tableContainer = new HashMapWrapper();
+    MapJoinPersistableTableContainer tableContainer = new HashMapWrapper();
     out.writeUTF(tableContainer.getClass().getName());
     out.writeObject(tableContainer.getMetaData());
     out.writeInt(tableContainer.size());
   }
-  
-  private MapJoinTableContainer create(String name, Map<String, String> metaData) throws HiveException {
+
+  private MapJoinPersistableTableContainer create(
+      String name, Map<String, String> metaData) throws HiveException {
     try {
       @SuppressWarnings("unchecked")
-      Class<? extends MapJoinTableContainer> clazz = (Class<? extends MapJoinTableContainer>) Class.forName(name);
-      Constructor<? extends MapJoinTableContainer> constructor = clazz.getDeclaredConstructor(Map.class);
+      Class<? extends MapJoinPersistableTableContainer> clazz =
+          (Class<? extends MapJoinPersistableTableContainer>)Class.forName(name);
+      Constructor<? extends MapJoinPersistableTableContainer> constructor =
+          clazz.getDeclaredConstructor(Map.class);
       return constructor.newInstance(metaData);
     } catch (Exception e) {
       String msg = "Error while attemping to create table container" +

Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/HashTableLoader.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/HashTableLoader.java?rev=1594713&r1=1594712&r2=1594713&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/HashTableLoader.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/HashTableLoader.java Wed May 14 20:53:30 2014
@@ -18,26 +18,44 @@
 package org.apache.hadoop.hive.ql.exec.tez;
 
 import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Map;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.debug.Utils;
 import org.apache.hadoop.hive.ql.exec.MapJoinOperator;
 import org.apache.hadoop.hive.ql.exec.MapredContext;
 import org.apache.hadoop.hive.ql.exec.mr.ExecMapperContext;
 import org.apache.hadoop.hive.ql.exec.persistence.HashMapWrapper;
+import org.apache.hadoop.hive.ql.exec.persistence.MapJoinBytesTableContainer;
+import org.apache.hadoop.hive.ql.exec.persistence.MapJoinKey;
 import org.apache.hadoop.hive.ql.exec.persistence.MapJoinKeyObject;
 import org.apache.hadoop.hive.ql.exec.persistence.LazyFlatRowContainer;
 import org.apache.hadoop.hive.ql.exec.persistence.MapJoinKey;
+import org.apache.hadoop.hive.ql.exec.persistence.MapJoinObjectSerDeContext;
 import org.apache.hadoop.hive.ql.exec.persistence.MapJoinTableContainer;
 import org.apache.hadoop.hive.ql.exec.persistence.MapJoinTableContainerSerDe;
 import org.apache.hadoop.hive.ql.metadata.HiveException;
 import org.apache.hadoop.hive.ql.plan.MapJoinDesc;
 import org.apache.hadoop.hive.serde2.ByteStream.Output;
-import org.apache.hadoop.hive.serde2.lazybinary.LazyBinarySerDe;
 import org.apache.hadoop.hive.serde2.SerDeException;
+import org.apache.hadoop.hive.serde2.lazybinary.LazyBinaryFactory;
+import org.apache.hadoop.hive.serde2.lazybinary.LazyBinarySerDe;
+import org.apache.hadoop.hive.serde2.lazybinary.LazyBinaryStruct;
+import org.apache.hadoop.hive.serde2.lazybinary.LazyBinaryUtils;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.PrimitiveObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.StructField;
+import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector.Category;
+import org.apache.hadoop.hive.serde2.objectinspector.PrimitiveObjectInspector.PrimitiveCategory;
+import org.apache.hadoop.hive.serde2.typeinfo.TypeInfo;
+import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoFactory;
+import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoUtils;
 import org.apache.hadoop.io.BytesWritable;
 import org.apache.hadoop.io.Writable;
 import org.apache.tez.runtime.api.LogicalInput;
@@ -70,18 +88,11 @@ public class HashTableLoader implements 
 
     TezContext tezContext = (TezContext) MapredContext.get();
     Map<Integer, String> parentToInput = desc.getParentToInput();
-    int hashTableThreshold = HiveConf.getIntVar(hconf, HiveConf.ConfVars.HIVEHASHTABLETHRESHOLD);
-    float hashTableLoadFactor = HiveConf.getFloatVar(hconf,
-        HiveConf.ConfVars.HIVEHASHTABLELOADFACTOR);
-    boolean useLazyRows = HiveConf.getBoolVar(hconf, HiveConf.ConfVars.HIVEMAPJOINLAZYHASHTABLE);
 
+    boolean useOptimizedTables = HiveConf.getBoolVar(
+        hconf, HiveConf.ConfVars.HIVEMAPJOINUSEOPTIMIZEDTABLE);
+    boolean isFirstKey = true;
     TezCacheAccess tezCacheAccess = TezCacheAccess.createInstance(hconf);
-    // We only check if we can use optimized keys here; that is ok because we don't
-    // create optimized keys in MapJoin if hash map doesn't have optimized keys.
-    if (!HiveConf.getBoolVar(hconf, HiveConf.ConfVars.HIVEMAPJOINUSEOPTIMIZEDKEYS)) {
-      lastKey = new MapJoinKeyObject();
-    }
-    Output output = new Output(); // Reusable output for serialization.
     for (int pos = 0; pos < mapJoinTables.length; pos++) {
       if (pos == desc.getPosBigTable()) {
         continue;
@@ -92,27 +103,29 @@ public class HashTableLoader implements 
 
       try {
         KeyValueReader kvReader = (KeyValueReader) input.getReader();
-
-        MapJoinTableContainer tableContainer = new HashMapWrapper(hashTableThreshold,
-            hashTableLoadFactor);
-
-        // simply read all the kv pairs into the hashtable.
+        MapJoinObjectSerDeContext keyCtx = mapJoinTableSerdes[pos].getKeyContext(),
+          valCtx = mapJoinTableSerdes[pos].getValueContext();
+        if (useOptimizedTables) {
+          ObjectInspector keyOi = keyCtx.getSerDe().getObjectInspector();
+          if (!MapJoinBytesTableContainer.isSupportedKey(keyOi)) {
+            if (isFirstKey) {
+              useOptimizedTables = false;
+            } else {
+              throw new HiveException(describeOi(
+                  "Only a subset of mapjoin keys is supported. Unsupported key: ", keyOi));
+            }
+          }
+        }
+        isFirstKey = false;
+        MapJoinTableContainer tableContainer = useOptimizedTables
+            ? new MapJoinBytesTableContainer(hconf, valCtx) : new HashMapWrapper(hconf);
 
         while (kvReader.next()) {
-          // We pass key in as reference, to find out quickly if optimized keys can be used.
-          // However, we do not reuse the object since we are putting them into the hashmap.
-          lastKey = MapJoinKey.read(output, lastKey, mapJoinTableSerdes[pos].getKeyContext(),
-              (Writable)kvReader.getCurrentKey(), false);
-
-          LazyFlatRowContainer values = (LazyFlatRowContainer)tableContainer.get(lastKey);
-          if (values == null) {
-            values = new LazyFlatRowContainer();
-            tableContainer.put(lastKey, values);
-          }
-          values.add(mapJoinTableSerdes[pos].getValueContext(),
-              (BytesWritable)kvReader.getCurrentValue(), useLazyRows);
+          lastKey = tableContainer.putRow(keyCtx, (Writable)kvReader.getCurrentKey(),
+              valCtx, (Writable)kvReader.getCurrentValue());
         }
 
+        tableContainer.seal();
         mapJoinTables[pos] = tableContainer;
       } catch (IOException e) {
         throw new HiveException(e);
@@ -130,8 +143,17 @@ public class HashTableLoader implements 
         LOG.info("Setting Input: " + inputName + " as cached");
       }
     }
-    if (lastKey == null) {
-      lastKey = new MapJoinKeyObject(); // No rows in tables, the key type doesn't matter.
+  }
+
+  private String describeOi(String desc, ObjectInspector keyOi) {
+    for (StructField field : ((StructObjectInspector)keyOi).getAllStructFieldRefs()) {
+      ObjectInspector oi = field.getFieldObjectInspector();
+      String cat = oi.getCategory().toString();
+      if (oi.getCategory() == Category.PRIMITIVE) {
+        cat = ((PrimitiveObjectInspector)oi).getPrimitiveCategory().toString();
+      }
+      desc += field.getFieldName() + ":" + cat + ", ";
     }
+    return desc;
   }
 }

Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorMapJoinOperator.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorMapJoinOperator.java?rev=1594713&r1=1594712&r2=1594713&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorMapJoinOperator.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorMapJoinOperator.java Wed May 14 20:53:30 2014
@@ -32,6 +32,8 @@ import org.apache.hadoop.hive.ql.exec.Ta
 import org.apache.hadoop.hive.ql.exec.Utilities;
 import org.apache.hadoop.hive.ql.exec.persistence.MapJoinKeyObject;
 import org.apache.hadoop.hive.ql.exec.persistence.MapJoinKey;
+import org.apache.hadoop.hive.ql.exec.persistence.MapJoinKey;
+import org.apache.hadoop.hive.ql.exec.persistence.MapJoinTableContainer.ReusableGetAdaptor;
 import org.apache.hadoop.hive.ql.exec.vector.expressions.VectorExpression;
 import org.apache.hadoop.hive.ql.exec.vector.expressions.VectorExpressionWriter;
 import org.apache.hadoop.hive.ql.exec.vector.expressions.VectorExpressionWriterFactory;
@@ -72,7 +74,6 @@ public class VectorMapJoinOperator exten
   private VectorExpression[] bigTableValueExpressions;
 
   private transient VectorizedRowBatch outputBatch;
-  private transient MapJoinKeyEvaluator keyEvaluator;
   private transient VectorExpressionWriter[] valueWriters;
   private transient Map<ObjectInspector, VectorColumnAssign[]> outputVectorAssigners;
 
@@ -89,10 +90,6 @@ public class VectorMapJoinOperator exten
     super();
   }
 
-  private interface MapJoinKeyEvaluator {
-      MapJoinKey evaluate(VectorHashKeyWrapper kw)
-          throws HiveException;
-  }
 
   public VectorMapJoinOperator (VectorizationContext vContext, OperatorDesc conf)
     throws HiveException {
@@ -142,30 +139,11 @@ public class VectorMapJoinOperator exten
 
     vrbCtx = new VectorizedRowBatchCtx();
     vrbCtx.init(hconf, this.fileKey, (StructObjectInspector) this.outputObjInspector);
-    
+
     outputBatch = vrbCtx.createVectorizedRowBatch();
 
     keyWrapperBatch =VectorHashKeyWrapperBatch.compileKeyWrapperBatch(keyExpressions);
 
-    // This key evaluator translates from the vectorized VectorHashKeyWrapper format
-    // into the row-mode MapJoinKey
-    keyEvaluator = new MapJoinKeyEvaluator() {
-      private MapJoinKey key = null;
-      private final Output output = new Output();
-
-      public MapJoinKeyEvaluator init() {
-        return this;
-      }
-
-      @Override
-      public MapJoinKey evaluate(VectorHashKeyWrapper kw) throws HiveException {
-        MapJoinKey refKey = getRefKey(key, alias);
-        key = MapJoinKey.readFromVector(
-            output, refKey, kw, keyOutputWriters, keyWrapperBatch, refKey == key);
-        return key;
-      }
-    }.init();
-
     Map<Byte, List<ExprNodeDesc>> valueExpressions = conf.getExprs();
     List<ExprNodeDesc> bigTableExpressions = valueExpressions.get(posBigTable);
 
@@ -258,8 +236,9 @@ public class VectorMapJoinOperator exten
   }
 
   @Override
-  protected MapJoinKey computeMapJoinKey(Object row, byte alias) throws HiveException {
-    return keyEvaluator.evaluate(keyValues[batchIndex]);
+  protected void setMapJoinKey(ReusableGetAdaptor dest, Object row, byte alias)
+      throws HiveException {
+    dest.setFromVector(keyValues[batchIndex], keyOutputWriters, keyWrapperBatch);
   }
 
   @Override

Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorizedColumnarSerDe.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorizedColumnarSerDe.java?rev=1594713&r1=1594712&r2=1594713&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorizedColumnarSerDe.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorizedColumnarSerDe.java Wed May 14 20:53:30 2014
@@ -194,8 +194,8 @@ public class VectorizedColumnarSerDe ext
           }
 
           byteRow.get(k).set(serializeVectorStream.getData(), count, serializeVectorStream
-              .getCount() - count);
-          count = serializeVectorStream.getCount();
+              .getLength() - count);
+          count = serializeVectorStream.getLength();
         }
 
       }

Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/ConvertJoinMapJoin.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/ConvertJoinMapJoin.java?rev=1594713&r1=1594712&r2=1594713&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/ConvertJoinMapJoin.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/ConvertJoinMapJoin.java Wed May 14 20:53:30 2014
@@ -362,7 +362,6 @@ public class ConvertJoinMapJoin implemen
 
     Operator<? extends OperatorDesc> parentBigTableOp
       = mapJoinOp.getParentOperators().get(bigTablePosition);
-
     if (parentBigTableOp instanceof ReduceSinkOperator) {
       mapJoinOp.getParentOperators().remove(bigTablePosition);
       if (!(mapJoinOp.getParentOperators().contains(

Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/ReduceSinkMapJoinProc.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/ReduceSinkMapJoinProc.java?rev=1594713&r1=1594712&r2=1594713&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/ReduceSinkMapJoinProc.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/ReduceSinkMapJoinProc.java Wed May 14 20:53:30 2014
@@ -77,7 +77,9 @@ public class ReduceSinkMapJoinProc imple
     context.currentRootOperator = null;
 
     ReduceSinkOperator parentRS = (ReduceSinkOperator)stack.get(stack.size() - 2);
-
+    // remove the tag for in-memory side of mapjoin
+    parentRS.getConf().setSkipTag(true);
+    parentRS.setSkipTag(true);
     // remember the original parent list before we start modifying it.
     if (!context.mapJoinParentMap.containsKey(mapJoinOp)) {
       List<Operator<?>> parents = new ArrayList(mapJoinOp.getParentOperators());
@@ -166,8 +168,7 @@ public class ReduceSinkMapJoinProc imple
     context.linkWorkWithReduceSinkMap.put(parentWork, reduceSinks);
 
     // create the dummy operators
-    List<Operator<? extends OperatorDesc>> dummyOperators =
-        new ArrayList<Operator<? extends OperatorDesc>>();
+    List<Operator<?>> dummyOperators = new ArrayList<Operator<?>>();
 
     // create an new operator: HashTableDummyOperator, which share the table desc
     HashTableDummyDesc desc = new HashTableDummyDesc();

Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/plan/ReduceSinkDesc.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/plan/ReduceSinkDesc.java?rev=1594713&r1=1594712&r2=1594713&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/plan/ReduceSinkDesc.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/plan/ReduceSinkDesc.java Wed May 14 20:53:30 2014
@@ -21,6 +21,9 @@ package org.apache.hadoop.hive.ql.plan;
 import java.util.ArrayList;
 import java.util.List;
 
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
 
 /**
  * ReduceSinkDesc.
@@ -83,7 +86,9 @@ public class ReduceSinkDesc extends Abst
   private int topN = -1;
   private float topNMemoryUsage = -1;
   private boolean mapGroupBy;  // for group-by, values with same key on top-K should be forwarded
+  private boolean skipTag; // Skip writing tags when feeding into mapjoin hashtable
 
+  private static transient Log LOG = LogFactory.getLog(ReduceSinkDesc.class);
   public ReduceSinkDesc() {
   }
 
@@ -133,6 +138,7 @@ public class ReduceSinkDesc extends Abst
     desc.setNumBuckets(numBuckets);
     desc.setBucketCols(bucketCols);
     desc.setStatistics(this.getStatistics());
+    desc.setSkipTag(skipTag);
     return desc;
   }
 
@@ -326,4 +332,12 @@ public class ReduceSinkDesc extends Abst
   public void setBucketCols(List<ExprNodeDesc> bucketCols) {
     this.bucketCols = bucketCols;
   }
+
+  public void setSkipTag(boolean value) {
+    this.skipTag = value;
+  }
+
+  public boolean getSkipTag() {
+    return skipTag;
+  }
 }

Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/udf/UDFToString.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/udf/UDFToString.java?rev=1594713&r1=1594712&r2=1594713&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/udf/UDFToString.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/udf/UDFToString.java Wed May 14 20:53:30 2014
@@ -70,7 +70,7 @@ public class UDFToString extends UDF {
     } else {
       out.reset();
       LazyInteger.writeUTF8NoException(out, i.get());
-      t.set(out.getData(), 0, out.getCount());
+      t.set(out.getData(), 0, out.getLength());
       return t;
     }
   }
@@ -81,7 +81,7 @@ public class UDFToString extends UDF {
     } else {
       out.reset();
       LazyInteger.writeUTF8NoException(out, i.get());
-      t.set(out.getData(), 0, out.getCount());
+      t.set(out.getData(), 0, out.getLength());
       return t;
     }
   }
@@ -92,7 +92,7 @@ public class UDFToString extends UDF {
     } else {
       out.reset();
       LazyInteger.writeUTF8NoException(out, i.get());
-      t.set(out.getData(), 0, out.getCount());
+      t.set(out.getData(), 0, out.getLength());
       return t;
     }
   }
@@ -103,7 +103,7 @@ public class UDFToString extends UDF {
     } else {
       out.reset();
       LazyLong.writeUTF8NoException(out, i.get());
-      t.set(out.getData(), 0, out.getCount());
+      t.set(out.getData(), 0, out.getLength());
       return t;
     }
   }

Added: 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=1594713&view=auto
==============================================================================
--- hive/trunk/ql/src/test/org/apache/hadoop/hive/ql/exec/persistence/TestBytesBytesMultiHashMap.java (added)
+++ hive/trunk/ql/src/test/org/apache/hadoop/hive/ql/exec/persistence/TestBytesBytesMultiHashMap.java Wed May 14 20:53:30 2014
@@ -0,0 +1,237 @@
+/**
+ * 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.hadoop.hive.ql.exec.persistence;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Random;
+
+import org.apache.hadoop.hive.serde2.ByteStream.RandomAccessOutput;
+import org.apache.hadoop.hive.serde2.lazybinary.LazyBinaryUtils;
+import org.apache.hadoop.hive.serde2.SerDeException;
+import org.apache.hadoop.hive.serde2.WriteBuffers;
+import org.junit.Test;
+import static org.junit.Assert.*;
+
+public class TestBytesBytesMultiHashMap {
+  private static final float LOAD_FACTOR = 0.75f;
+  private static final int CAPACITY = 8;
+  private static final int WB_SIZE = 128; // Make sure we cross some buffer boundaries...
+
+  @Test
+  public void testCapacityValidation() {
+    BytesBytesMultiHashMap map = new BytesBytesMultiHashMap(CAPACITY, LOAD_FACTOR, WB_SIZE);
+    assertEquals(CAPACITY, map.getCapacity());
+    map = new BytesBytesMultiHashMap(9, LOAD_FACTOR, WB_SIZE);
+    assertEquals(16, map.getCapacity());
+  }
+
+  @Test
+  public void testPutGetOne() throws Exception {
+    BytesBytesMultiHashMap map = new BytesBytesMultiHashMap(CAPACITY, LOAD_FACTOR, WB_SIZE);
+    RandomKvSource kv = new RandomKvSource(0, 0);
+    map.put(kv);
+    verifyResults(map, kv.getLastKey(), kv.getLastValue());
+    kv = new RandomKvSource(10, 100);
+    map.put(kv);
+    verifyResults(map, kv.getLastKey(), kv.getLastValue());
+  }
+
+  @Test
+  public void testPutGetMultiple() throws Exception {
+    BytesBytesMultiHashMap map = new BytesBytesMultiHashMap(CAPACITY, LOAD_FACTOR, WB_SIZE);
+    RandomKvSource kv = new RandomKvSource(0, 100);
+    map.put(kv);
+    verifyResults(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);
+      verifyResults(map, kv2.key, kv2.values.toArray(new byte[kv2.values.size()][]));
+    }
+  }
+
+  @Test
+  public void testGetNonExistent() throws Exception {
+    BytesBytesMultiHashMap map = new BytesBytesMultiHashMap(CAPACITY, LOAD_FACTOR, WB_SIZE);
+    RandomKvSource kv = new RandomKvSource(1, 100);
+    map.put(kv);
+    byte[] key = kv.getLastKey();
+    key[0] = (byte)(key[0] + 1);
+    FixedKeyKvSource kv2 = new FixedKeyKvSource(kv.getLastKey(), 0, 100);
+    map.put(kv2);
+    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());
+  }
+
+  @Test
+  public void testPutWithFullMap() throws Exception {
+    // Make sure the map does not expand; should be able to find space.
+    BytesBytesMultiHashMap map = new BytesBytesMultiHashMap(CAPACITY, 1f, WB_SIZE);
+    UniqueKeysKvSource kv = new UniqueKeysKvSource();
+    for (int i = 0; i < CAPACITY; ++i) {
+      map.put(kv);
+    }
+    for (int i = 0; i < kv.keys.size(); ++i) {
+      verifyResults(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());
+  }
+
+  @Test
+  public void testExpand() throws Exception {
+    // Start with capacity 1; make sure we expand on every put.
+    BytesBytesMultiHashMap map = new BytesBytesMultiHashMap(1, 0.0000001f, WB_SIZE);
+    UniqueKeysKvSource kv = new UniqueKeysKvSource();
+    for (int i = 0; i < 18; ++i) {
+      map.put(kv);
+      for (int j = 0; j <= i; ++j) {
+        verifyResults(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());
+    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());
+    }
+    for (int i = 0; i < values.length; ++i) {
+      assertTrue(hs.contains(ByteBuffer.wrap(values[i])));
+    }
+  }
+
+  private static class FixedKeyKvSource extends RandomKvSource {
+    private byte[] key;
+
+    public FixedKeyKvSource(byte[] key, int minLength, int maxLength) {
+      super(minLength, maxLength);
+      this.key = key;
+    }
+
+    @Override
+    public void writeKey(RandomAccessOutput dest) throws SerDeException {
+      try {
+        dest.write(key);
+      } catch (IOException e) {
+        e.printStackTrace();
+        fail("Thrown " + e.getMessage());
+      }
+    }
+  }
+
+  private static class UniqueKeysKvSource extends RandomKvSource {
+    private long lastKey = -1;
+    private byte[] buffer = new byte[9];
+    private byte[] lastBuffer;
+
+    public UniqueKeysKvSource() {
+      super(0, 0);
+    }
+
+    @Override
+    public void writeKey(RandomAccessOutput dest) throws SerDeException {
+      lastKey += 465623573; // This number is certified to be random.
+      int len = LazyBinaryUtils.writeVLongToByteArray(buffer, lastKey);
+      lastBuffer = Arrays.copyOf(buffer, len);
+      keys.add(lastBuffer);
+      writeLastBuffer(dest);
+    }
+
+    private void writeLastBuffer(RandomAccessOutput dest) {
+      try {
+        dest.write(lastBuffer);
+      } catch (IOException e) {
+        e.printStackTrace();
+        fail("Thrown " + e.getMessage());
+      }
+    }
+
+    @Override
+    public void writeValue(RandomAccessOutput dest) throws SerDeException {
+      // Assumes value is written after key.
+      values.add(lastBuffer);
+      writeLastBuffer(dest);
+    }
+  }
+
+  private static class RandomKvSource implements BytesBytesMultiHashMap.KvSource {
+    private int minLength, maxLength;
+    private final Random rdm = new Random(43);
+    public List<byte[]> keys = new ArrayList<byte[]>(), values = new ArrayList<byte[]>();
+
+    public RandomKvSource(int minLength, int maxLength) {
+      this.minLength = minLength;
+      this.maxLength = maxLength;
+    }
+
+    public byte[] getLastValue() {
+      return values.get(values.size() - 1);
+    }
+
+    public byte[] getLastKey() {
+      return keys.get(keys.size() - 1);
+    }
+
+    @Override
+    public void writeKey(RandomAccessOutput dest) throws SerDeException {
+      keys.add(write(dest));
+    }
+
+    @Override
+    public void writeValue(RandomAccessOutput dest) throws SerDeException {
+      values.add(write(dest));
+    }
+
+    protected byte[] write(RandomAccessOutput dest) {
+      byte[] bytes = new byte[minLength + rdm.nextInt(maxLength - minLength + 1)];
+      rdm.nextBytes(bytes);
+      try {
+        dest.write(bytes);
+      } catch (IOException e) {
+        e.printStackTrace();
+        fail("Thrown " + e.getMessage());
+      }
+      return bytes;
+    }
+
+    @Override
+    public byte updateStateByte(Byte previousValue) {
+      return (byte)(previousValue == null ? 1 : previousValue + 1);
+    }
+  }
+}

Modified: hive/trunk/ql/src/test/org/apache/hadoop/hive/ql/exec/persistence/TestMapJoinEqualityTableContainer.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/test/org/apache/hadoop/hive/ql/exec/persistence/TestMapJoinEqualityTableContainer.java?rev=1594713&r1=1594712&r2=1594713&view=diff
==============================================================================
--- hive/trunk/ql/src/test/org/apache/hadoop/hive/ql/exec/persistence/TestMapJoinEqualityTableContainer.java (original)
+++ hive/trunk/ql/src/test/org/apache/hadoop/hive/ql/exec/persistence/TestMapJoinEqualityTableContainer.java Wed May 14 20:53:30 2014
@@ -34,7 +34,7 @@ public class TestMapJoinEqualityTableCon
   private static final MapJoinKeyObject KEY3 = new MapJoinKeyObject(new Object[] {new Text("key3")});
   private static final MapJoinKeyObject KEY4 = new MapJoinKeyObject(new Object[] {new Text("key4")});
   private static final Object[] VALUE = new Object[] {new Text("value")};
-  private MapJoinTableContainer container;
+  private HashMapWrapper container;
   private MapJoinRowContainer rowContainer;
   @Before
   public void setup() throws Exception {

Modified: hive/trunk/ql/src/test/org/apache/hadoop/hive/ql/exec/persistence/TestMapJoinTableContainer.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/test/org/apache/hadoop/hive/ql/exec/persistence/TestMapJoinTableContainer.java?rev=1594713&r1=1594712&r2=1594713&view=diff
==============================================================================
--- hive/trunk/ql/src/test/org/apache/hadoop/hive/ql/exec/persistence/TestMapJoinTableContainer.java (original)
+++ hive/trunk/ql/src/test/org/apache/hadoop/hive/ql/exec/persistence/TestMapJoinTableContainer.java Wed May 14 20:53:30 2014
@@ -39,7 +39,7 @@ public class TestMapJoinTableContainer {
   private ByteArrayOutputStream baos;
   private ObjectOutputStream out;
   private ObjectInputStream in;
-  private MapJoinTableContainer container;
+  private MapJoinPersistableTableContainer container;
   private MapJoinTableContainerSerDe containerSerde;
   private MapJoinKeyObject key;
   private MapJoinRowContainer rowContainer;

Modified: hive/trunk/ql/src/test/queries/clientpositive/mapjoin_decimal.q
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/test/queries/clientpositive/mapjoin_decimal.q?rev=1594713&r1=1594712&r2=1594713&view=diff
==============================================================================
--- hive/trunk/ql/src/test/queries/clientpositive/mapjoin_decimal.q (original)
+++ hive/trunk/ql/src/test/queries/clientpositive/mapjoin_decimal.q Wed May 14 20:53:30 2014
@@ -26,10 +26,15 @@ INSERT INTO TABLE t2 select dec from ove
 explain
 select t1.dec, t2.dec from t1 join t2 on (t1.dec=t2.dec);
 
+set hive.mapjoin.optimized.hashtable=false;
 set hive.mapjoin.optimized.keys=false;
 
 select t1.dec, t2.dec from t1 join t2 on (t1.dec=t2.dec);
 
 set hive.mapjoin.optimized.keys=true;
 
+select t1.dec, t2.dec from t1 join t2 on (t1.dec=t2.dec);
+
+set hive.mapjoin.optimized.hashtable=true;
+
 select t1.dec, t2.dec from t1 join t2 on (t1.dec=t2.dec);
\ No newline at end of file

Modified: hive/trunk/ql/src/test/queries/clientpositive/mapjoin_mapjoin.q
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/test/queries/clientpositive/mapjoin_mapjoin.q?rev=1594713&r1=1594712&r2=1594713&view=diff
==============================================================================
--- hive/trunk/ql/src/test/queries/clientpositive/mapjoin_mapjoin.q (original)
+++ hive/trunk/ql/src/test/queries/clientpositive/mapjoin_mapjoin.q Wed May 14 20:53:30 2014
@@ -8,12 +8,18 @@ explain select srcpart.key from srcpart 
 
 explain
 select srcpart.key from srcpart join src on (srcpart.value=src.value) join src1 on (srcpart.key=src1.key) where srcpart.value > 'val_450';
-select srcpart.key from srcpart join src on (srcpart.value=src.value) join src1 on (srcpart.key=src1.key) where srcpart.value > 'val_450';
 
 explain
 select count(*) from srcpart join src on (srcpart.value=src.value) join src src1 on (srcpart.key=src1.key) group by ds;
+
+set hive.mapjoin.optimized.hashtable=false;
+
+select srcpart.key from srcpart join src on (srcpart.value=src.value) join src1 on (srcpart.key=src1.key) where srcpart.value > 'val_450';
+
 select count(*) from srcpart join src on (srcpart.value=src.value) join src src1 on (srcpart.key=src1.key) group by ds;
 
-set hive.mapjoin.lazy.hashtable=false;
+set hive.mapjoin.optimized.hashtable=true;
+
+select srcpart.key from srcpart join src on (srcpart.value=src.value) join src1 on (srcpart.key=src1.key) where srcpart.value > 'val_450';
 
-select count(*) from srcpart join src src on (srcpart.value=src.value) join src src1 on (srcpart.key=src1.key) group by ds;
+select count(*) from srcpart join src on (srcpart.value=src.value) join src src1 on (srcpart.key=src1.key) group by ds;

Modified: hive/trunk/ql/src/test/queries/clientpositive/tez_union.q
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/test/queries/clientpositive/tez_union.q?rev=1594713&r1=1594712&r2=1594713&view=diff
==============================================================================
--- hive/trunk/ql/src/test/queries/clientpositive/tez_union.q (original)
+++ hive/trunk/ql/src/test/queries/clientpositive/tez_union.q Wed May 14 20:53:30 2014
@@ -91,4 +91,4 @@ select u.key as ukey, s.key as skey from
 right outer join src s on u.key = s.key;
 
 select * from ut order by ukey, skey limit 20;
-drop table ut;
\ No newline at end of file
+drop table ut;

Modified: hive/trunk/ql/src/test/results/clientpositive/mapjoin_mapjoin.q.out
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/test/results/clientpositive/mapjoin_mapjoin.q.out?rev=1594713&r1=1594712&r2=1594713&view=diff
==============================================================================
--- hive/trunk/ql/src/test/results/clientpositive/mapjoin_mapjoin.q.out (original)
+++ hive/trunk/ql/src/test/results/clientpositive/mapjoin_mapjoin.q.out Wed May 14 20:53:30 2014
@@ -192,46 +192,6 @@ STAGE PLANS:
       Processor Tree:
         ListSink
 
-PREHOOK: query: select srcpart.key from srcpart join src on (srcpart.value=src.value) join src1 on (srcpart.key=src1.key) where srcpart.value > 'val_450'
-PREHOOK: type: QUERY
-PREHOOK: Input: default@src
-PREHOOK: Input: default@src1
-PREHOOK: Input: default@srcpart
-PREHOOK: Input: default@srcpart@ds=2008-04-08/hr=11
-PREHOOK: Input: default@srcpart@ds=2008-04-08/hr=12
-PREHOOK: Input: default@srcpart@ds=2008-04-09/hr=11
-PREHOOK: Input: default@srcpart@ds=2008-04-09/hr=12
-#### A masked pattern was here ####
-POSTHOOK: query: select srcpart.key from srcpart join src on (srcpart.value=src.value) join src1 on (srcpart.key=src1.key) where srcpart.value > 'val_450'
-POSTHOOK: type: QUERY
-POSTHOOK: Input: default@src
-POSTHOOK: Input: default@src1
-POSTHOOK: Input: default@srcpart
-POSTHOOK: Input: default@srcpart@ds=2008-04-08/hr=11
-POSTHOOK: Input: default@srcpart@ds=2008-04-08/hr=12
-POSTHOOK: Input: default@srcpart@ds=2008-04-09/hr=11
-POSTHOOK: Input: default@srcpart@ds=2008-04-09/hr=12
-#### A masked pattern was here ####
-98
-98
-66
-98
-98
-98
-98
-66
-98
-98
-98
-98
-66
-98
-98
-98
-98
-66
-98
-98
 PREHOOK: query: explain
 select count(*) from srcpart join src on (srcpart.value=src.value) join src src1 on (srcpart.key=src1.key) group by ds
 PREHOOK: type: QUERY
@@ -368,6 +328,46 @@ STAGE PLANS:
       Processor Tree:
         ListSink
 
+PREHOOK: query: select srcpart.key from srcpart join src on (srcpart.value=src.value) join src1 on (srcpart.key=src1.key) where srcpart.value > 'val_450'
+PREHOOK: type: QUERY
+PREHOOK: Input: default@src
+PREHOOK: Input: default@src1
+PREHOOK: Input: default@srcpart
+PREHOOK: Input: default@srcpart@ds=2008-04-08/hr=11
+PREHOOK: Input: default@srcpart@ds=2008-04-08/hr=12
+PREHOOK: Input: default@srcpart@ds=2008-04-09/hr=11
+PREHOOK: Input: default@srcpart@ds=2008-04-09/hr=12
+#### A masked pattern was here ####
+POSTHOOK: query: select srcpart.key from srcpart join src on (srcpart.value=src.value) join src1 on (srcpart.key=src1.key) where srcpart.value > 'val_450'
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@src
+POSTHOOK: Input: default@src1
+POSTHOOK: Input: default@srcpart
+POSTHOOK: Input: default@srcpart@ds=2008-04-08/hr=11
+POSTHOOK: Input: default@srcpart@ds=2008-04-08/hr=12
+POSTHOOK: Input: default@srcpart@ds=2008-04-09/hr=11
+POSTHOOK: Input: default@srcpart@ds=2008-04-09/hr=12
+#### A masked pattern was here ####
+98
+98
+66
+98
+98
+98
+98
+66
+98
+98
+98
+98
+66
+98
+98
+98
+98
+66
+98
+98
 PREHOOK: query: select count(*) from srcpart join src on (srcpart.value=src.value) join src src1 on (srcpart.key=src1.key) group by ds
 PREHOOK: type: QUERY
 PREHOOK: Input: default@src
@@ -388,16 +388,56 @@ POSTHOOK: Input: default@srcpart@ds=2008
 #### A masked pattern was here ####
 5308
 5308
-PREHOOK: query: select count(*) from srcpart join src src on (srcpart.value=src.value) join src src1 on (srcpart.key=src1.key) group by ds
+PREHOOK: query: select srcpart.key from srcpart join src on (srcpart.value=src.value) join src1 on (srcpart.key=src1.key) where srcpart.value > 'val_450'
 PREHOOK: type: QUERY
 PREHOOK: Input: default@src
+PREHOOK: Input: default@src1
 PREHOOK: Input: default@srcpart
 PREHOOK: Input: default@srcpart@ds=2008-04-08/hr=11
 PREHOOK: Input: default@srcpart@ds=2008-04-08/hr=12
 PREHOOK: Input: default@srcpart@ds=2008-04-09/hr=11
 PREHOOK: Input: default@srcpart@ds=2008-04-09/hr=12
 #### A masked pattern was here ####
-POSTHOOK: query: select count(*) from srcpart join src src on (srcpart.value=src.value) join src src1 on (srcpart.key=src1.key) group by ds
+POSTHOOK: query: select srcpart.key from srcpart join src on (srcpart.value=src.value) join src1 on (srcpart.key=src1.key) where srcpart.value > 'val_450'
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@src
+POSTHOOK: Input: default@src1
+POSTHOOK: Input: default@srcpart
+POSTHOOK: Input: default@srcpart@ds=2008-04-08/hr=11
+POSTHOOK: Input: default@srcpart@ds=2008-04-08/hr=12
+POSTHOOK: Input: default@srcpart@ds=2008-04-09/hr=11
+POSTHOOK: Input: default@srcpart@ds=2008-04-09/hr=12
+#### A masked pattern was here ####
+98
+98
+66
+98
+98
+98
+98
+66
+98
+98
+98
+98
+66
+98
+98
+98
+98
+66
+98
+98
+PREHOOK: query: select count(*) from srcpart join src on (srcpart.value=src.value) join src src1 on (srcpart.key=src1.key) group by ds
+PREHOOK: type: QUERY
+PREHOOK: Input: default@src
+PREHOOK: Input: default@srcpart
+PREHOOK: Input: default@srcpart@ds=2008-04-08/hr=11
+PREHOOK: Input: default@srcpart@ds=2008-04-08/hr=12
+PREHOOK: Input: default@srcpart@ds=2008-04-09/hr=11
+PREHOOK: Input: default@srcpart@ds=2008-04-09/hr=12
+#### A masked pattern was here ####
+POSTHOOK: query: select count(*) from srcpart join src on (srcpart.value=src.value) join src src1 on (srcpart.key=src1.key) group by ds
 POSTHOOK: type: QUERY
 POSTHOOK: Input: default@src
 POSTHOOK: Input: default@srcpart

Modified: hive/trunk/ql/src/test/results/clientpositive/tez/mapjoin_decimal.q.out
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/test/results/clientpositive/tez/mapjoin_decimal.q.out?rev=1594713&r1=1594712&r2=1594713&view=diff
==============================================================================
--- hive/trunk/ql/src/test/results/clientpositive/tez/mapjoin_decimal.q.out (original)
+++ hive/trunk/ql/src/test/results/clientpositive/tez/mapjoin_decimal.q.out Wed May 14 20:53:30 2014
@@ -371,3 +371,121 @@ POSTHOOK: Lineage: t2.dec EXPRESSION [(o
 14	14
 14	14
 14	14
+PREHOOK: query: select t1.dec, t2.dec from t1 join t2 on (t1.dec=t2.dec)
+PREHOOK: type: QUERY
+PREHOOK: Input: default@t1
+PREHOOK: Input: default@t2
+#### A masked pattern was here ####
+POSTHOOK: query: select t1.dec, t2.dec from t1 join t2 on (t1.dec=t2.dec)
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@t1
+POSTHOOK: Input: default@t2
+#### A masked pattern was here ####
+POSTHOOK: Lineage: t1.dec SIMPLE [(over1k)over1k.FieldSchema(name:dec, type:decimal(4,2), comment:null), ]
+POSTHOOK: Lineage: t2.dec EXPRESSION [(over1k)over1k.FieldSchema(name:dec, type:decimal(4,2), comment:null), ]
+9	9
+9	9
+9	9
+9	9
+9	9
+9	9
+9	9
+9	9
+9	9
+9	9
+9	9
+9	9
+9	9
+9	9
+9	9
+9	9
+9	9
+9	9
+9	9
+9	9
+9	9
+45	45
+45	45
+45	45
+45	45
+45	45
+79	79
+79	79
+79	79
+79	79
+79	79
+79	79
+17	17
+17	17
+17	17
+17	17
+17	17
+17	17
+17	17
+17	17
+17	17
+17	17
+6	6
+6	6
+6	6
+6	6
+6	6
+6	6
+62	62
+62	62
+62	62
+62	62
+62	62
+62	62
+62	62
+62	62
+62	62
+62	62
+62	62
+62	62
+64	64
+64	64
+64	64
+64	64
+64	64
+64	64
+64	64
+64	64
+64	64
+64	64
+64	64
+64	64
+64	64
+64	64
+64	64
+64	64
+64	64
+64	64
+89	89
+89	89
+89	89
+89	89
+89	89
+89	89
+89	89
+89	89
+89	89
+89	89
+89	89
+89	89
+70	70
+70	70
+70	70
+70	70
+70	70
+70	70
+70	70
+14	14
+14	14
+14	14
+14	14
+14	14
+14	14
+14	14
+14	14
+14	14

Modified: hive/trunk/ql/src/test/results/clientpositive/tez/mapjoin_mapjoin.q.out
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/test/results/clientpositive/tez/mapjoin_mapjoin.q.out?rev=1594713&r1=1594712&r2=1594713&view=diff
==============================================================================
--- hive/trunk/ql/src/test/results/clientpositive/tez/mapjoin_mapjoin.q.out (original)
+++ hive/trunk/ql/src/test/results/clientpositive/tez/mapjoin_mapjoin.q.out Wed May 14 20:53:30 2014
@@ -170,46 +170,6 @@ STAGE PLANS:
       Processor Tree:
         ListSink
 
-PREHOOK: query: select srcpart.key from srcpart join src on (srcpart.value=src.value) join src1 on (srcpart.key=src1.key) where srcpart.value > 'val_450'
-PREHOOK: type: QUERY
-PREHOOK: Input: default@src
-PREHOOK: Input: default@src1
-PREHOOK: Input: default@srcpart
-PREHOOK: Input: default@srcpart@ds=2008-04-08/hr=11
-PREHOOK: Input: default@srcpart@ds=2008-04-08/hr=12
-PREHOOK: Input: default@srcpart@ds=2008-04-09/hr=11
-PREHOOK: Input: default@srcpart@ds=2008-04-09/hr=12
-#### A masked pattern was here ####
-POSTHOOK: query: select srcpart.key from srcpart join src on (srcpart.value=src.value) join src1 on (srcpart.key=src1.key) where srcpart.value > 'val_450'
-POSTHOOK: type: QUERY
-POSTHOOK: Input: default@src
-POSTHOOK: Input: default@src1
-POSTHOOK: Input: default@srcpart
-POSTHOOK: Input: default@srcpart@ds=2008-04-08/hr=11
-POSTHOOK: Input: default@srcpart@ds=2008-04-08/hr=12
-POSTHOOK: Input: default@srcpart@ds=2008-04-09/hr=11
-POSTHOOK: Input: default@srcpart@ds=2008-04-09/hr=12
-#### A masked pattern was here ####
-98
-98
-66
-98
-98
-98
-98
-66
-98
-98
-98
-98
-66
-98
-98
-98
-98
-66
-98
-98
 PREHOOK: query: explain
 select count(*) from srcpart join src on (srcpart.value=src.value) join src src1 on (srcpart.key=src1.key) group by ds
 PREHOOK: type: QUERY
@@ -317,6 +277,46 @@ STAGE PLANS:
       Processor Tree:
         ListSink
 
+PREHOOK: query: select srcpart.key from srcpart join src on (srcpart.value=src.value) join src1 on (srcpart.key=src1.key) where srcpart.value > 'val_450'
+PREHOOK: type: QUERY
+PREHOOK: Input: default@src
+PREHOOK: Input: default@src1
+PREHOOK: Input: default@srcpart
+PREHOOK: Input: default@srcpart@ds=2008-04-08/hr=11
+PREHOOK: Input: default@srcpart@ds=2008-04-08/hr=12
+PREHOOK: Input: default@srcpart@ds=2008-04-09/hr=11
+PREHOOK: Input: default@srcpart@ds=2008-04-09/hr=12
+#### A masked pattern was here ####
+POSTHOOK: query: select srcpart.key from srcpart join src on (srcpart.value=src.value) join src1 on (srcpart.key=src1.key) where srcpart.value > 'val_450'
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@src
+POSTHOOK: Input: default@src1
+POSTHOOK: Input: default@srcpart
+POSTHOOK: Input: default@srcpart@ds=2008-04-08/hr=11
+POSTHOOK: Input: default@srcpart@ds=2008-04-08/hr=12
+POSTHOOK: Input: default@srcpart@ds=2008-04-09/hr=11
+POSTHOOK: Input: default@srcpart@ds=2008-04-09/hr=12
+#### A masked pattern was here ####
+98
+98
+66
+98
+98
+98
+98
+66
+98
+98
+98
+98
+66
+98
+98
+98
+98
+66
+98
+98
 PREHOOK: query: select count(*) from srcpart join src on (srcpart.value=src.value) join src src1 on (srcpart.key=src1.key) group by ds
 PREHOOK: type: QUERY
 PREHOOK: Input: default@src
@@ -337,16 +337,56 @@ POSTHOOK: Input: default@srcpart@ds=2008
 #### A masked pattern was here ####
 5308
 5308
-PREHOOK: query: select count(*) from srcpart join src src on (srcpart.value=src.value) join src src1 on (srcpart.key=src1.key) group by ds
+PREHOOK: query: select srcpart.key from srcpart join src on (srcpart.value=src.value) join src1 on (srcpart.key=src1.key) where srcpart.value > 'val_450'
 PREHOOK: type: QUERY
 PREHOOK: Input: default@src
+PREHOOK: Input: default@src1
 PREHOOK: Input: default@srcpart
 PREHOOK: Input: default@srcpart@ds=2008-04-08/hr=11
 PREHOOK: Input: default@srcpart@ds=2008-04-08/hr=12
 PREHOOK: Input: default@srcpart@ds=2008-04-09/hr=11
 PREHOOK: Input: default@srcpart@ds=2008-04-09/hr=12
 #### A masked pattern was here ####
-POSTHOOK: query: select count(*) from srcpart join src src on (srcpart.value=src.value) join src src1 on (srcpart.key=src1.key) group by ds
+POSTHOOK: query: select srcpart.key from srcpart join src on (srcpart.value=src.value) join src1 on (srcpart.key=src1.key) where srcpart.value > 'val_450'
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@src
+POSTHOOK: Input: default@src1
+POSTHOOK: Input: default@srcpart
+POSTHOOK: Input: default@srcpart@ds=2008-04-08/hr=11
+POSTHOOK: Input: default@srcpart@ds=2008-04-08/hr=12
+POSTHOOK: Input: default@srcpart@ds=2008-04-09/hr=11
+POSTHOOK: Input: default@srcpart@ds=2008-04-09/hr=12
+#### A masked pattern was here ####
+98
+98
+66
+98
+98
+98
+98
+66
+98
+98
+98
+98
+66
+98
+98
+98
+98
+66
+98
+98
+PREHOOK: query: select count(*) from srcpart join src on (srcpart.value=src.value) join src src1 on (srcpart.key=src1.key) group by ds
+PREHOOK: type: QUERY
+PREHOOK: Input: default@src
+PREHOOK: Input: default@srcpart
+PREHOOK: Input: default@srcpart@ds=2008-04-08/hr=11
+PREHOOK: Input: default@srcpart@ds=2008-04-08/hr=12
+PREHOOK: Input: default@srcpart@ds=2008-04-09/hr=11
+PREHOOK: Input: default@srcpart@ds=2008-04-09/hr=12
+#### A masked pattern was here ####
+POSTHOOK: query: select count(*) from srcpart join src on (srcpart.value=src.value) join src src1 on (srcpart.key=src1.key) group by ds
 POSTHOOK: type: QUERY
 POSTHOOK: Input: default@src
 POSTHOOK: Input: default@srcpart

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=1594713&r1=1594712&r2=1594713&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 Wed May 14 20:53:30 2014
@@ -18,12 +18,13 @@
 
 package org.apache.hadoop.hive.serde2;
 
+import java.io.IOException;
+
 import org.apache.hadoop.hive.common.io.NonSyncByteArrayInputStream;
 import org.apache.hadoop.hive.common.io.NonSyncByteArrayOutputStream;
 
 /**
  * Extensions to bytearrayinput/output streams.
- * 
  */
 public class ByteStream {
   /**
@@ -62,16 +63,13 @@ public class ByteStream {
    * Output.
    *
    */
-  public static class Output extends NonSyncByteArrayOutputStream {
+  public static final class Output
+    extends NonSyncByteArrayOutputStream implements RandomAccessOutput {
     @Override
     public byte[] getData() {
       return buf;
     }
 
-    public int getCount() {
-      return count;
-    }
-
     public Output() {
       super();
     }
@@ -79,5 +77,30 @@ public class ByteStream {
     public Output(int size) {
       super(size);
     }
+
+    @Override
+    public void writeInt(long offset, int value) {
+      int offset2 = (int)offset;
+      getData()[offset2++] = (byte) (value >> 24);
+      getData()[offset2++] = (byte) (value >> 16);
+      getData()[offset2++] = (byte) (value >> 8);
+      getData()[offset2] = (byte) (value);
+    }
+
+    @Override
+    public void reserve(int byteCount) {
+      for (int i = 0; i < byteCount; ++i) {
+        write(0);
+      }
+    }
+  }
+
+  public static interface RandomAccessOutput {
+    public void writeInt(long offset, int value);
+    public void reserve(int byteCount);
+    public void write(int b);
+    public void write(byte b[]) throws IOException;
+    public void write(byte b[], int off, int len);
+    public int getLength();
   }
 }