PARQUET-415: Fix ByteBuffer Binary serialization.

parquet-column/src/main/java/org/apache/parquet/io/api/Binary.java

@@ -355,9 +355,9 @@ public static Binary fromByteArray(final byte[] value) {
 
   private static class ByteBufferBackedBinary extends Binary {
     private ByteBuffer value;
-    private byte[] cachedBytes;
-    private final int offset;
-    private final int length;
+    private transient byte[] cachedBytes;
+    private int offset;
+    private int length;
 
     public ByteBufferBackedBinary(ByteBuffer value, int offset, int length, boolean isBackingBytesReused) {
       this.value = value;
@@ -502,6 +502,8 @@ private void readObject(java.io.ObjectInputStream in) throws IOException, ClassN
       byte[] bytes = new byte[length];
       in.readFully(bytes, 0, length);
       this.value = ByteBuffer.wrap(bytes);
+      this.offset = 0;
+      this.length = length;
     }
 
     private void readObjectNoData() throws ObjectStreamException {

parquet-column/src/test/java/org/apache/parquet/io/api/TestBinary.java

@@ -18,8 +18,11 @@
  */
 package org.apache.parquet.io.api;
 
+import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
 import java.nio.ByteBuffer;
 import java.util.Arrays;
 
@@ -29,6 +32,7 @@
 import static org.junit.Assert.assertArrayEquals;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
 
 public class TestBinary {
 
@@ -227,6 +231,22 @@ private void testReusedCopy(BinaryFactory bf) throws Exception {
     assertArrayEquals(testString.getBytes(UTF8), copy.copy().getBytes());
   }
 
+  private void testSerializable(BinaryFactory bf, boolean reused) throws Exception {
+    BinaryAndOriginal bao = bf.get("polygon".getBytes(UTF8), reused);
+
+    ByteArrayOutputStream baos = new ByteArrayOutputStream();
+    ObjectOutputStream out = new ObjectOutputStream(baos);
+    out.writeObject(bao.binary);
+    out.close();
+    baos.close();
+
+    ObjectInputStream in = new ObjectInputStream(new ByteArrayInputStream(
+        baos.toByteArray()));
+    Object object = in.readObject();
+    assertTrue(object instanceof Binary);
+    assertEquals(bao.binary, object);
+  }
+
   private void testBinary(BinaryFactory bf, boolean reused) throws Exception {
     testSlice(bf, reused);
 
@@ -236,5 +256,6 @@ private void testBinary(BinaryFactory bf, boolean reused) throws Exception {
       testConstantCopy(bf);
     }
 
+    testSerializable(bf, reused);
   }
 }