/**
 * {@inheritDoc}
 * @throws IOException If the child InputSplit cannot be read, typically
 *                     for faliing access checks.
 */
@SuppressWarnings("unchecked")  // Generic array assignment
public void readFields(DataInput in) throws IOException {
  int card = WritableUtils.readVInt(in);
  if (splits == null || splits.length != card) {
    splits = new InputSplit[card];
  }
  Class<? extends InputSplit>[] cls = new Class[card];
  try {
    for (int i = 0; i < card; ++i) {
      cls[i] =
        Class.forName(Text.readString(in)).asSubclass(InputSplit.class);
    }
    for (int i = 0; i < card; ++i) {
      splits[i] = ReflectionUtils.newInstance(cls[i], null);
      splits[i].readFields(in);
    }
  } catch (ClassNotFoundException e) {
    throw (IOException)new IOException("Failed split init").initCause(e);
  }
}
 

@Override
public void fromData(DataInput in) throws IOException,
    ClassNotFoundException {
  super.fromData(in);
  regionPath = DataSerializer.readString(in);
  processorId = in.readInt();
  boolean hasId = in.readBoolean();
  if(hasId) {
    id = new PersistentMemberID();
    InternalDataSerializer.invokeFromData(id, in);
  }
  boolean hasInitializingId = in.readBoolean();
  if(hasInitializingId ) {
    initializingId = new PersistentMemberID();
    InternalDataSerializer.invokeFromData(initializingId, in);
  }
}
 

private Writable updateColumnValue(RCFileColumn column, BytesRefWritable bytesRef) throws IOException {
  if(bytesRef.getLength() == 0) {
    // This is a null field.
    return NullWritable.get();
  }
  Writable newColumnValue = column.newWritable();
  // Small optimization to bypass DataInput read if the column writable is
  // BytesRefWritable
  if (newColumnValue.getClass() == BytesRefWritable.class) {
    newColumnValue = bytesRef;
  } else {
    byte[] currentRowBytes = Arrays.copyOfRange(bytesRef.getData(),
        bytesRef.getStart(), bytesRef.getStart() + bytesRef.getLength());
    DataInput dataInput = ByteStreams.newDataInput(currentRowBytes);
    newColumnValue.readFields(dataInput);
  }
  return newColumnValue;
}
 

@Override
  public void fromData(DataInput in)
      throws IOException, ClassNotFoundException {
    ser_deser_time = this.timeStatsEnabled ? (ser_deser_time == 0 ? -1 /*record*/
    : -2/*ignore nested call*/) : 0;
    super.fromData(in);
//    this.inKeys = InternalDataSerializer.readObject(in);
//    this.inRoutingObjects = InternalDataSerializer.readObject(in);
    this.keysToBucketIds = InternalDataSerializer.readObject(in);
    if ((flags & IS_PARTITIONED_TABLE) != 0) {
      this.prId = (int)InternalDataSerializer.readUnsignedVL(in);
    }
    else {
      this.regionPath = DataSerializer.readString(in);
    }
    if (this.timeStatsEnabled && ser_deser_time == -1) {
      this.ser_deser_time = XPLAINUtil.recordStdTiming(getTimestamp());
    }
  }
 

/**
 * Reads a bitset from the stream that has been written with
 * {@link #writeBitSet(DataOutput, int, BitSet)}.
 */
private static final void readBitSet(DataInput stream, int nbits, 
    BitSet bitSet) throws IOException {
  bitSet.clear();
  long initialBits = WritableUtils.readVLong(stream);
  long last = 0L;
  while (0L != initialBits) {
    last = Long.lowestOneBit(initialBits);
    initialBits ^= last;
    bitSet.set(Long.numberOfTrailingZeros(last));
  }
  
  for (int offset=Long.SIZE; offset < nbits; offset+=Byte.SIZE) {
    byte bits = stream.readByte();
    while (0 != bits) {
      last = Long.lowestOneBit(bits);
      bits ^= last;
      bitSet.set(Long.numberOfTrailingZeros(last) + offset);
    }
  }
}
 

/**
 * Similar to readFully(). Skips bytes in a loop.
 * @param in The DataInput to skip bytes from
 * @param len number of bytes to skip.
 * @throws java.io.IOException if it could not skip requested number of bytes
 * for any reason (including EOF)
 */
