Restore Block.bodyLength to long

Change-Id: I4bae06d5833ffd24f94522ad23ea2dfcc459d86b

Author: wesm

format/File.fbs

@@ -43,7 +43,7 @@ struct Block {
 
  /// Length of the data (this is aligned so there can be a gap between this and
  /// the metatdata).
- bodyLength: int;
+ bodyLength: long;
 }
 
 root_type Footer;

java/vector/src/main/java/org/apache/arrow/vector/file/ArrowBlock.java

@@ -26,9 +26,9 @@ public class ArrowBlock implements FBSerializable {
 
  private final long offset;
  private final int metadataLength;
- private final int bodyLength;
+ private final long bodyLength;
 
- public ArrowBlock(long offset, int metadataLength, int bodyLength) {
+ public ArrowBlock(long offset, int metadataLength, long bodyLength) {
  super();
  this.offset = offset;
  this.metadataLength = metadataLength;
@@ -43,7 +43,7 @@ public int getMetadataLength() {
  return metadataLength;
  }
 
- public int getBodyLength() {
+ public long getBodyLength() {
  return bodyLength;
  }
 
@@ -56,7 +56,7 @@ public int writeTo(FlatBufferBuilder builder) {
  public int hashCode() {
  final int prime = 31;
  int result = 1;
- result = prime * result + bodyLength;
+ result = prime * result + (int) (bodyLength ^ (bodyLength >>> 32));
  result = prime * result + metadataLength;
  result = prime * result + (int) (offset ^ (offset >>> 32));
  return result;

java/vector/src/main/java/org/apache/arrow/vector/stream/MessageSerializer.java

@@ -154,7 +154,7 @@ public static ArrowBlock serialize(WriteChannel out, ArrowRecordBatch batch)
  " != " + startPosition + layout.getSize());
  }
  }
- return new ArrowBlock(batchStart, metadataSize, (int)(out.getCurrentPosition() - bufferStart));
+ return new ArrowBlock(batchStart, metadataSize, out.getCurrentPosition() - bufferStart);
  }
 
  /**
@@ -188,18 +188,23 @@ public static ArrowRecordBatch deserializeRecordBatch(ReadChannel in,
  public static ArrowRecordBatch deserializeRecordBatch(ReadChannel in, ArrowBlock block,
  BufferAllocator alloc) throws IOException {
  long readPosition = in.getCurrentPositiion();
- int totalLen = block.getMetadataLength() + block.getBodyLength();
+ long totalLen = block.getMetadataLength() + block.getBodyLength();
  if ((readPosition + block.getMetadataLength()) % 8 != 0) {
  // Compute padded size.
  totalLen += (8 - (readPosition + block.getMetadataLength()) % 8);
  }
 
- ArrowBuf buffer = alloc.buffer(totalLen);
- if (in.readFully(buffer, totalLen) != totalLen) {
+ if (totalLen > Integer.MAX_VALUE) {
+ throw new IOException("Cannot currently deserialize record batches over 2GB");
+ }
+
+
+ ArrowBuf buffer = alloc.buffer((int) totalLen);
+ if (in.readFully(buffer, (int) totalLen) != totalLen) {
  throw new IOException("Unexpected end of input trying to read batch.");
  }
 
- return deserializeRecordBatch(buffer, readPosition, block.getMetadataLength(), totalLen);
+ return deserializeRecordBatch(buffer, readPosition, block.getMetadataLength(), (int) totalLen);
  }
 
  // Deserializes a record batch. Buffer should start at the RecordBatch and include