/**
 * Copyright 2013 Criteo.
 *
 * Licensed 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 parquet.hive.writable;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

import org.apache.hadoop.io.Writable;

import parquet.io.api.Binary;
/**
 *
 * A Parquet InputFormat for Hive (with the deprecated package mapred)
 *
 *
 * @author Mickaël Lacour <m.lacour@criteo.com>
 * @author Rémy Pecqueur <r.pecqueur@criteo.com>
 *
 */
public class BinaryWritable implements Writable {

  private byte[] bytes;

  public BinaryWritable(final Binary binary) {
    bytes = binary.getBytes();
  }

  public BinaryWritable(final String string) {
    bytes = string.getBytes();
  }

  @Override
  public void readFields(final DataInput input) throws IOException {
    final int size = input.readInt();

    // Define a new byte of array of the exact size of the payload
    final byte[] bytes = new byte[size];
    input.readFully(bytes);
  }

  @Override
  public void write(final DataOutput output) throws IOException {
    if (bytes != null) {
      output.writeInt(bytes.length);
      output.write(bytes);
    }
  }

  public byte[] getBytes() {
    return bytes;
  }

  public void setBytes(final byte[] bytes) {
    this.bytes = bytes;
  }
}

@@ -694,16 +694,84 @@
 eption;%0A
+import java.util.Arrays;%0A%0Aimport org.apache.hadoop.io.BytesWritable;
 %0Aimport 
@@ -1382,16 +1382,38 @@
 Int();%0A%0A
+    if (size %3E 0) %7B%0A  
     // D
@@ -1471,24 +1471,26 @@
 payload%0A    
+  
 final byte%5B%5D
@@ -1514,16 +1514,18 @@
 %5Bsize%5D;%0A
+  
     inpu
@@ -1541,18 +1541,33 @@
 ly(bytes
-);
+, 0, size);%0A    %7D
 %0A  %7D%0A%0A  
@@ -1728,16 +1728,72 @@
 te(bytes
+, 0, bytes.length);%0A    %7D else %7B%0A      output.writeInt(0
 );%0A    %7D
@@ -1923,10 +1923,687 @@
 es;%0A  %7D%0A
+%0A  @Override%0A  public boolean equals(final Object obj) %7B%0A    if (obj == null) %7B%0A      return false;%0A    %7D%0A%0A    if (this == obj) %7B%0A      return true;%0A    %7D%0A%0A    if (obj instanceof BinaryWritable) %7B%0A      final BinaryWritable bWrit = (BinaryWritable) obj;%0A      final byte%5B%5D otherBytes = bWrit.getBytes();%0A      final int thisLen = bytes == null ? 0 : bytes.length;%0A      final int otherLen = otherBytes == null ? 0 : otherBytes.length;%0A      return BytesWritable.Comparator.compareBytes(bytes, 0, thisLen,%0A          otherBytes, 0, otherLen) == 0;%0A    %7D%0A%0A    return false;%0A  %7D%0A%0A  @Override%0A  public int hashCode() %7B%0A    return (bytes == null) ? 31 : Arrays.hashCode(bytes);%0A  %7D%0A%0A
 %7D%0A

Add equals and hashcode methods to BinaryWritable