public static void skipFully(DataInput in, int len) throws IOException {
  int amt = len;
  while (amt > 0) {
    long ret = in.skipBytes(amt);
    if (ret == 0) {
      // skip may return 0 even if we're not at EOF.  Luckily, we can
      // use the read() method to figure out if we're at the end.
      int b = in.readByte();
      if (b == -1) {
        throw new EOFException( "Premature EOF from inputStream after " +
            "skipping " + (len - amt) + " byte(s).");
      }
      ret = 1;
    }
    amt -= ret;
  }
}
 

public void fromData(DataInput in) throws IOException,
    ClassNotFoundException {
  type=in.readInt();
  if (type == GEMFIRE_VERSION) {
    versionOrdinal = in.readShort();
  } else {
    group = in.readUTF();
    mbr = JChannel.getGfFunctions().readObject(in);
    mbrs = JChannel.getGfFunctions().readObject(in);
    //this.locators = (Vector)DataSerializer.readObject(in);
    hasDistributedSystem = in.readBoolean();
    this.floatingCoordinatorDisabled = in.readBoolean();
    this.networkPartitionDetectionEnabled = in.readBoolean();
    this.locators = JChannel.getGfFunctions().readObject(in);
    this.localAddress = JChannel.getGfFunctions().readObject(in);
  }
}
 

@Override
public void readFields(DataInput in) throws IOException {
  //
  // Read fetchers
  //
  
  int nfetchers = WritableUtils.readVInt(in);

  this.fetchers = new String[nfetchers];
  
  for (int i = 0; i < nfetchers; i++) {
    String currentFetcher = WritableUtils.readString(in);
    this.fetchers[i] = currentFetcher;
  }
  
  //
  // Read splits
  //
  
  int splitsize = WritableUtils.readVInt(in);

  this.splits = WritableUtils.readCompressedByteArray(in);
  this.complete = true;
}
 

public CacheBlock deserializeBlock() 
	throws IOException
{
	CacheBlock ret = null;
	
	if( !_shallow ) { //sparse matrix / string frame
		DataInput din = _matrix ? new CacheDataInput(_bdata) :
			new DataInputStream(new ByteArrayInputStream(_bdata));
		ret = _matrix ? new MatrixBlock() : new FrameBlock();
		ret.readFields(din);
	}
	else { //dense matrix/frame
		ret = _cdata;
	}
	
	return ret;
}
 

/**
 * {@inheritDoc}
 * @throws IOException If the child InputSplit cannot be read, typically
 *                     for faliing access checks.
 */
@SuppressWarnings("unchecked")  // Generic array assignment
public void readFields(DataInput in) throws IOException {
  int card = WritableUtils.readVInt(in);
  if (splits == null || splits.length != card) {
    splits = new InputSplit[card];
  }
  Class<? extends InputSplit>[] cls = new Class[card];
  try {
    for (int i = 0; i < card; ++i) {
      cls[i] =
        Class.forName(Text.readString(in)).asSubclass(InputSplit.class);
    }
    for (int i = 0; i < card; ++i) {
      splits[i] = ReflectionUtils.newInstance(cls[i], null);
      splits[i].readFields(in);
    }
  } catch (ClassNotFoundException e) {
    throw (IOException)new IOException("Failed split init").initCause(e);
  }
}
 

private static MetricTimeSeries fromBytes(byte[] buf, MetricSchema schema) throws IOException {
  MetricTimeSeries series = new MetricTimeSeries(schema);
  DataInput in = new DataInputStream(new ByteArrayInputStream(buf));
  int numTimeWindows = in.readInt();
  int bufferSize = in.readInt();
  for (int i = 0; i < numTimeWindows; i++) {
    long timeWindow = in.readLong();
    byte[] bytes = new byte[bufferSize];
    in.readFully(bytes);
    series.metricsValue.put(timeWindow, ByteBuffer.wrap(bytes));
    boolean[] hasValues = new boolean[schema.getNumMetrics()];
    for (int numMetrics = 0; numMetrics < schema.getNumMetrics(); numMetrics++) {
      hasValues[numMetrics] = true;
    }
    series.hasValue.put(timeWindow, hasValues);
  }
  return series;
}
 

@Override
public void readFields(DataInput in) throws IOException {
    int size = in.readInt();
    if (size == -1) {
        return;
    }
    if (columnValues == null) {
        this.columnValues = new Object[size];
        this.columnTypes = new Type[size];
    } else {
        clear();
    }
    for (int i = 0; i < size; i++) {
        Type kuduType = WritableUtils.readEnum(in, Type.class);
        columnTypes[i] = kuduType;
        Object v = HiveKuduBridgeUtils.readObject(in, kuduType);
        columnValues[i] = v;
    }
}
 

