| /* |
| * 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.pig.data; |
| |
| import java.io.ByteArrayInputStream; |
| import java.io.DataInput; |
| import java.io.DataInputStream; |
| import java.io.DataOutput; |
| import java.io.IOException; |
| import java.io.UnsupportedEncodingException; |
| import java.math.BigDecimal; |
| import java.math.BigInteger; |
| import java.nio.ByteBuffer; |
| import java.util.HashMap; |
| import java.util.Iterator; |
| 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.io.DataInputBuffer; |
| import org.apache.hadoop.io.Writable; |
| import org.apache.hadoop.io.WritableComparable; |
| import org.apache.hadoop.io.WritableComparator; |
| import org.apache.hadoop.mapred.JobConf; |
| import org.apache.pig.PigException; |
| import org.apache.pig.backend.executionengine.ExecException; |
| import org.apache.pig.classification.InterfaceAudience; |
| import org.apache.pig.classification.InterfaceStability; |
| import org.apache.pig.data.utils.SedesHelper; |
| import org.apache.pig.impl.util.ObjectSerializer; |
| import org.joda.time.DateTime; |
| import org.joda.time.DateTimeZone; |
| |
| /** |
| * A class to handle reading and writing of intermediate results of data types. The serialization format used by this |
| * class more efficient than what was used in DataReaderWriter . The format used by the functions in this class is |
| * subject to change, so it should be used ONLY to store intermediate results within a pig query. |
| */ |
| @InterfaceAudience.Private |
| @InterfaceStability.Stable |
| public class BinInterSedes implements InterSedes { |
| |
| private static final int ONE_MINUTE = 60000; |
| |
| public static final byte BOOLEAN_TRUE = 0; |
| public static final byte BOOLEAN_FALSE = 1; |
| |
| public static final byte BYTE = 2; |
| |
| public static final byte INTEGER = 3; |
| // since boolean is not supported yet(v0.7) as external type, lot of people use int instead and some data with old |
| // schema is likely stay for some time. so optimizing for that case as well |
| public static final byte INTEGER_0 = 4; |
| public static final byte INTEGER_1 = 5; |
| public static final byte INTEGER_INSHORT = 6; |
| public static final byte INTEGER_INBYTE = 7; |
| |
| public static final byte LONG = 8; |
| public static final byte FLOAT = 9; |
| public static final byte DOUBLE = 10; |
| |
| public static final byte BYTEARRAY = 11; |
| public static final byte SMALLBYTEARRAY = 12; |
| public static final byte TINYBYTEARRAY = 13; |
| |
| public static final byte CHARARRAY = 14; |
| public static final byte SMALLCHARARRAY = 15; |
| |
| public static final byte MAP = 16; |
| public static final byte SMALLMAP = 17; |
| public static final byte TINYMAP = 18; |
| |
| public static final byte TUPLE = 19; |
| public static final byte SMALLTUPLE = 20; |
| public static final byte TINYTUPLE = 21; |
| |
| public static final byte BAG = 22; |
| public static final byte SMALLBAG = 23; |
| public static final byte TINYBAG = 24; |
| |
| public static final byte GENERIC_WRITABLECOMPARABLE = 25; |
| public static final byte INTERNALMAP = 26; |
| |
| public static final byte NULL = 27; |
| |
| public static final byte SCHEMA_TUPLE_BYTE_INDEX = 28; |
| public static final byte SCHEMA_TUPLE_SHORT_INDEX = 29; |
| public static final byte SCHEMA_TUPLE = 30; |
| |
| public static final byte LONG_INBYTE = 31; |
| public static final byte LONG_INSHORT = 32; |
| public static final byte LONG_ININT = 33; |
| public static final byte LONG_0 = 34; |
| public static final byte LONG_1 = 35; |
| |
| public static final byte TUPLE_0 = 36; |
| public static final byte TUPLE_1 = 37; |
| public static final byte TUPLE_2 = 38; |
| public static final byte TUPLE_3 = 39; |
| public static final byte TUPLE_4 = 40; |
| public static final byte TUPLE_5 = 41; |
| public static final byte TUPLE_6 = 42; |
| public static final byte TUPLE_7 = 43; |
| public static final byte TUPLE_8 = 44; |
| public static final byte TUPLE_9 = 45; |
| |
| public static final byte BIGINTEGER = 46; |
| public static final byte BIGDECIMAL = 47; |
| |
| public static final byte DATETIME = 48; |
| |
| private static TupleFactory mTupleFactory = TupleFactory.getInstance(); |
| private static BagFactory mBagFactory = BagFactory.getInstance(); |
| public static final int UNSIGNED_SHORT_MAX = 65535; |
| public static final int UNSIGNED_BYTE_MAX = 255; |
| public static final String UTF8 = "UTF-8"; |
| |
| public Tuple readTuple(DataInput in, byte type) throws IOException { |
| switch (type) { |
| case TUPLE_0: |
| case TUPLE_1: |
| case TUPLE_2: |
| case TUPLE_3: |
| case TUPLE_4: |
| case TUPLE_5: |
| case TUPLE_6: |
| case TUPLE_7: |
| case TUPLE_8: |
| case TUPLE_9: |
| case TUPLE: |
| case TINYTUPLE: |
| case SMALLTUPLE: |
| return SedesHelper.readGenericTuple(in, type); |
| case SCHEMA_TUPLE_BYTE_INDEX: |
| case SCHEMA_TUPLE_SHORT_INDEX: |
| case SCHEMA_TUPLE: |
| return readSchemaTuple(in, type); |
| default: |
| throw new ExecException("Unknown Tuple type found in stream: " + type); |
| } |
| } |
| |
| private Tuple readSchemaTuple(DataInput in, byte type) throws IOException { |
| int id; |
| switch (type) { |
| case (SCHEMA_TUPLE_BYTE_INDEX): id = in.readUnsignedByte(); break; |
| case (SCHEMA_TUPLE_SHORT_INDEX): id = in.readUnsignedShort(); break; |
| case (SCHEMA_TUPLE): id = in.readInt(); break; |
| default: throw new RuntimeException("Invalid type given to readSchemaTuple: " + type); |
| } |
| |
| Tuple st = SchemaTupleFactory.getInstance(id).newTuple(); |
| st.readFields(in); |
| |
| return st; |
| } |
| |
| public int getTupleSize(DataInput in, byte type) throws IOException { |
| int sz; |
| switch (type) { |
| case TUPLE_0: |
| return 0; |
| case TUPLE_1: |
| return 1; |
| case TUPLE_2: |
| return 2; |
| case TUPLE_3: |
| return 3; |
| case TUPLE_4: |
| return 4; |
| case TUPLE_5: |
| return 5; |
| case TUPLE_6: |
| return 6; |
| case TUPLE_7: |
| return 7; |
| case TUPLE_8: |
| return 8; |
| case TUPLE_9: |
| return 9; |
| case TINYTUPLE: |
| sz = in.readUnsignedByte(); |
| break; |
| case SMALLTUPLE: |
| sz = in.readUnsignedShort(); |
| break; |
| case TUPLE: |
| sz = in.readInt(); |
| break; |
| default: { |
| int errCode = 2112; |
| String msg = "Unexpected datatype " + type + " while reading tuple" + "from binary file."; |
| throw new ExecException(msg, errCode, PigException.BUG); |
| } |
| } |
| // if sz == 0, we construct an "empty" tuple - presumably the writer wrote an empty tuple! |
| if (sz < 0) { |
| throw new IOException("Invalid size " + sz + " for a tuple"); |
| } |
| return sz; |
| } |
| |
| private DataBag readBag(DataInput in, byte type) throws IOException { |
| DataBag bag = mBagFactory.newDefaultBag(); |
| long size; |
| // determine size of bag |
| switch (type) { |
| case TINYBAG: |
| size = in.readUnsignedByte(); |
| break; |
| case SMALLBAG: |
| size = in.readUnsignedShort(); |
| break; |
| case BAG: |
| size = in.readLong(); |
| break; |
| default: |
| int errCode = 2219; |
| String msg = "Unexpected data while reading bag " + "from binary file."; |
| throw new ExecException(msg, errCode, PigException.BUG); |
| } |
| |
| for (long i = 0; i < size; i++) { |
| try { |
| Object o = readDatum(in); |
| bag.add((Tuple) o); |
| } catch (ExecException ee) { |
| throw ee; |
| } |
| } |
| return bag; |
| } |
| |
| private Map<String, Object> readMap(DataInput in, byte type) throws IOException { |
| int size; |
| switch (type) { |
| case TINYMAP: |
| size = in.readUnsignedByte(); |
| break; |
| case SMALLMAP: |
| size = in.readUnsignedShort(); |
| break; |
| case MAP: |
| size = in.readInt(); |
| break; |
| default: { |
| int errCode = 2220; |
| String msg = "Unexpected data while reading map" + "from binary file."; |
| throw new ExecException(msg, errCode, PigException.BUG); |
| } |
| } |
| Map<String, Object> m = new HashMap<String, Object>(size); |
| for (int i = 0; i < size; i++) { |
| String key = (String) readDatum(in); |
| m.put(key, readDatum(in)); |
| } |
| return m; |
| } |
| |
| private InternalMap readInternalMap(DataInput in) throws IOException { |
| int size = in.readInt(); |
| InternalMap m = new InternalMap(size); |
| for (int i = 0; i < size; i++) { |
| Object key = readDatum(in); |
| m.put(key, readDatum(in)); |
| } |
| return m; |
| } |
| |
| private WritableComparable readWritable(DataInput in) throws IOException { |
| String className = (String) readDatum(in); |
| // create the writeable class . It needs to have a default constructor |
| Class<?> objClass = null; |
| try { |
| objClass = Class.forName(className); |
| } catch (ClassNotFoundException e) { |
| throw new IOException("Could not find class " + className + ", while attempting to de-serialize it ", e); |
| } |
| WritableComparable writable = null; |
| try { |
| writable = (WritableComparable) objClass.newInstance(); |
| } catch (Exception e) { |
| String msg = "Could create instance of class " + className |
| + ", while attempting to de-serialize it. (no default constructor ?)"; |
| throw new IOException(msg, e); |
| } |
| |
| // read the fields of the object from DataInput |
| writable.readFields(in); |
| return writable; |
| } |
| |
| /* |
| * (non-Javadoc) |
| * |
| * @see org.apache.pig.data.InterSedes#readDatum(java.io.DataInput) |
| */ |
| @Override |
| public Object readDatum(DataInput in) throws IOException, ExecException { |
| // Read the data type |
| byte b = in.readByte(); |
| return readDatum(in, b); |
| } |
| |
| private static Object readBytes(DataInput in, int size) throws IOException { |
| byte[] ba = new byte[size]; |
| in.readFully(ba); |
| return new DataByteArray(ba); |
| } |
| |
| /** |
| * Expects binInterSedes data types (NOT DataType types!) |
| * <p> |
| * |
| * @see org.apache.pig.data.InterSedes#readDatum(java.io.DataInput, byte) |
| */ |
| @Override |
| public Object readDatum(DataInput in, byte type) throws IOException, ExecException { |
| switch (type) { |
| case TUPLE_0: |
| case TUPLE_1: |
| case TUPLE_2: |
| case TUPLE_3: |
| case TUPLE_4: |
| case TUPLE_5: |
| case TUPLE_6: |
| case TUPLE_7: |
| case TUPLE_8: |
| case TUPLE_9: |
| case TUPLE: |
| case TINYTUPLE: |
| case SMALLTUPLE: |
| return SedesHelper.readGenericTuple(in, type); |
| |
| case BAG: |
| case TINYBAG: |
| case SMALLBAG: |
| return readBag(in, type); |
| |
| case MAP: |
| case TINYMAP: |
| case SMALLMAP: |
| return readMap(in, type); |
| |
| case INTERNALMAP: |
| return readInternalMap(in); |
| |
| case INTEGER_0: |
| return Integer.valueOf(0); |
| case INTEGER_1: |
| return Integer.valueOf(1); |
| case INTEGER_INBYTE: |
| return Integer.valueOf(in.readByte()); |
| case INTEGER_INSHORT: |
| return Integer.valueOf(in.readShort()); |
| case INTEGER: |
| return Integer.valueOf(in.readInt()); |
| |
| case LONG_0: |
| return Long.valueOf(0); |
| case LONG_1: |
| return Long.valueOf(1); |
| case LONG_INBYTE: |
| return Long.valueOf(in.readByte()); |
| case LONG_INSHORT: |
| return Long.valueOf(in.readShort()); |
| case LONG_ININT: |
| return Long.valueOf(in.readInt()); |
| case LONG: |
| return Long.valueOf(in.readLong()); |
| |
| case DATETIME: |
| return new DateTime(in.readLong(), DateTimeZone.forOffsetMillis(in.readShort() * ONE_MINUTE)); |
| |
| case FLOAT: |
| return Float.valueOf(in.readFloat()); |
| |
| case DOUBLE: |
| return Double.valueOf(in.readDouble()); |
| |
| case BIGINTEGER: |
| return readBigInteger(in); |
| |
| case BIGDECIMAL: |
| return readBigDecimal(in); |
| |
| case BOOLEAN_TRUE: |
| return Boolean.valueOf(true); |
| |
| case BOOLEAN_FALSE: |
| return Boolean.valueOf(false); |
| |
| case BYTE: |
| return Byte.valueOf(in.readByte()); |
| |
| case TINYBYTEARRAY: |
| case SMALLBYTEARRAY: |
| case BYTEARRAY: |
| return new DataByteArray(SedesHelper.readBytes(in, type)); |
| |
| case CHARARRAY: |
| case SMALLCHARARRAY: |
| return SedesHelper.readChararray(in, type); |
| |
| case GENERIC_WRITABLECOMPARABLE: |
| return readWritable(in); |
| |
| case SCHEMA_TUPLE_BYTE_INDEX: |
| case SCHEMA_TUPLE_SHORT_INDEX: |
| case SCHEMA_TUPLE: |
| return readSchemaTuple(in, type); |
| |
| case NULL: |
| return null; |
| |
| default: |
| throw new RuntimeException("Unexpected data type " + type + " found in stream."); |
| } |
| } |
| |
| private Object readBigDecimal(DataInput in) throws IOException { |
| return new BigDecimal((String)readDatum(in)); |
| } |
| |
| private Object readBigInteger(DataInput in) throws IOException { |
| return new BigInteger((String)readDatum(in)); |
| } |
| |
| private void writeBigInteger(DataOutput out, BigInteger bi) throws IOException { |
| writeDatum(out, bi.toString()); |
| } |
| |
| private void writeBigDecimal(DataOutput out, BigDecimal bd) throws IOException { |
| writeDatum(out, bd.toString()); |
| } |
| |
| /* |
| * (non-Javadoc) |
| * |
| * @see org.apache.pig.data.InterSedes#writeDatum(java.io.DataOutput, java.lang.Object) |
| */ |
| @Override |
| public void writeDatum(DataOutput out, Object val) throws IOException { |
| // Read the data type |
| byte type = DataType.findType(val); |
| writeDatum(out, val, type); |
| } |
| |
| @Override |
| @SuppressWarnings("unchecked") |
| public void writeDatum(DataOutput out, Object val, byte type) throws IOException { |
| switch (type) { |
| case DataType.TUPLE: |
| writeTuple(out, (Tuple) val); |
| break; |
| |
| case DataType.BAG: |
| writeBag(out, (DataBag) val); |
| break; |
| |
| case DataType.MAP: { |
| writeMap(out, (Map<String, Object>) val); |
| break; |
| } |
| |
| case DataType.INTERNALMAP: { |
| out.writeByte(INTERNALMAP); |
| Map<Object, Object> m = (Map<Object, Object>) val; |
| out.writeInt(m.size()); |
| Iterator<Map.Entry<Object, Object>> i = m.entrySet().iterator(); |
| while (i.hasNext()) { |
| Map.Entry<Object, Object> entry = i.next(); |
| writeDatum(out, entry.getKey()); |
| writeDatum(out, entry.getValue()); |
| } |
| break; |
| } |
| |
| case DataType.INTEGER: |
| int i = (Integer) val; |
| if (i == 0) { |
| out.writeByte(INTEGER_0); |
| } else if (i == 1) { |
| out.writeByte(INTEGER_1); |
| } else if (Byte.MIN_VALUE <= i && i <= Byte.MAX_VALUE) { |
| out.writeByte(INTEGER_INBYTE); |
| out.writeByte(i); |
| } else if (Short.MIN_VALUE <= i && i <= Short.MAX_VALUE) { |
| out.writeByte(INTEGER_INSHORT); |
| out.writeShort(i); |
| } else { |
| out.writeByte(INTEGER); |
| out.writeInt(i); |
| } |
| break; |
| |
| case DataType.LONG: |
| long lng = (Long) val; |
| if (lng == 0) { |
| out.writeByte(LONG_0); |
| } else if (lng == 1) { |
| out.writeByte(LONG_1); |
| } else if (Byte.MIN_VALUE <= lng && lng <= Byte.MAX_VALUE) { |
| out.writeByte(LONG_INBYTE); |
| out.writeByte((int)lng); |
| } else if (Short.MIN_VALUE <= lng && lng <= Short.MAX_VALUE) { |
| out.writeByte(LONG_INSHORT); |
| out.writeShort((int)lng); |
| } else if (Integer.MIN_VALUE <= lng && lng <= Integer.MAX_VALUE) { |
| out.writeByte(LONG_ININT); |
| out.writeInt((int)lng); |
| } else { |
| out.writeByte(LONG); |
| out.writeLong(lng); |
| } |
| break; |
| |
| case DataType.DATETIME: |
| out.writeByte(DATETIME); |
| out.writeLong(((DateTime) val).getMillis()); |
| out.writeShort(((DateTime) val).getZone().getOffset((DateTime) val) / ONE_MINUTE); |
| break; |
| |
| case DataType.FLOAT: |
| out.writeByte(FLOAT); |
| out.writeFloat((Float) val); |
| break; |
| |
| case DataType.BIGINTEGER: |
| out.writeByte(BIGINTEGER); |
| writeBigInteger(out, (BigInteger)val); |
| break; |
| |
| case DataType.BIGDECIMAL: |
| out.writeByte(BIGDECIMAL); |
| writeBigDecimal(out, (BigDecimal)val); |
| break; |
| |
| case DataType.DOUBLE: |
| out.writeByte(DOUBLE); |
| out.writeDouble((Double) val); |
| break; |
| |
| case DataType.BOOLEAN: |
| if ((Boolean) val) |
| out.writeByte(BOOLEAN_TRUE); |
| else |
| out.writeByte(BOOLEAN_FALSE); |
| break; |
| |
| case DataType.BYTE: |
| out.writeByte(BYTE); |
| out.writeByte((Byte) val); |
| break; |
| |
| case DataType.BYTEARRAY: { |
| DataByteArray bytes = (DataByteArray) val; |
| SedesHelper.writeBytes(out, bytes.mData); |
| break; |
| |
| } |
| |
| case DataType.CHARARRAY: { |
| SedesHelper.writeChararray(out, (String) val); |
| break; |
| } |
| case DataType.GENERIC_WRITABLECOMPARABLE: |
| out.writeByte(GENERIC_WRITABLECOMPARABLE); |
| // store the class name, so we know the class to create on read |
| writeDatum(out, val.getClass().getName()); |
| Writable writable = (Writable) val; |
| writable.write(out); |
| break; |
| |
| case DataType.NULL: |
| out.writeByte(NULL); |
| break; |
| |
| default: |
| throw new RuntimeException("Unexpected data type " + val.getClass().getName() + " found in stream. " + |
| "Note only standard Pig type is supported when you output from UDF/LoadFunc"); |
| } |
| } |
| |
| private void writeMap(DataOutput out, Map<String, Object> m) throws IOException { |
| |
| final int sz = m.size(); |
| if (sz < UNSIGNED_BYTE_MAX) { |
| out.writeByte(TINYMAP); |
| out.writeByte(sz); |
| } else if (sz < UNSIGNED_SHORT_MAX) { |
| out.writeByte(SMALLMAP); |
| out.writeShort(sz); |
| } else { |
| out.writeByte(MAP); |
| out.writeInt(sz); |
| } |
| Iterator<Map.Entry<String, Object>> i = m.entrySet().iterator(); |
| while (i.hasNext()) { |
| Map.Entry<String, Object> entry = i.next(); |
| writeDatum(out, entry.getKey()); |
| writeDatum(out, entry.getValue()); |
| } |
| } |
| |
| private void writeBag(DataOutput out, DataBag bag) throws IOException { |
| // We don't care whether this bag was sorted or distinct because |
| // using the iterator to write it will guarantee those things come |
| // correctly. And on the other end there'll be no reason to waste |
| // time re-sorting or re-applying distinct. |
| final long sz = bag.size(); |
| if (sz < UNSIGNED_BYTE_MAX) { |
| out.writeByte(TINYBAG); |
| out.writeByte((int) sz); |
| } else if (sz < UNSIGNED_SHORT_MAX) { |
| out.writeByte(SMALLBAG); |
| out.writeShort((int) sz); |
| } else { |
| out.writeByte(BAG); |
| out.writeLong(sz); |
| } |
| |
| Iterator<Tuple> it = bag.iterator(); |
| while (it.hasNext()) { |
| writeTuple(out, it.next()); |
| } |
| |
| } |
| |
| private void writeTuple(DataOutput out, Tuple t) throws IOException { |
| if (t instanceof TypeAwareTuple) { |
| t.write(out); |
| } else { |
| SedesHelper.writeGenericTuple(out, t); |
| } |
| } |
| |
| /* |
| * (non-Javadoc) |
| * |
| * @see org.apache.pig.data.InterSedes#addColsToTuple(java.io.DataInput, org.apache.pig.data.Tuple) |
| */ |
| @Override |
| public void addColsToTuple(DataInput in, Tuple t) throws IOException { |
| byte type = in.readByte(); |
| int sz = getTupleSize(in, type); |
| for (int i = 0; i < sz; i++) { |
| t.append(readDatum(in)); |
| } |
| } |
| |
| public static class BinInterSedesTupleRawComparator extends WritableComparator implements TupleRawComparator { |
| |
| private final Log mLog = LogFactory.getLog(getClass()); |
| private boolean[] mAsc; |
| private boolean[] mSecondaryAsc; |
| private static final boolean[] EMPTY_ASC = new boolean[] {}; |
| private boolean mWholeTuple; |
| private boolean mIsSecondarySort; |
| private boolean mHasNullField; |
| private TupleFactory mFact; |
| private InterSedes mSedes; |
| |
| public BinInterSedesTupleRawComparator() { |
| super(BinSedesTuple.class); |
| } |
| |
| @Override |
| public Configuration getConf() { |
| return null; |
| } |
| |
| @Override |
| public void setConf(Configuration conf) { |
| try { |
| mAsc = (boolean[]) ObjectSerializer.deserialize(conf.get("pig.sortOrder")); |
| mSecondaryAsc = (boolean[]) ObjectSerializer.deserialize(conf.get("pig.secondarySortOrder")); |
| mIsSecondarySort = true; |
| } catch (IOException ioe) { |
| mLog.error("Unable to deserialize sort order object" + ioe.getMessage()); |
| throw new RuntimeException(ioe); |
| } |
| if (mAsc == null) { |
| mAsc = new boolean[1]; |
| mAsc[0] = true; |
| } |
| if (mSecondaryAsc == null) { |
| mIsSecondarySort = false; |
| } |
| // If there's only one entry in mAsc, it means it's for the whole |
| // tuple. So we can't be looking for each column. |
| mWholeTuple = (mAsc.length == 1); |
| mFact = TupleFactory.getInstance(); |
| mSedes = InterSedesFactory.getInterSedesInstance(); |
| } |
| |
| @Override |
| public boolean hasComparedTupleNull() { |
| return mHasNullField; |
| } |
| |
| /** |
| * Compare two BinSedesTuples as raw bytes. We assume the Tuples are NOT PigNullableWritable, so client classes |
| * need to deal with Null and Index. |
| */ |
| @Override |
| public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) { |
| int rc = 0; |
| ByteBuffer bb1 = ByteBuffer.wrap(b1, s1, l1); |
| ByteBuffer bb2 = ByteBuffer.wrap(b2, s2, l2); |
| try { |
| rc = compareBinSedesTuple(bb1, bb2); |
| } catch (IOException ioe) { |
| mLog.error("I/O error during tuple comparison: " + ioe.getMessage()); |
| throw new RuntimeException(ioe.getMessage(), ioe); |
| } |
| return rc; |
| } |
| |
| /** |
| * Compare two BinSedesTuples as raw bytes. We deal with sort ordering in this method. |
| * |
| * @throws IOException |
| */ |
| private int compareBinSedesTuple(ByteBuffer bb1, ByteBuffer bb2) throws IOException { |
| mHasNullField = false; |
| // store the position in case of deserialization |
| int s1 = bb1.position(); |
| int s2 = bb2.position(); |
| // treat the outermost tuple differently because we have to deal with sort order |
| int result = 0; |
| try { |
| // first compare sizes |
| int tsz1 = readSize(bb1); |
| int tsz2 = readSize(bb2); |
| if (tsz1 > tsz2) |
| return 1; |
| else if (tsz1 < tsz2) |
| return -1; |
| else { |
| // if sizes are the same, compare field by field |
| if (mIsSecondarySort) { |
| // we have a compound tuple key (main_key, secondary_key). Each key has its own sort order, so |
| // we have to deal with them separately. We delegate it to the first invocation of |
| // compareBinInterSedesDatum() |
| assert (tsz1 == 2); // main_key, secondary_key |
| result = compareBinInterSedesDatum(bb1, bb2, mAsc); |
| if (result == 0) |
| result = compareBinInterSedesDatum(bb1, bb2, mSecondaryAsc); |
| } else { |
| // we have just one tuple key, we deal with sort order here |
| for (int i = 0; i < tsz1 && result == 0; i++) { |
| // EMPTY_ASC is used to distinguish original calls from recursive ones (hack-ish) |
| result = compareBinInterSedesDatum(bb1, bb2, EMPTY_ASC); |
| // flip if the order is descending |
| if (result != 0) { |
| if (!mWholeTuple && !mAsc[i]) |
| result *= -1; |
| else if (mWholeTuple && !mAsc[0]) |
| result *= -1; |
| } |
| } |
| } |
| } |
| } catch (UnsupportedEncodingException uee) { |
| Tuple t1 = mFact.newTuple(); |
| Tuple t2 = mFact.newTuple(); |
| t1.readFields(new DataInputStream(new ByteArrayInputStream(bb1.array(), s1, bb1.limit()))); |
| t2.readFields(new DataInputStream(new ByteArrayInputStream(bb2.array(), s2, bb2.limit()))); |
| // delegate to compare() |
| result = compare(t1, t2); |
| } |
| return result; |
| } |
| |
| private int compareBinInterSedesDatum(ByteBuffer bb1, ByteBuffer bb2, boolean[] asc) throws IOException { |
| int rc = 0; |
| byte type1, type2; |
| byte dt1 = bb1.get(); |
| byte dt2 = bb2.get(); |
| switch (dt1) { |
| case BinInterSedes.NULL: { |
| type1 = DataType.NULL; |
| type2 = getGeneralizedDataType(dt2); |
| if (asc != null) // we are scanning the top-level Tuple (original call) |
| mHasNullField = true; |
| if (type1 == type2) |
| rc = 0; |
| break; |
| } |
| case BinInterSedes.BOOLEAN_TRUE: |
| case BinInterSedes.BOOLEAN_FALSE: { |
| type1 = DataType.BOOLEAN; |
| type2 = getGeneralizedDataType(dt2); |
| if (type1 == type2) { |
| // false < true |
| int bv1 = (dt1 == BinInterSedes.BOOLEAN_TRUE) ? 1 : 0; |
| int bv2 = (dt2 == BinInterSedes.BOOLEAN_TRUE) ? 1 : 0; |
| rc = bv1 - bv2; |
| } |
| break; |
| } |
| case BinInterSedes.BYTE: { |
| type1 = DataType.BYTE; |
| type2 = getGeneralizedDataType(dt2); |
| if (type1 == type2) { |
| byte bv1 = bb1.get(); |
| byte bv2 = bb2.get(); |
| rc = (bv1 < bv2 ? -1 : (bv1 == bv2 ? 0 : 1)); |
| } |
| break; |
| } |
| case BinInterSedes.INTEGER_0: |
| case BinInterSedes.INTEGER_1: |
| case BinInterSedes.INTEGER_INBYTE: |
| case BinInterSedes.INTEGER_INSHORT: |
| case BinInterSedes.INTEGER: { |
| type1 = DataType.INTEGER; |
| type2 = getGeneralizedDataType(dt2); |
| if (type1 == type2) { |
| int iv1 = readInt(bb1, dt1); |
| int iv2 = readInt(bb2, dt2); |
| rc = (iv1 < iv2 ? -1 : (iv1 == iv2 ? 0 : 1)); |
| } |
| break; |
| } |
| case BinInterSedes.LONG_0: |
| case BinInterSedes.LONG_1: |
| case BinInterSedes.LONG_INBYTE: |
| case BinInterSedes.LONG_INSHORT: |
| case BinInterSedes.LONG_ININT: |
| case BinInterSedes.LONG: { |
| type1 = DataType.LONG; |
| type2 = getGeneralizedDataType(dt2); |
| if (type1 == type2) { |
| long lv1 = readLong(bb1, dt1); |
| long lv2 = readLong(bb2, dt2); |
| rc = (lv1 < lv2 ? -1 : (lv1 == lv2 ? 0 : 1)); |
| } |
| break; |
| } |
| case BinInterSedes.DATETIME: { |
| type1 = DataType.DATETIME; |
| type2 = getGeneralizedDataType(dt2); |
| if (type1 == type2) { |
| long lv1 = bb1.getLong(); |
| bb1.position(bb1.position() + 2); // move cursor forward without read the timezone bytes |
| long lv2 = bb2.getLong(); |
| bb2.position(bb2.position() + 2); |
| rc = (lv1 < lv2 ? -1 : (lv1 == lv2 ? 0 : 1)); |
| } |
| break; |
| } |
| case BinInterSedes.FLOAT: { |
| type1 = DataType.FLOAT; |
| type2 = getGeneralizedDataType(dt2); |
| if (type1 == type2) { |
| float fv1 = bb1.getFloat(); |
| float fv2 = bb2.getFloat(); |
| rc = Float.compare(fv1, fv2); |
| } |
| break; |
| } |
| case BinInterSedes.DOUBLE: { |
| type1 = DataType.DOUBLE; |
| type2 = getGeneralizedDataType(dt2); |
| if (type1 == type2) { |
| double dv1 = bb1.getDouble(); |
| double dv2 = bb2.getDouble(); |
| rc = Double.compare(dv1, dv2); |
| } |
| break; |
| } |
| case BinInterSedes.BIGINTEGER: { |
| type1 = DataType.BIGINTEGER; |
| type2 = getGeneralizedDataType(dt2); |
| if (type1 == type2) { |
| int sz1 = readSize(bb1, bb1.get()); |
| int sz2 = readSize(bb2, bb2.get()); |
| byte[] ca1 = new byte[sz1]; |
| byte[] ca2 = new byte[sz2]; |
| bb1.get(ca1); |
| bb2.get(ca2); |
| String str1 = null, str2 = null; |
| try { |
| str1 = new String(ca1, BinInterSedes.UTF8); |
| str2 = new String(ca2, BinInterSedes.UTF8); |
| } catch (UnsupportedEncodingException uee) { |
| mLog.warn("Unsupported string encoding", uee); |
| uee.printStackTrace(); |
| } |
| if (str1 != null && str2 != null) { |
| rc = new BigInteger(str1).compareTo(new BigInteger(str2)); |
| } |
| } |
| break; |
| } |
| case BinInterSedes.BIGDECIMAL: { |
| type1 = DataType.BIGDECIMAL; |
| type2 = getGeneralizedDataType(dt2); |
| if (type1 == type2) { |
| int sz1 = readSize(bb1, bb1.get()); |
| int sz2 = readSize(bb2, bb2.get()); |
| byte[] ca1 = new byte[sz1]; |
| byte[] ca2 = new byte[sz2]; |
| bb1.get(ca1); |
| bb2.get(ca2); |
| String str1 = null, str2 = null; |
| try { |
| str1 = new String(ca1, BinInterSedes.UTF8); |
| str2 = new String(ca2, BinInterSedes.UTF8); |
| } catch (UnsupportedEncodingException uee) { |
| mLog.warn("Unsupported string encoding", uee); |
| uee.printStackTrace(); |
| } |
| if (str1 != null && str2 != null) { |
| rc = new BigDecimal(str1).compareTo(new BigDecimal(str2)); |
| } |
| } |
| break; |
| } |
| case BinInterSedes.TINYBYTEARRAY: |
| case BinInterSedes.SMALLBYTEARRAY: |
| case BinInterSedes.BYTEARRAY: { |
| type1 = DataType.BYTEARRAY; |
| type2 = getGeneralizedDataType(dt2); |
| if (type1 == type2) { |
| int basz1 = readSize(bb1, dt1); |
| int basz2 = readSize(bb2, dt2); |
| rc = WritableComparator.compareBytes( |
| bb1.array(), bb1.position(), basz1, |
| bb2.array(), bb2.position(), basz2); |
| bb1.position(bb1.position() + basz1); |
| bb2.position(bb2.position() + basz2); |
| } |
| break; |
| } |
| case BinInterSedes.SMALLCHARARRAY: |
| case BinInterSedes.CHARARRAY: { |
| type1 = DataType.CHARARRAY; |
| type2 = getGeneralizedDataType(dt2); |
| if (type1 == type2) { |
| int casz1 = readSize(bb1, dt1); |
| int casz2 = readSize(bb2, dt2); |
| String str1 = null, str2 = null; |
| try { |
| str1 = new String(bb1.array(), bb1.position(), casz1, BinInterSedes.UTF8); |
| str2 = new String(bb2.array(), bb2.position(), casz2, BinInterSedes.UTF8); |
| } catch (UnsupportedEncodingException uee) { |
| mLog.warn("Unsupported string encoding", uee); |
| uee.printStackTrace(); |
| } finally { |
| bb1.position(bb1.position() + casz1); |
| bb2.position(bb2.position() + casz2); |
| } |
| if (str1 != null && str2 != null) |
| rc = str1.compareTo(str2); |
| } |
| break; |
| } |
| case BinInterSedes.TUPLE_0: |
| case BinInterSedes.TUPLE_1: |
| case BinInterSedes.TUPLE_2: |
| case BinInterSedes.TUPLE_3: |
| case BinInterSedes.TUPLE_4: |
| case BinInterSedes.TUPLE_5: |
| case BinInterSedes.TUPLE_6: |
| case BinInterSedes.TUPLE_7: |
| case BinInterSedes.TUPLE_8: |
| case BinInterSedes.TUPLE_9: |
| case BinInterSedes.TINYTUPLE: |
| case BinInterSedes.SMALLTUPLE: |
| case BinInterSedes.TUPLE: { |
| type1 = DataType.TUPLE; |
| type2 = getGeneralizedDataType(dt2); |
| if (type1 == type2) { |
| // first compare sizes |
| int tsz1 = readSize(bb1, dt1); |
| int tsz2 = readSize(bb2, dt2); |
| if (tsz1 > tsz2) |
| return 1; |
| else if (tsz1 < tsz2) |
| return -1; |
| else { |
| // if sizes are the same, compare field by field. If we are doing secondary sort, use the sort |
| // order passed by the caller. Inner tuples never have sort order (so we pass null). |
| for (int i = 0; i < tsz1 && rc == 0; i++) { |
| rc = compareBinInterSedesDatum(bb1, bb2, null); |
| if (rc != 0 && asc != null && asc.length > 1 && !asc[i]) |
| rc *= -1; |
| } |
| } |
| } |
| break; |
| } |
| case BinInterSedes.TINYBAG: |
| case BinInterSedes.SMALLBAG: |
| case BinInterSedes.BAG: { |
| type1 = DataType.BAG; |
| type2 = getGeneralizedDataType(dt2); |
| if (type1 == type2) |
| rc = compareBinInterSedesBag(bb1, bb2, dt1, dt2); |
| break; |
| } |
| case BinInterSedes.TINYMAP: |
| case BinInterSedes.SMALLMAP: |
| case BinInterSedes.MAP: { |
| type1 = DataType.MAP; |
| type2 = getGeneralizedDataType(dt2); |
| if (type1 == type2) |
| rc = compareBinInterSedesMap(bb1, bb2, dt1, dt2); |
| break; |
| } |
| case BinInterSedes.GENERIC_WRITABLECOMPARABLE: { |
| type1 = DataType.GENERIC_WRITABLECOMPARABLE; |
| type2 = getGeneralizedDataType(dt2); |
| if (type1 == type2) |
| rc = compareBinInterSedesGenericWritableComparable(bb1, bb2); |
| break; |
| } |
| default: { |
| mLog.info("Unsupported DataType for binary comparison, switching to object deserialization: " |
| + DataType.genTypeToNameMap().get(dt1) + "(" + dt1 + ")"); |
| throw new UnsupportedEncodingException(); |
| } |
| } |
| // compare generalized data types |
| if (type1 != type2) |
| rc = (type1 < type2) ? -1 : 1; |
| // apply sort order for keys that are not tuples or for whole tuples |
| if (asc != null && asc.length == 1 && !asc[0]) |
| rc *= -1; |
| return rc; |
| } |
| |
| @SuppressWarnings("rawtypes") |
| @Override |
| public int compare(WritableComparable o1, WritableComparable o2) { |
| Tuple t1 = (Tuple) o1; |
| Tuple t2 = (Tuple) o2; |
| mHasNullField = false; |
| // treat the outermost tuple differently because we have to deal with sort order |
| int result = 0; |
| |
| // first compare sizes |
| int tsz1 = t1.size(); |
| int tsz2 = t2.size(); |
| if (tsz1 > tsz2) |
| return 1; |
| else if (tsz1 < tsz2) |
| return -1; |
| else { |
| try { |
| // if sizes are the same, compare field by field |
| if (mIsSecondarySort) { |
| // we have a compound tuple key (main_key, secondary_key). Each key has its own sort order, so |
| // we have to deal with them separately. We delegate it to the first invocation of |
| // compareDatum() |
| assert (tsz1 == 2); // main_key, secondary_key |
| result = compareDatum(t1.get(0), t2.get(0), mAsc); |
| if (result == 0) |
| result = compareDatum(t1.get(1), t2.get(1), mSecondaryAsc); |
| } else { |
| // we have just one tuple key and no chance of recursion, we delegate dealing with sort order to |
| // compareDatum() |
| result = compareDatum(t1, t2, mAsc); |
| } |
| } catch (ExecException e) { |
| throw new RuntimeException("Unable to compare tuples", e); |
| } |
| } |
| return result; |
| } |
| |
| private int compareDatum(Object o1, Object o2, boolean[] asc) { |
| int rc = 0; |
| if (o1 != null && o2 != null && o1 instanceof Tuple && o2 instanceof Tuple) { |
| // objects are Tuples, we may need to apply sort order inside them |
| Tuple t1 = (Tuple) o1; |
| Tuple t2 = (Tuple) o2; |
| int sz1 = t1.size(); |
| int sz2 = t2.size(); |
| if (sz2 < sz1) { |
| return 1; |
| } else if (sz2 > sz1) { |
| return -1; |
| } else { |
| for (int i = 0; i < sz1; i++) { |
| try { |
| rc = DataType.compare(t1.get(i), t2.get(i)); |
| if (rc != 0 && asc != null && asc.length > 1 && !asc[i]) |
| rc *= -1; |
| if (t1.get(i) == null) // (PIG-927) record if the tuple has a null field |
| mHasNullField = true; |
| if (rc!=0) break; |
| } catch (ExecException e) { |
| throw new RuntimeException("Unable to compare tuples", e); |
| } |
| } |
| } |
| } else { |
| // objects are NOT Tuples, delegate to DataType.compare() |
| rc = DataType.compare(o1, o2); |
| } |
| // apply sort order for keys that are not tuples or for whole tuples |
| if (asc != null && asc.length == 1 && !asc[0]) |
| rc *= -1; |
| return rc; |
| } |
| |
| @SuppressWarnings({ "unchecked", "rawtypes" }) |
| private int compareBinInterSedesGenericWritableComparable(ByteBuffer bb1, ByteBuffer bb2) throws ExecException, |
| IOException { |
| DataInputBuffer buffer1 = new DataInputBuffer(); |
| DataInputBuffer buffer2 = new DataInputBuffer(); |
| buffer1.reset(bb1.array(), bb1.position(), bb1.remaining()); |
| buffer2.reset(bb2.array(), bb2.position(), bb2.remaining()); |
| Comparable writable1 = (Comparable) mSedes.readDatum(buffer1); |
| Comparable writable2 = (Comparable) mSedes.readDatum(buffer2); |
| bb1.position(buffer1.getPosition()); |
| bb2.position(buffer2.getPosition()); |
| return writable1.compareTo(writable2); |
| } |
| |
| @SuppressWarnings("unchecked") |
| private int compareBinInterSedesBag(ByteBuffer bb1, ByteBuffer bb2, byte dt1, byte dt2) throws IOException { |
| int s1 = bb1.position(); |
| int s2 = bb2.position(); |
| int l1 = bb1.remaining(); |
| int l2 = bb2.remaining(); |
| // first compare sizes |
| int bsz1 = readSize(bb1, dt1); |
| int bsz2 = readSize(bb2, dt2); |
| if (bsz1 > bsz2) |
| return 1; |
| else if (bsz1 < bsz2) |
| return -1; |
| else { |
| DataInputBuffer buffer1 = new DataInputBuffer(); |
| DataInputBuffer buffer2 = new DataInputBuffer(); |
| buffer1.reset(bb1.array(), s1, l1); |
| buffer2.reset(bb2.array(), s2, l2); |
| DataBag bag1 = (DataBag) mSedes.readDatum(buffer1, dt1); |
| DataBag bag2 = (DataBag) mSedes.readDatum(buffer2, dt2); |
| bb1.position(buffer1.getPosition()); |
| bb2.position(buffer2.getPosition()); |
| return bag1.compareTo(bag2); |
| } |
| } |
| |
| @SuppressWarnings("unchecked") |
| private int compareBinInterSedesMap(ByteBuffer bb1, ByteBuffer bb2, byte dt1, byte dt2) throws ExecException, |
| IOException { |
| int s1 = bb1.position(); |
| int s2 = bb2.position(); |
| int l1 = bb1.remaining(); |
| int l2 = bb2.remaining(); |
| // first compare sizes |
| int bsz1 = readSize(bb1, dt1); |
| int bsz2 = readSize(bb2, dt2); |
| if (bsz1 > bsz2) |
| return 1; |
| else if (bsz1 < bsz2) |
| return -1; |
| else { |
| DataInputBuffer buffer1 = new DataInputBuffer(); |
| DataInputBuffer buffer2 = new DataInputBuffer(); |
| buffer1.reset(bb1.array(), s1, l1); |
| buffer2.reset(bb2.array(), s2, l2); |
| Map<String, Object> map1 = (Map<String, Object>) mSedes.readDatum(buffer1, dt1); |
| Map<String, Object> map2 = (Map<String, Object>) mSedes.readDatum(buffer2, dt2); |
| bb1.position(buffer1.getPosition()); |
| bb2.position(buffer2.getPosition()); |
| return DataType.compare(map1, map2, DataType.MAP, DataType.MAP); |
| } |
| } |
| |
| private static byte getGeneralizedDataType(byte type) { |
| switch (type) { |
| case BinInterSedes.NULL: |
| return DataType.NULL; |
| case BinInterSedes.BOOLEAN_TRUE: |
| case BinInterSedes.BOOLEAN_FALSE: |
| return DataType.BOOLEAN; |
| case BinInterSedes.BYTE: |
| return DataType.BYTE; |
| case BinInterSedes.INTEGER_0: |
| case BinInterSedes.INTEGER_1: |
| case BinInterSedes.INTEGER_INBYTE: |
| case BinInterSedes.INTEGER_INSHORT: |
| case BinInterSedes.INTEGER: |
| return DataType.INTEGER; |
| case BinInterSedes.LONG_0: |
| case BinInterSedes.LONG_1: |
| case BinInterSedes.LONG_INBYTE: |
| case BinInterSedes.LONG_INSHORT: |
| case BinInterSedes.LONG_ININT: |
| case BinInterSedes.LONG: |
| return DataType.LONG; |
| case BinInterSedes.DATETIME: |
| return DataType.DATETIME; |
| case BinInterSedes.FLOAT: |
| return DataType.FLOAT; |
| case BinInterSedes.DOUBLE: |
| return DataType.DOUBLE; |
| case BinInterSedes.BIGINTEGER: |
| return DataType.BIGINTEGER; |
| case BinInterSedes.BIGDECIMAL: |
| return DataType.BIGDECIMAL; |
| case BinInterSedes.TINYBYTEARRAY: |
| case BinInterSedes.SMALLBYTEARRAY: |
| case BinInterSedes.BYTEARRAY: |
| return DataType.BYTEARRAY; |
| case BinInterSedes.SMALLCHARARRAY: |
| case BinInterSedes.CHARARRAY: |
| return DataType.CHARARRAY; |
| case BinInterSedes.TUPLE_0: |
| case BinInterSedes.TUPLE_1: |
| case BinInterSedes.TUPLE_2: |
| case BinInterSedes.TUPLE_3: |
| case BinInterSedes.TUPLE_4: |
| case BinInterSedes.TUPLE_5: |
| case BinInterSedes.TUPLE_6: |
| case BinInterSedes.TUPLE_7: |
| case BinInterSedes.TUPLE_8: |
| case BinInterSedes.TUPLE_9: |
| case BinInterSedes.TUPLE: |
| case BinInterSedes.TINYTUPLE: |
| case BinInterSedes.SMALLTUPLE: |
| return DataType.TUPLE; |
| case BinInterSedes.BAG: |
| case BinInterSedes.TINYBAG: |
| case BinInterSedes.SMALLBAG: |
| return DataType.BAG; |
| case BinInterSedes.MAP: |
| case BinInterSedes.TINYMAP: |
| case BinInterSedes.SMALLMAP: |
| return DataType.MAP; |
| case BinInterSedes.INTERNALMAP: |
| return DataType.INTERNALMAP; |
| case BinInterSedes.GENERIC_WRITABLECOMPARABLE: |
| return DataType.GENERIC_WRITABLECOMPARABLE; |
| default: |
| throw new RuntimeException("Unexpected data type " + type + " found in stream."); |
| } |
| } |
| |
| private static long readLong(ByteBuffer bb, byte type) { |
| int bytesToRead = 0; |
| switch (type) { |
| case BinInterSedes.LONG_0: return 0L; |
| case BinInterSedes.LONG_1: return 1L; |
| case BinInterSedes.LONG_INBYTE: return bb.get(); |
| case BinInterSedes.LONG_INSHORT: return bb.getShort(); |
| case BinInterSedes.LONG_ININT: return bb.getInt(); |
| case BinInterSedes.LONG: return bb.getLong(); |
| default: |
| throw new RuntimeException("Unexpected data type " + type + " found in stream."); |
| } |
| } |
| |
| private static int readInt(ByteBuffer bb, byte type) { |
| switch (type) { |
| case BinInterSedes.INTEGER_0: |
| return 0; |
| case BinInterSedes.INTEGER_1: |
| return 1; |
| case BinInterSedes.INTEGER_INBYTE: |
| return bb.get(); |
| case BinInterSedes.INTEGER_INSHORT: |
| return bb.getShort(); |
| case BinInterSedes.INTEGER: |
| return bb.getInt(); |
| default: |
| throw new RuntimeException("Unexpected data type " + type + " found in stream."); |
| } |
| } |
| |
| /** |
| * @param bb ByteBuffer having serialized object, including the type information |
| * @param type serialized type information |
| * @return the size of this type |
| */ |
| private static int readSize(ByteBuffer bb) { |
| return readSize(bb, bb.get()); |
| } |
| |
| /** |
| * @param bb ByteBuffer having serialized object, minus the type information |
| * @param type serialized type information |
| * @return the size of this type |
| */ |
| private static int readSize(ByteBuffer bb, byte type) { |
| switch (type) { |
| case BinInterSedes.TINYBYTEARRAY: |
| case BinInterSedes.TINYTUPLE: |
| case BinInterSedes.TINYBAG: |
| case BinInterSedes.TINYMAP: |
| return getUnsignedByte(bb); |
| case BinInterSedes.SMALLBYTEARRAY: |
| case BinInterSedes.SMALLCHARARRAY: |
| case BinInterSedes.SMALLTUPLE: |
| case BinInterSedes.SMALLBAG: |
| case BinInterSedes.SMALLMAP: |
| return getUnsignedShort(bb); |
| case BinInterSedes.BYTEARRAY: |
| case BinInterSedes.CHARARRAY: |
| case BinInterSedes.TUPLE: |
| case BinInterSedes.BAG: |
| case BinInterSedes.MAP: |
| return bb.getInt(); |
| case BinInterSedes.TUPLE_0: |
| return 0; |
| case BinInterSedes.TUPLE_1: |
| return 1; |
| case BinInterSedes.TUPLE_2: |
| return 2; |
| case BinInterSedes.TUPLE_3: |
| return 3; |
| case BinInterSedes.TUPLE_4: |
| return 4; |
| case BinInterSedes.TUPLE_5: |
| return 5; |
| case BinInterSedes.TUPLE_6: |
| return 6; |
| case BinInterSedes.TUPLE_7: |
| return 7; |
| case BinInterSedes.TUPLE_8: |
| return 8; |
| case BinInterSedes.TUPLE_9: |
| return 9; |
| default: |
| throw new RuntimeException("Unexpected data type " + type + " found in stream."); |
| } |
| } |
| |
| //same as format used by DataInput/DataOutput for unsigned short |
| private static int getUnsignedShort(ByteBuffer bb) { |
| return (((bb.get() & 0xff) << 8) | (bb.get() & 0xff)); |
| } |
| |
| //same as format used by DataInput/DataOutput for unsigned byte |
| private static int getUnsignedByte(ByteBuffer bb) { |
| return bb.get() & 0xff; |
| } |
| } |
| |
| @Override |
| public Class<? extends TupleRawComparator> getTupleRawComparatorClass() { |
| return BinInterSedesTupleRawComparator.class; |
| } |
| |
| public Tuple readTuple(DataInput in) throws IOException { |
| return readTuple(in, in.readByte()); |
| } |
| |
| public static boolean isTupleByte(byte b) { |
| return b == BinInterSedes.TUPLE |
| || b == BinInterSedes.SMALLTUPLE |
| || b == BinInterSedes.TINYTUPLE |
| || b == BinInterSedes.SCHEMA_TUPLE |
| || b == BinInterSedes.SCHEMA_TUPLE_BYTE_INDEX |
| || b == BinInterSedes.SCHEMA_TUPLE_SHORT_INDEX |
| || b == BinInterSedes.TUPLE_0 |
| || b == BinInterSedes.TUPLE_1 |
| || b == BinInterSedes.TUPLE_2 |
| || b == BinInterSedes.TUPLE_3 |
| || b == BinInterSedes.TUPLE_4 |
| || b == BinInterSedes.TUPLE_5 |
| || b == BinInterSedes.TUPLE_6 |
| || b == BinInterSedes.TUPLE_7 |
| || b == BinInterSedes.TUPLE_8 |
| || b == BinInterSedes.TUPLE_9; |
| } |
| } |
