ARROW-264: File format

cpp/src/arrow/ipc/metadata-internal.cc

@@ -219,6 +219,7 @@ static Status FieldToFlatbuffer(
   RETURN_NOT_OK(TypeToFlatbuffer(fbb, field->type, &children, &type_enum, &type_data));
   auto fb_children = fbb.CreateVector(children);
 
+  // TODO: produce the list of VectorTypes
   *offset = flatbuf::CreateField(
       fbb, fb_name, field->nullable, type_enum, type_data, field->dictionary,
       fb_children);

format/File.fbs

@@ -0,0 +1,28 @@
+include "Message.fbs";
+
+namespace org.apache.arrow.flatbuf;
+
+/// ----------------------------------------------------------------------
+/// Arrow File metadata
+///
+
+table Footer {
+
+  schema: org.apache.arrow.flatbuf.Schema;
+
+  dictionaries: [ Block ];
+
+  recordBatches: [ Block ];
+}
+
+struct Block {
+
+  offset: long;
+
+  metaDataLength: int;
+
+  bodyLength: long;
+
+}
+
+root_type Footer;

format/Message.fbs

@@ -17,7 +17,7 @@ table Tuple {
 table List {
 }
 
-enum UnionMode:int { Sparse, Dense }
+enum UnionMode:short { Sparse, Dense }
 
 table Union {
   mode: UnionMode;
@@ -28,7 +28,7 @@ table Int {
   is_signed: bool;
 }
 
-enum Precision:int {SINGLE, DOUBLE}
+enum Precision:short {SINGLE, DOUBLE}
 
 table FloatingPoint {
   precision: Precision;
@@ -91,6 +91,17 @@ union Type {
   JSONScalar
 }
 
+enum VectorType: short {
+  /// used in List type Dense Union and variable length primitive types (String, Binary)
+  OFFSET,
+  /// fixed length primitive values
+  VALUES,
+  /// Bit vector indicated if each value is null
+  VALIDITY,
+  /// Type vector used in Union type
+  TYPE
+}
+
 /// ----------------------------------------------------------------------
 /// A field represents a named column in a record / row batch or child of a
 /// nested type.
@@ -109,12 +120,16 @@ table Field {
   dictionary: long;
   // children apply only to Nested data types like Struct, List and Union
   children: [Field];
+  /// the buffers produced for this type (as derived from the Type)
+  /// does not include children
+  /// each recordbatch will return instances of those Buffers.
+  buffers: [ VectorType ];
 }
 
 /// ----------------------------------------------------------------------
 /// Endianness of the platform that produces the RecordBatch
 
-enum Endianness:int { Little, Big }
+enum Endianness:short { Little, Big }
 
 /// ----------------------------------------------------------------------
 /// A Schema describes the columns in a row batch

java/format/pom.xml

@@ -106,6 +106,7 @@
               <argument>-o</argument>
               <argument>target/generated-sources/</argument>
               <argument>../../format/Message.fbs</argument>
+              <argument>../../format/File.fbs</argument>
             </arguments>
           </configuration>
         </execution>

java/memory/src/main/java/io/netty/buffer/ArrowBuf.java

@@ -17,8 +17,6 @@
  */
 package io.netty.buffer;
 
-import io.netty.util.internal.PlatformDependent;
-
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
@@ -30,16 +28,18 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 
+import org.apache.arrow.memory.AllocationManager.BufferLedger;
 import org.apache.arrow.memory.BaseAllocator;
+import org.apache.arrow.memory.BaseAllocator.Verbosity;
 import org.apache.arrow.memory.BoundsChecking;
 import org.apache.arrow.memory.BufferAllocator;
 import org.apache.arrow.memory.BufferManager;
-import org.apache.arrow.memory.AllocationManager.BufferLedger;
-import org.apache.arrow.memory.BaseAllocator.Verbosity;
 import org.apache.arrow.memory.util.HistoricalLog;
 
 import com.google.common.base.Preconditions;
 
+import io.netty.util.internal.PlatformDependent;
+
 public final class ArrowBuf extends AbstractByteBuf implements AutoCloseable {
   private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(ArrowBuf.class);
 
@@ -307,7 +307,7 @@ public ByteOrder order() {
   }
 
   @Override
-  public ByteBuf order(ByteOrder endianness) {
+  public ArrowBuf order(ByteOrder endianness) {
     return this;
   }
 
@@ -344,7 +344,7 @@ public ByteBuf copy(int index, int length) {
   }
 
   @Override
-  public ByteBuf slice() {
+  public ArrowBuf slice() {
     return slice(readerIndex(), readableBytes());
   }
 
@@ -467,7 +467,7 @@ public boolean equals(Object obj) {
   }
 
   @Override
-  public ByteBuf retain(int increment) {
+  public ArrowBuf retain(int increment) {
     Preconditions.checkArgument(increment > 0, "retain(%d) argument is not positive", increment);
 
     if (isEmpty) {
@@ -484,7 +484,7 @@ public ByteBuf retain(int increment) {
   }
 
   @Override
-  public ByteBuf retain() {
+  public ArrowBuf retain() {
     return retain(1);
   }
 
@@ -535,109 +535,109 @@ public short getShort(int index) {
   }
 
   @Override
-  public ByteBuf setShort(int index, int value) {
+  public ArrowBuf setShort(int index, int value) {
     chk(index, 2);
     PlatformDependent.putShort(addr(index), (short) value);
     return this;
   }
 
   @Override
-  public ByteBuf setInt(int index, int value) {
+  public ArrowBuf setInt(int index, int value) {
     chk(index, 4);
     PlatformDependent.putInt(addr(index), value);
     return this;
   }
 
   @Override
-  public ByteBuf setLong(int index, long value) {
+  public ArrowBuf setLong(int index, long value) {
     chk(index, 8);
     PlatformDependent.putLong(addr(index), value);
     return this;
   }
 
   @Override
-  public ByteBuf setChar(int index, int value) {
+  public ArrowBuf setChar(int index, int value) {
     chk(index, 2);
     PlatformDependent.putShort(addr(index), (short) value);
     return this;
   }
 
   @Override
-  public ByteBuf setFloat(int index, float value) {
+  public ArrowBuf setFloat(int index, float value) {
     chk(index, 4);
     PlatformDependent.putInt(addr(index), Float.floatToRawIntBits(value));
     return this;
   }
 
   @Override
-  public ByteBuf setDouble(int index, double value) {
+  public ArrowBuf setDouble(int index, double value) {
     chk(index, 8);
     PlatformDependent.putLong(addr(index), Double.doubleToRawLongBits(value));
     return this;
   }
 
   @Override
-  public ByteBuf writeShort(int value) {
+  public ArrowBuf writeShort(int value) {
     ensure(2);
     PlatformDependent.putShort(addr(writerIndex), (short) value);
     writerIndex += 2;
     return this;
   }
 
   @Override
-  public ByteBuf writeInt(int value) {
+  public ArrowBuf writeInt(int value) {
     ensure(4);
     PlatformDependent.putInt(addr(writerIndex), value);
     writerIndex += 4;
     return this;
   }
 
   @Override
-  public ByteBuf writeLong(long value) {
+  public ArrowBuf writeLong(long value) {
     ensure(8);
     PlatformDependent.putLong(addr(writerIndex), value);
     writerIndex += 8;
     return this;
   }
 
   @Override
-  public ByteBuf writeChar(int value) {
+  public ArrowBuf writeChar(int value) {
     ensure(2);
     PlatformDependent.putShort(addr(writerIndex), (short) value);
     writerIndex += 2;
     return this;
   }
 
   @Override
-  public ByteBuf writeFloat(float value) {
+  public ArrowBuf writeFloat(float value) {
     ensure(4);
     PlatformDependent.putInt(addr(writerIndex), Float.floatToRawIntBits(value));
     writerIndex += 4;
     return this;
   }
 
   @Override
-  public ByteBuf writeDouble(double value) {
+  public ArrowBuf writeDouble(double value) {
     ensure(8);
     PlatformDependent.putLong(addr(writerIndex), Double.doubleToRawLongBits(value));
     writerIndex += 8;
     return this;
   }
 
   @Override
-  public ByteBuf getBytes(int index, byte[] dst, int dstIndex, int length) {
+  public ArrowBuf getBytes(int index, byte[] dst, int dstIndex, int length) {
     udle.getBytes(index + offset, dst, dstIndex, length);
     return this;
   }
 
   @Override
-  public ByteBuf getBytes(int index, ByteBuffer dst) {
+  public ArrowBuf getBytes(int index, ByteBuffer dst) {
     udle.getBytes(index + offset, dst);
     return this;
   }
 
   @Override
-  public ByteBuf setByte(int index, int value) {
+  public ArrowBuf setByte(int index, int value) {
     chk(index, 1);
     PlatformDependent.putByte(addr(index), (byte) value);
     return this;
@@ -699,13 +699,13 @@ protected void _setLong(int index, long value) {
   }
 
   @Override
-  public ByteBuf getBytes(int index, ByteBuf dst, int dstIndex, int length) {
+  public ArrowBuf getBytes(int index, ByteBuf dst, int dstIndex, int length) {
     udle.getBytes(index + offset, dst, dstIndex, length);
     return this;
   }
 
   @Override
-  public ByteBuf getBytes(int index, OutputStream out, int length) throws IOException {
+  public ArrowBuf getBytes(int index, OutputStream out, int length) throws IOException {
     udle.getBytes(index + offset, out, length);
     return this;
   }
@@ -724,12 +724,12 @@ public int getBytes(int index, GatheringByteChannel out, int length) throws IOEx
   }
 
   @Override
-  public ByteBuf setBytes(int index, ByteBuf src, int srcIndex, int length) {
+  public ArrowBuf setBytes(int index, ByteBuf src, int srcIndex, int length) {
     udle.setBytes(index + offset, src, srcIndex, length);
     return this;
   }
 
-  public ByteBuf setBytes(int index, ByteBuffer src, int srcIndex, int length) {
+  public ArrowBuf setBytes(int index, ByteBuffer src, int srcIndex, int length) {
     if (src.isDirect()) {
       checkIndex(index, length);
       PlatformDependent.copyMemory(PlatformDependent.directBufferAddress(src) + srcIndex, this.memoryAddress() + index,
@@ -749,13 +749,13 @@ public ByteBuf setBytes(int index, ByteBuffer src, int srcIndex, int length) {
   }
 
   @Override
-  public ByteBuf setBytes(int index, byte[] src, int srcIndex, int length) {
+  public ArrowBuf setBytes(int index, byte[] src, int srcIndex, int length) {
     udle.setBytes(index + offset, src, srcIndex, length);
     return this;
   }
 
   @Override
-  public ByteBuf setBytes(int index, ByteBuffer src) {
+  public ArrowBuf setBytes(int index, ByteBuffer src) {
     udle.setBytes(index + offset, src);
     return this;
   }
@@ -860,4 +860,17 @@ public void print(StringBuilder sb, int indent, Verbosity verbosity) {
     }
   }
 
+  @Override
+  public ArrowBuf readerIndex(int readerIndex) {
+    super.readerIndex(readerIndex);
+    return this;
+  }
+
+  @Override
+  public ArrowBuf writerIndex(int writerIndex) {
+    super.writerIndex(writerIndex);
+    return this;
+  }
+
+
 }

java/vector/src/main/codegen/data/ArrowTypes.tdd

@@ -30,15 +30,15 @@
     },
     {
       name: "Union",
-      fields: []
+      fields: [{name: "mode", type: short}]
     },
     {
       name: "Int",
       fields: [{name: "bitWidth", type: int}, {name: "isSigned", type: boolean}]
     },
     {
       name: "FloatingPoint",
-      fields: [{name: precision, type: int}]
+      fields: [{name: precision, type: short}]
     },
     {
       name: "Utf8",