private static String readLine(DataInput in) throws IOException {
    ByteArrayOutputStream out = new ByteArrayOutputStream();
    boolean seenCR = false, seenCRLF = false;
    while (!seenCRLF) {
        if (out.size() > MAX_LINE_LENGTH) {
            throw new IllegalStateException("Exceeded maximum line length");
        }
        byte b = in.readByte();
        if (!seenCR && b == 13) {
            seenCR = true;
        } else if (seenCR && b == 10) {
            seenCRLF = true;
        } else {
            seenCR = false;
            out.write(b);
        }
    }
    return out.toString("UTF-8");
}
 

@Override
public void readFields(DataInput in) throws IOException {
  bytes = in.readLong();
  nLoc = in.readInt();
  if (null == locations || locations.length < nLoc) {
    locations = new String[nLoc];
  }
  for (int i = 0; i < nLoc; ++i) {
    locations[i] = Text.readString(in);
  }
}
 

private Object[] deserializeArrayObject(DataInput is)
    throws IOException, ClassNotFoundException {
  int size = LongPacker.unpackInt(is);

  Object[] s = (Object[]) Array.newInstance(Object.class, size);
  for (int i = 0; i < size; i++) {
    s[i] = deserialize(is);
  }
  return s;
}
 

private static DataInput getDataInput(int operationId, int streamIndex) throws IOException {
    ByteArrayOutputStream baos = new ByteArrayOutputStream();
    DataOutputStream dos = new DataOutputStream(baos);
    dos.writeByte(ModelControllerProtocol.PARAM_OPERATION);
    dos.writeInt(operationId);
    dos.writeByte(ModelControllerProtocol.PARAM_INPUTSTREAM_INDEX);
    dos.writeInt(streamIndex);
    dos.flush();
    return new DataInputStream(new ByteArrayInputStream(baos.toByteArray()));
}
 

static long readEpochSec(DataInput in) throws IOException {
    int hiByte = in.readByte() & 255;
    if (hiByte == 255) {
        return in.readLong();
    } else {
        int midByte = in.readByte() & 255;
        int loByte = in.readByte() & 255;
        long tot = ((hiByte << 16) + (midByte << 8) + loByte);
        return (tot * 900) - 4575744000L;
    }
}
 

/** read a message payload transmitted on the given socket input stream */
private void readPayload(final DataInput in) throws Exception {
  if (payload.length > 0) {
    logForDebugging("reading payload...");
    in.readFully(new byte[payload.length]);
  }
}
 

public void fromData(DataInput in) throws IOException, ClassNotFoundException {
  this.controllers = SerializationHelper.readServerLocationList(in);
  this.isBalanced = in.readBoolean();
  if (this.controllers != null && !this.controllers.isEmpty()) {
    this.locatorsFound = true;
  }
}
 

@Override
public void fromData(DataInput in)
throws IOException, ClassNotFoundException {
  super.fromData(in);
  this.regionName = in.readUTF();
  this.key = DataSerializer.readObject(in);
  this.timeoutMs = in.readInt();
  if ((flags & HAS_TTL) != 0) {
    this.ttl = (int)InternalDataSerializer.readSignedVL(in);
  }
  if ((flags & HAS_IDLE_TIME) != 0) {
    this.idleTime = (int)InternalDataSerializer.readSignedVL(in);
  }
  this.alwaysSendResult = (flags & ALWAYS_SEND_RESULT) != 0;
}
 

/** {@inheritDoc} */
@SuppressWarnings("unchecked")
@Override
public void readFields(DataInput in) throws IOException {
  super.readFields(in);
  
  // First clear the map.  Otherwise we will just accumulate
  // entries every time this method is called.
  this.instance.clear();
  
  // Read the number of entries in the map
  
  int entries = in.readInt();
  
  // Then read each key/value pair
  
  for (int i = 0; i < entries; i++) {
    Writable key = (Writable) ReflectionUtils.newInstance(getClass(
        in.readByte()), getConf());
    
    key.readFields(in);
    
    Writable value = (Writable) ReflectionUtils.newInstance(getClass(
        in.readByte()), getConf());
    
    value.readFields(in);
    instance.put(key, value);
  }
}
 

