ARROW-264: File format

Author: julienledem

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;

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/vector/src/main/codegen/data/ArrowTypes.tdd

@@ -30,7 +30,7 @@
     },
     {
       name: "Union",
-      fields: []
+      fields: [{name: "mode", type: int}]
     },
     {
       name: "Int",

java/vector/src/main/codegen/templates/ArrowType.java

@@ -24,9 +24,8 @@
 
 <@pp.dropOutputFile />
 <@pp.changeOutputFile name="/org/apache/arrow/vector/types/pojo/ArrowType.java" />
-
-
 <#include "/@includes/license.ftl" />
+
 package org.apache.arrow.vector.types.pojo;
 
 import com.google.flatbuffers.FlatBufferBuilder;
@@ -38,7 +37,13 @@ public abstract class ArrowType {
 
   public abstract byte getTypeType();
   public abstract int getType(FlatBufferBuilder builder);
+  public abstract <T> T accept(ArrowTypeVisitor<T> visitor);
 
+  public static interface ArrowTypeVisitor<T> {
+  <#list arrowTypes.types as type>
+    T visit(${type.name} type); 
+  </#list>
+  }
 
   <#list arrowTypes.types as type>
   <#assign name = type.name>
@@ -102,6 +107,11 @@ public boolean equals(Object obj) {
       </#list>
       </#if>
     }
+
+    @Override
+    public <T> T accept(ArrowTypeVisitor<T> visitor) {
+      return visitor.visit(this);
+    }
   }
   </#list>
 

java/vector/src/main/codegen/templates/UnionVector.java

@@ -43,6 +43,8 @@
 import org.apache.arrow.vector.complex.impl.ComplexCopier;
 import org.apache.arrow.vector.util.CallBack;
 
+import static org.apache.arrow.flatbuf.UnionMode.Sparse;
+
 /*
  * This class is generated using freemarker and the ${.template_name} template.
  */
@@ -203,7 +205,7 @@ public Field getField() {
     for (ValueVector v : internalMap.getChildren()) {
       childFields.add(v.getField());
     }
-    return new Field(name, true, new ArrowType.Union(), childFields);
+    return new Field(name, true, new ArrowType.Union(Sparse), childFields);
   }
 
   @Override

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

@@ -0,0 +1,82 @@
+/**
+ * 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.arrow.file;
+
+import org.apache.arrow.flatbuf.Block;
+import org.apache.arrow.schema.FBSerializable;
+
+import com.google.flatbuffers.FlatBufferBuilder;
+
+public class ArrowBlock implements FBSerializable {
+
+  private final long offset;
+  private final int metadataLength;
+  private final long bodyLength;
+
+  public ArrowBlock(long offset, int metadataLength, long bodyLength) {
+    super();
+    this.offset = offset;
+    this.metadataLength = metadataLength;
+    this.bodyLength = bodyLength;
+  }
+
+  public long getOffset() {
+    return offset;
+  }
+
+  public int getMetadataLength() {
+    return metadataLength;
+  }
+
+  public long getBodyLength() {
+    return bodyLength;
+  }
+
+  @Override
+  public int writeTo(FlatBufferBuilder builder) {
+    return Block.createBlock(builder, offset, metadataLength, bodyLength);
+  }
+
+  @Override
+  public int hashCode() {
+    final int prime = 31;
+    int result = 1;
+    result = prime * result + (int) (bodyLength ^ (bodyLength >>> 32));
+    result = prime * result + metadataLength;
+    result = prime * result + (int) (offset ^ (offset >>> 32));
+    return result;
+  }
+
+  @Override
+  public boolean equals(Object obj) {
+    if (this == obj)
+      return true;
+    if (obj == null)
+      return false;
+    if (getClass() != obj.getClass())
+      return false;
+    ArrowBlock other = (ArrowBlock) obj;
+    if (bodyLength != other.bodyLength)
+      return false;
+    if (metadataLength != other.metadataLength)
+      return false;
+    if (offset != other.offset)
+      return false;
+    return true;
+  }
+}

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

@@ -0,0 +1,144 @@
+/**
+ * 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.arrow.file;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.arrow.flatbuf.Block;
+import org.apache.arrow.flatbuf.Footer;
+import org.apache.arrow.schema.FBSerializable;
+import org.apache.arrow.vector.types.pojo.Schema;
+
+import com.google.flatbuffers.FlatBufferBuilder;
+
+public class ArrowFooter implements FBSerializable {
+
+  private final Schema schema;
+
+  private final List<ArrowBlock> dictionaries;
+
+  private final List<ArrowBlock> recordBatches;
+
+  public ArrowFooter(Schema schema, List<ArrowBlock> dictionaries, List<ArrowBlock> recordBatches) {
+    super();
+    this.schema = schema;
+    this.dictionaries = dictionaries;
+    this.recordBatches = recordBatches;
+  }
+
+  public ArrowFooter(Footer footer) {
+    this(
+        Schema.convertSchema(footer.schema()),
+        dictionaries(footer),
+        recordBatches(footer)
+        );
+  }
+
+  private static List<ArrowBlock> recordBatches(Footer footer) {
+    List<ArrowBlock> recordBatches = new ArrayList<>();
+    Block tempBLock = new Block();
+    int recordBatchesLength = footer.recordBatchesLength();
+    for (int i = 0; i < recordBatchesLength; i++) {
+      Block block = footer.recordBatches(tempBLock, i);
+      recordBatches.add(new ArrowBlock(block.offset(), block.metaDataLength(), block.bodyLength()));
+    }
+    return recordBatches;
+  }
+
+  private static List<ArrowBlock> dictionaries(Footer footer) {
+    List<ArrowBlock> dictionaries = new ArrayList<>();
+    Block tempBLock = new Block();
+    int dictionariesLength = footer.dictionariesLength();
+    for (int i = 0; i < dictionariesLength; i++) {
+      Block block = footer.dictionaries(tempBLock, i);
+      dictionaries.add(new ArrowBlock(block.offset(), block.metaDataLength(), block.bodyLength()));
+    }
+    return dictionaries;
+  }
+
+  public Schema getSchema() {
+    return schema;
+  }
+
+  public List<ArrowBlock> getDictionaries() {
+    return dictionaries;
+  }
+
+  public List<ArrowBlock> getRecordBatches() {
+    return recordBatches;
+  }
+
+  @Override
+  public int writeTo(FlatBufferBuilder builder) {
+    int schemaIndex = schema.getSchema(builder);
+    Footer.startDictionariesVector(builder, dictionaries.size());
+    int dicsOffset = endVector(builder, dictionaries);
+    Footer.startRecordBatchesVector(builder, recordBatches.size());
+    int rbsOffset = endVector(builder, recordBatches);
+    Footer.startFooter(builder);
+    Footer.addSchema(builder, schemaIndex);
+    Footer.addDictionaries(builder, dicsOffset);
+    Footer.addRecordBatches(builder, rbsOffset);
+    return Footer.endFooter(builder);
+  }
+
+  private int endVector(FlatBufferBuilder builder, List<ArrowBlock> blocks) {
+    for (ArrowBlock block : blocks) {
+      block.writeTo(builder);
+    }
+    return builder.endVector();
+  }
+
+  @Override
+  public int hashCode() {
+    final int prime = 31;
+    int result = 1;
+    result = prime * result + ((dictionaries == null) ? 0 : dictionaries.hashCode());
+    result = prime * result + ((recordBatches == null) ? 0 : recordBatches.hashCode());
+    result = prime * result + ((schema == null) ? 0 : schema.hashCode());
+    return result;
+  }
+
+  @Override
+  public boolean equals(Object obj) {
+    if (this == obj)
+      return true;
+    if (obj == null)
+      return false;
+    if (getClass() != obj.getClass())
+      return false;
+    ArrowFooter other = (ArrowFooter) obj;
+    if (dictionaries == null) {
+      if (other.dictionaries != null)
+        return false;
+    } else if (!dictionaries.equals(other.dictionaries))
+      return false;
+    if (recordBatches == null) {
+      if (other.recordBatches != null)
+        return false;
+    } else if (!recordBatches.equals(other.recordBatches))
+      return false;
+    if (schema == null) {
+      if (other.schema != null)
+        return false;
+    } else if (!schema.equals(other.schema))
+      return false;
+    return true;
+  }
+}