@Override
public void fromData(DataInput in) throws IOException, ClassNotFoundException {
  super.fromData(in);
  
  final long heapBytesUsed = in.readLong();
  MemoryState heapState = MemoryState.fromData(in);
  MemoryThresholds heapThresholds = MemoryThresholds.fromData(in);
  setHeapData(heapBytesUsed, heapState, heapThresholds);
  
  final long offHeapBytesUsed = in.readLong();
  MemoryState offHeapState = MemoryState.fromData(in);
  MemoryThresholds offHeapThresholds = MemoryThresholds.fromData(in);
  setOffHeapData(offHeapBytesUsed, offHeapState, offHeapThresholds);
}
 

/**
 * Modified from {@link java.time.zone.Ser#read}.
 */
private static Object serRead(DataInput in) throws IOException {
    byte type = in.readByte();
    switch (type) {
    case ZRULES:
        return invokeReadExternal(ZoneRules.class, in); // ZoneRules.readExternal(in)
    case ZOT:
        return invokeReadExternal(ZoneOffsetTransition.class, in); // ZoneOffsetTransition.readExternal(in)
    case ZOTRULE:
        return invokeReadExternal(ZoneOffsetTransitionRule.class, in); // ZoneOffsetTransitionRule.readExternal(in)
    default:
        throw new StreamCorruptedException("Unknown serialized type");
    }
}
 

@Override
public void fromData(DataInput in)
    throws IOException, ClassNotFoundException {
  super.fromData(in);
  this.schema = DataSerializer.readString(in);
  this.stmtUUID = DataSerializer.readString(in);
  this.xmlForm = XMLForms.values()[DataSerializer.readInteger(in)];
  this.embedXslFileName = DataSerializer.readString(in);
}
 

/**
 * Decodes a built DateTimeZone from the given stream, as encoded by
 * writeTo.
 *
 * @param in input stream to read encoded DateTimeZone from.
 * @param id time zone id to assign
 */
public static DateTimeZone readFrom(InputStream in, String id) throws IOException {
    if (in instanceof DataInput) {
        return readFrom((DataInput)in, id);
    } else {
        return readFrom((DataInput)new DataInputStream(in), id);
    }
}
 

@Override
public void readFields(DataInput in) throws IOException {
    int len = schemaSize() + 1;
    boolean[] b = SedesHelper.readBooleanArray(in, len);
    generatedCodeReadFields(in, b);
    if (!b[len - 1]) {
        setAppendedFields(SedesHelper.readGenericTuple(in, in.readByte()));
    }
}
 

/**
 * Old interface in hbase-0.94
 *
 * @param in
 * @throws IOException
 */
// @Override
@Deprecated
public void readFields(DataInput in) throws IOException {
    this.comparator = new BooleanExpressionComparator();
    this.comparator.readFields(in);
}
 

@Override
public void fromData(DataInput in)
    throws IOException, ClassNotFoundException {
  super.fromData(in);
  ser_deser_time = this.timeStatsEnabled ? (ser_deser_time == 0 ? -1 /*record*/
      : -2/*ignore nested call*/) : 0;
  this.referenceKeyColumnIndexes = DataSerializer.readIntArray(in);
  // recording end of de-serialization here instead of AbstractOperationMessage.
  if (this.timeStatsEnabled && ser_deser_time == -1) {
    this.ser_deser_time = XPLAINUtil.recordStdTiming(getTimestamp());
  }
}
 

@Override
public void fromData(DataInput in)
    throws IOException, ClassNotFoundException {
  super.fromData(in);
  this.implementation = DataSerializer.readString(in);
  this.initInfoStr = DataSerializer.readString(in);
}
 

private static long innerLongDecode(long l, final DataInput in)
	throws IOException
{
	int b = in.readByte() & 0xFF;
	l ^= (b & 0x7FL) << 28;
	if (b > 0x7F) {
		b = in.readByte() & 0xFF;
		l ^= (b & 0x7FL) << 35;
		if (b > 0x7F) {
			b = in.readByte() & 0xFF;
			l ^= (b & 0x7FL) << 42;
			if (b > 0x7F) {
				b = in.readByte() & 0xFF;
				l ^= (b & 0x7FL) << 49;
				if (b > 0x7F) {
					b = in.readByte() & 0xFF;
					l ^= (b & 0x7FL) << 56;
					if (b > 0x7F) {
						b = in.readByte() & 0xFF;
						l ^= (b & 0x7FL) << 63;
						if (b > 0x7F) {
							throw new IOException("Invalid long encoding.");
						}
					}
				}
			}
		}
	}
	return l;
}