gt:                     currentByte = in.read();
context: /*
   * JBoss, Home of Professional Open Source.
   * See the COPYRIGHT.txt file distributed with this work for information
   * regarding copyright ownership.  Some portions may be licensed
   * to Red Hat, Inc. under one or more contributor license agreements.
   *
   * This library is free software; you can redistribute it and/or
   * modify it under the terms of the GNU Lesser General Public
   * License as published by the Free Software Foundation; either
   * version 2.1 of the License, or (at your option) any later version.
   *
   * This library is distributed in the hope that it will be useful,
   * but WITHOUT ANY WARRANTY; without even the implied warranty of
   * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
   * Lesser General Public License for more details.
   *
   * You should have received a copy of the GNU Lesser General Public
   * License along with this library; if not, write to the Free Software
   * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
   * 02110-1301 USA.
   */
  
  package org.teiid.client;
  
  import java.io.Externalizable;
  import java.io.IOException;
  import java.io.ObjectInput;
  import java.io.ObjectOutput;
  import java.io.ObjectStreamConstants;
  import java.math.BigDecimal;
  import java.math.BigInteger;
  import java.sql.Array;
  import java.sql.SQLException;
  import java.sql.Time;
  import java.sql.Timestamp;
  import java.util.ArrayList;
  import java.util.Arrays;
  import java.util.Calendar;
  import java.util.HashMap;
  import java.util.List;
  import java.util.Map;
  import java.util.TimeZone;
  
  import org.teiid.core.TeiidRuntimeException;
  import org.teiid.core.types.ArrayImpl;
  import org.teiid.core.types.BinaryType;
  import org.teiid.core.types.BlobType;
  import org.teiid.core.types.ClobType;
  import org.teiid.core.types.DataTypeManager;
  import org.teiid.core.types.GeometryType;
  import org.teiid.core.types.XMLType;
  import org.teiid.jdbc.JDBCPlugin;
  
  
  
  /**
   * @since 4.2
   *
   * <ul>
   * <li>version 0: starts with 7.1 and uses simple serialization too broadly
   * <li>version 1: starts with 8.0 uses better string, blob, clob, xml, etc.
   *   add varbinary support.
   *   however was possibly silently truncating date/time values that were
   *   outside of jdbc allowed values
   * <li>version 2: starts with 8.2 and adds better array serialization and
   *   uses a safer date/time serialization
   * <li>version 3: starts with 8.6 and adds better repeated string performance
   * <li>version 4: starts with 8.10 and adds the geometry type
   * </ul>
   */
  public class BatchSerializer {
  
  	public static final byte VERSION_GEOMETRY = (byte)4;
      static final byte CURRENT_VERSION = VERSION_GEOMETRY;
  
  	private BatchSerializer() {} // Uninstantiable
  
      private static ColumnSerializer defaultSerializer = new ColumnSerializer();
  
      private static final Map<String, ColumnSerializer[]> serializers = new HashMap<String, ColumnSerializer[]>(128);
      static {
          serializers.put(DataTypeManager.DefaultDataTypes.BIG_DECIMAL,   new ColumnSerializer[] {new BigDecimalColumnSerializer()});
          serializers.put(DataTypeManager.DefaultDataTypes.BIG_INTEGER,   new ColumnSerializer[] {new BigIntegerColumnSerializer()});
          serializers.put(DataTypeManager.DefaultDataTypes.BOOLEAN,       new ColumnSerializer[] {new BooleanColumnSerializer()});
          serializers.put(DataTypeManager.DefaultDataTypes.BYTE,          new ColumnSerializer[] {new ByteColumnSerializer()});
          serializers.put(DataTypeManager.DefaultDataTypes.CHAR,          new ColumnSerializer[] {new CharColumnSerializer()});
          serializers.put(DataTypeManager.DefaultDataTypes.DATE,          new ColumnSerializer[] {new DateColumnSerializer(), new DateColumnSerializer1(), new DateColumnSerializer()});
          serializers.put(DataTypeManager.DefaultDataTypes.DOUBLE,        new ColumnSerializer[] {new DoubleColumnSerializer()});
          serializers.put(DataTypeManager.DefaultDataTypes.FLOAT,         new ColumnSerializer[] {new FloatColumnSerializer()});
          serializers.put(DataTypeManager.DefaultDataTypes.INTEGER,       new ColumnSerializer[] {new IntColumnSerializer()});
          serializers.put(DataTypeManager.DefaultDataTypes.LONG,          new ColumnSerializer[] {new LongColumnSerializer()});
          serializers.put(DataTypeManager.DefaultDataTypes.SHORT,         new ColumnSerializer[] {new ShortColumnSerializer()});
          serializers.put(DataTypeManager.DefaultDataTypes.TIME,          new ColumnSerializer[] {new TimeColumnSerializer(), new TimeColumnSerializer1(), new TimeColumnSerializer()});
          serializers.put(DataTypeManager.DefaultDataTypes.TIMESTAMP,     new ColumnSerializer[] {new TimestampColumnSerializer()});
          serializers.put(DataTypeManager.DefaultDataTypes.STRING,     	new ColumnSerializer[] {defaultSerializer, new StringColumnSerializer1(), new StringColumnSerializer1(), new StringColumnSerializer3()});
          serializers.put(DataTypeManager.DefaultDataTypes.CLOB,  	   	new ColumnSerializer[] {defaultSerializer, new ClobColumnSerializer1()});
          serializers.put(DataTypeManager.DefaultDataTypes.BLOB,     		new ColumnSerializer[] {defaultSerializer, new BlobColumnSerializer1()});
          serializers.put(DataTypeManager.DefaultDataTypes.GEOMETRY,     	new ColumnSerializer[] {defaultSerializer, new GeometryColumnSerializer()});
          serializers.put(DataTypeManager.DefaultDataTypes.XML,     		new ColumnSerializer[] {defaultSerializer, new XmlColumnSerializer1()});
          serializers.put(DataTypeManager.DefaultDataTypes.NULL,     		new ColumnSerializer[] {defaultSerializer, new NullColumnSerializer1()});
          serializers.put(DataTypeManager.DefaultDataTypes.OBJECT,     	new ColumnSerializer[] {defaultSerializer, new ObjectColumnSerializer((byte)1)});
          serializers.put(DataTypeManager.DefaultDataTypes.VARBINARY,    	new ColumnSerializer[] {new BinaryColumnSerializer(), new BinaryColumnSerializer1()});
      }
  
      private static ColumnSerializer arrayColumnSerializer = new ColumnSerializer() {
  
      	@Override
      	protected void writeObject(ObjectOutput out, Object obj, Map<Object, Integer> cache, byte version)
      			throws IOException {
      		try {
  				super.writeObject(out, ((java.sql.Array)obj).getArray(), cache, version);
  			} catch (SQLException e) {
  				throw new IOException(e);
  			}
      	}
  
      	@Override
      	protected Object readObject(ObjectInput in, List<Object> cache, byte version) throws IOException,
      			ClassNotFoundException {
      		return new ArrayImpl((Object[]) in.readObject());
      	}
  
      };
  
      private static final ColumnSerializer arrayColumnSerialier2 = new ArrayColumnSerializer2(new ObjectColumnSerializer((byte)2));
  
  	private static final class ArrayColumnSerializer2 extends ColumnSerializer {
  
  		ObjectColumnSerializer ser;
  
  		public ArrayColumnSerializer2(ObjectColumnSerializer ser) {
  			this.ser = ser;
  		}
  
  		@Override
      	protected void writeObject(ObjectOutput out, Object obj, Map<Object, Integer> cache, byte version)
      			throws IOException {
  			Object[] values = null;
      		try {
      			values = (Object[]) ((Array)obj).getArray();
      		} catch (SQLException e) {
      			out.writeInt(-1);
      			return;
      		}
  			out.writeInt(values.length);
  			int code = DataTypeManager.getTypeCode(values.getClass().getComponentType());
  			if (code == DataTypeManager.DefaultTypeCodes.GEOMETRY && version < VERSION_GEOMETRY) {
  				code = DataTypeManager.DefaultTypeCodes.BLOB;
  			}
      		out.writeByte((byte)code);
      		for (int i = 0; i < values.length;) {
      			writeIsNullData(out, i, values);
      			int end = Math.min(values.length, i+8);
      			for (; i < end; i++) {
      				if (values[i] != null) {
  						this.ser.writeObject(out, values[i], code, cache, version);
  					}
      			}
      		}
      		out.writeBoolean((obj instanceof ArrayImpl && ((ArrayImpl)obj).isZeroBased()));
      	}
  
  		@Override
      	protected Object readObject(ObjectInput in, List<Object> cache, byte version) throws IOException,
      			ClassNotFoundException {
      		int length = in.readInt();
      		if (length == -1) {
          		return new ArrayImpl(null);
      		}
      		int code = in.readByte();
      		Object[] vals = (Object[])java.lang.reflect.Array.newInstance(DataTypeManager.getClass(code), length);
      		for (int i = 0; i < length;) {
      			byte b = in.readByte();
      			int end = Math.min(length, i+8);
      			for (; i < end; i++) {
  					if (!isNullObject(i, b)) {
  						vals[i] = this.ser.readObject(in, cache, code, version);
  					}
      			}
      		}
      		ArrayImpl result = new ArrayImpl(vals);
      		result.setZeroBased(in.readBoolean());
      		return result;
      	}
  		
  		@Override
  		public boolean usesCache(byte version) {
  			return version >= 3;
  		}
  	}
  
  	static class BinaryColumnSerializer1 extends ColumnSerializer {
  		@Override
  		protected void writeObject(ObjectOutput out, Object obj, Map<Object, Integer> cache, byte version)
  				throws IOException {
  			byte[] bytes = ((BinaryType)obj).getBytes();
  			out.writeInt(bytes.length); //in theory this could be a short, but we're not strictly enforcing the length
  			out.write(bytes);
  		}
  
  		@Override
  		protected Object readObject(ObjectInput in, List<Object> cache, byte version) throws IOException,
  				ClassNotFoundException {
  			int length = in.readInt();
  			byte[] bytes = new byte[length];
  			in.readFully(bytes);
  			return new BinaryType(bytes);
  		}
  	}
  
      static class BinaryColumnSerializer extends ColumnSerializer {
  		@Override
  		protected void writeObject(ObjectOutput out, Object obj, Map<Object, Integer> cache, byte version)
  				throws IOException {
  			//uses object serialization for compatibility with legacy clients
  			super.writeObject(out, ((BinaryType)obj).getBytesDirect(), cache, version);
  		}
  
  		@Override
  		protected Object readObject(ObjectInput in, List<Object> cache, byte version) throws IOException,
  				ClassNotFoundException {
  			//won't actually be used
  			byte[] bytes = (byte[])super.readObject(in, cache, version);
  			return new BinaryType(bytes);
  		}
  	}
  
  	public static final class ObjectColumnSerializer extends ColumnSerializer {
  
  		byte defaultVersion;
  
      	public ObjectColumnSerializer(byte version) {
      		this.defaultVersion = version;
  		}
  
  		@Override
      	protected void writeObject(ObjectOutput out, Object obj, Map<Object, Integer> cache, byte version)
      			throws IOException {
      		int code = DataTypeManager.getTypeCode(obj.getClass());
      		if (code == DataTypeManager.DefaultTypeCodes.GEOMETRY && version < VERSION_GEOMETRY) {
  				code = DataTypeManager.DefaultTypeCodes.BLOB;
  			}
      		out.writeByte((byte)code);
      		writeObject(out, obj, code, cache, version<VERSION_GEOMETRY?this.defaultVersion:version);
      	}
  
  		protected void writeObject(ObjectOutput out, Object obj, int code, Map<Object, Integer> cache, byte effectiveVersion)
  				throws IOException {
  			if (code == DataTypeManager.DefaultTypeCodes.BOOLEAN) {
      			if (Boolean.TRUE.equals(obj)) {
      				out.write((byte)1);
      			} else {
      				out.write((byte)0);
      			}
      		} else if (code == DataTypeManager.DefaultTypeCodes.OBJECT) {
      			super.writeObject(out, obj, cache, effectiveVersion);
      		} else {
      			String name = DataTypeManager.getDataTypeName(obj.getClass());
      			ColumnSerializer s = getSerializer(name, effectiveVersion);
      			s.writeObject(out, obj, cache, effectiveVersion);
      		}
      	}
  
      	@Override
      	protected Object readObject(ObjectInput in, List<Object> cache, byte version) throws IOException,
      			ClassNotFoundException {
      		int code = in.readByte();
      		return readObject(in, cache, code, version<VERSION_GEOMETRY?this.defaultVersion:version);
      	}
  
  		private Object readObject(ObjectInput in, List<Object> cache, int code, byte effectiveVersion) throws IOException,
  				ClassNotFoundException {
  			if (code == DataTypeManager.DefaultTypeCodes.BOOLEAN) {
      			if (in.readByte() == (byte)0) {
      				return Boolean.FALSE;
      			}
      			return Boolean.TRUE;
      		}
      		if (code != DataTypeManager.DefaultTypeCodes.OBJECT) {
      			ColumnSerializer s = getSerializer(DataTypeManager.getDataTypeName(DataTypeManager.getClass(code)), effectiveVersion);
      			return s.readObject(in, cache, effectiveVersion);
      		}
  			return super.readObject(in, cache, effectiveVersion);
      	}
  		
  		@Override
  		public boolean usesCache(byte version) {
  			return version >= 3;
  		}
  
      }
  
      private static final int MAX_UTF = 0xFFFF/3; //this is greater than the expected max length of Teiid Strings
  
      private static class StringColumnSerializer1 extends ColumnSerializer {
      	@Override
      	protected void writeObject(ObjectOutput out, Object obj, Map<Object, Integer> cache, byte version) throws IOException {
      		String str = (String)obj;
          	if (str.length() <= MAX_UTF) {
          		//skip object serialization if we have a short string
          	    out.writeByte(ObjectStreamConstants.TC_STRING);
          	    out.writeUTF(str);
          	} else {
          		out.writeByte(ObjectStreamConstants.TC_LONGSTRING);
          		out.writeObject(obj);
          	}
          }
  
      	@Override
      	protected Object readObject(ObjectInput in, List<Object> cache, byte version) throws IOException,
      			ClassNotFoundException {
      		if (in.readByte() == ObjectStreamConstants.TC_STRING) {
      			return in.readUTF();
      		}
      		return super.readObject(in, cache, version);
      	}
      	
      }
      
      private static class StringColumnSerializer3 extends StringColumnSerializer1 {
      	private static final int MAX_INLINE_STRING_LENGTH = 5;
      	private static final byte REPEATED_STRING = 0;
      	@Override
      	protected Object readObject(ObjectInput in, List<Object> cache, byte version)
      			throws IOException, ClassNotFoundException {
      		byte b = in.readByte();
      		if (b == ObjectStreamConstants.TC_STRING) {
      			String val = in.readUTF();
      			if (val.length() > MAX_INLINE_STRING_LENGTH) {
      				cache.add(val);
      			}
      			return val;
      		}
      		if (b == REPEATED_STRING) {
      			Integer val = in.readInt();
      			return cache.get(val);
      		}
      		String val = (String) in.readObject();
      		if (val.length() > MAX_INLINE_STRING_LENGTH) {
  				cache.add(val);
  			}
      		return val;
      	}
      	
      	@Override
      	protected void writeObject(ObjectOutput out, Object obj,
      			Map<Object, Integer> cache, byte version) throws IOException {
      		String str = (String)obj;
      		Integer val = cache.get(str);
      		if (val != null) {
      			out.writeByte(REPEATED_STRING);
      			out.writeInt(val);
      			return;
      		} 
      		if (str.length() > MAX_INLINE_STRING_LENGTH) {
      			cache.put(str, cache.size());
      		}
      		super.writeObject(out, obj, cache, version);
      	}
      	
      	@Override
      	public boolean usesCache(byte version) {
      		return true;
      	}
      }
  
      private static class NullColumnSerializer1 extends ColumnSerializer {
      	@Override
      	public void writeColumn(ObjectOutput out, int col,
      			List<? extends List<?>> batch, Map<Object, Integer> cache, byte version) throws IOException {
      	}
  
      	@Override
      	public void readColumn(ObjectInput in, int col,
      			List<List<Object>> batch, byte[] isNull, List<Object> cache, byte version) throws IOException,
      			ClassNotFoundException {
      	}
      }
  
      private static class ClobColumnSerializer1 extends ColumnSerializer {
          @Override
  		protected void writeObject(ObjectOutput out, Object obj, Map<Object, Integer> cache, byte version) throws IOException {
          	((Externalizable)obj).writeExternal(out);
          }
          @Override
  		protected Object readObject(ObjectInput in, List<Object> cache, byte version) throws IOException, ClassNotFoundException {
          	ClobType ct = new ClobType();
          	ct.readExternal(in);
              return ct;
          }
      }
  
      private static class BlobColumnSerializer1 extends ColumnSerializer {
          @Override
  		protected void writeObject(ObjectOutput out, Object obj, Map<Object, Integer> cache, byte version) throws IOException {
          	((Externalizable)obj).writeExternal(out);
          }
          @Override
  		protected Object readObject(ObjectInput in, List<Object> cache, byte version) throws IOException, ClassNotFoundException {
          	BlobType bt = new BlobType();
          	bt.readExternal(in);
              return bt;
          }
      }
      
      private static class GeometryColumnSerializer extends ColumnSerializer {
          @Override
  		protected void writeObject(ObjectOutput out, Object obj, Map<Object, Integer> cache, byte version) throws IOException {
      		((Externalizable)obj).writeExternal(out);
          }
          @Override
  		protected Object readObject(ObjectInput in, List<Object> cache, byte version) throws IOException, ClassNotFoundException {
          	if (version < 4) {
              	BlobType bt = new BlobType();
              	bt.readExternal(in);
                  return bt;
          	}
          	GeometryType bt = new GeometryType();
          	bt.readExternal(in);
              return bt;
          }
      }
  
      private static class XmlColumnSerializer1 extends ColumnSerializer {
          @Override
  		protected void writeObject(ObjectOutput out, Object obj, Map<Object, Integer> cache, byte version) throws IOException {
          	((XMLType)obj).writeExternal(out, (byte) 1);
          }
          @Override
  		protected Object readObject(ObjectInput in, List<Object> cache, byte version) throws IOException, ClassNotFoundException {
          	XMLType xt = new XMLType();
          	xt.readExternal(in, (byte)1);
              return xt;
          }
      }
  
      /**
       * Packs the (boolean) information about whether data values in the column are null
       * into bytes so that we send ~n/8 instead of n bytes.
       * @param out
       * @param col
       * @param batch
       * @throws IOException
       * @since 4.2
       */
      static void writeIsNullData(ObjectOutput out, int col, List<? extends List<?>> batch) throws IOException {
          int numBytes = batch.size() / 8, row = 0, currentByte = 0;
          for (int byteNum = 0; byteNum < numBytes; byteNum++, row+=8) {
              currentByte  = (batch.get(row).get(col) == null) ? 0x80 : 0;
              if (batch.get(row+1).get(col) == null) {
  				currentByte |= 0x40;
  			}
              if (batch.get(row+2).get(col) == null) {
  				currentByte |= 0x20;
  			}
              if (batch.get(row+3).get(col) == null) {
  				currentByte |= 0x10;
  			}
              if (batch.get(row+4).get(col) == null) {
  				currentByte |= 0x08;
  			}
              if (batch.get(row+5).get(col) == null) {
  				currentByte |= 0x04;
  			}
              if (batch.get(row+6).get(col) == null) {
  				currentByte |= 0x02;
  			}
              if (batch.get(row+7).get(col) == null) {
  				currentByte |= 0x01;
  			}
              out.write(currentByte);
          }
          if (batch.size() % 8 > 0) {
              currentByte = 0;
              for (int mask = 0x80; row < batch.size(); row++, mask >>= 1) {
                  if (batch.get(row).get(col) == null) {
  					currentByte |= mask;
  				}
              }
              out.write(currentByte);
          }
      }
  
      static void writeIsNullData(ObjectOutput out, int offset, Object[] batch) throws IOException {
          int currentByte = 0;
          for (int mask = 0x80; offset < batch.length; offset++, mask >>= 1) {
              if (batch[offset] == null) {
  				currentByte |= mask;
  			}
          }
          out.write(currentByte);
      }
  
      /**
       * Reads the isNull data into a byte array
       * @param in
       * @param isNullBytes
       * @throws IOException
       * @since 4.2
       */
      static void readIsNullData(ObjectInput in, byte[] isNullBytes) throws IOException {
          for (int i = 0; i < isNullBytes.length; i++) {
              isNullBytes[i] = in.readByte();
          }
      }
  
      /**
       * Gets whether a data value is null based on a packed byte array containing boolean data
       * @param isNull
       * @param row
       * @return
       * @since 4.2
       */
      static final boolean isNullObject(byte[] isNull, int row) {
          //              byte number           mask     bits to shift mask
          return (isNull [ row / 8 ]         & (0x01 << (7 - (row % 8))))   != 0;
      }
  
  	private static final boolean isNullObject(int row, byte b) {
  		return (b         & (0x01 << (7 - (row % 8))))   != 0;
  	}
  
      /**
       * An abstract serializer for native types
       * @since 4.2
       */
      private static class ColumnSerializer {
          public void writeColumn(ObjectOutput out, int col, List<? extends List<?>> batch, Map<Object, Integer> cache, byte version) throws IOException {
              writeIsNullData(out, col, batch);
              Object obj = null;
              for (int i = 0; i < batch.size(); i++) {
                  obj = batch.get(i).get(col);
                  if (obj != null) {
                      writeObject(out, obj, cache, version);
                  }
              }
          }
  
          public void readColumn(ObjectInput in, int col, List<List<Object>> batch, byte[] isNull, List<Object> cache, byte version) throws IOException, ClassNotFoundException {
              readIsNullData(in, isNull);
              for (int i = 0; i < batch.size(); i++) {
                  if (!isNullObject(isNull, i)) {
                      batch.get(i).set(col, DataTypeManager.getCanonicalValue(readObject(in, cache, version)));
                  }
              }
          }
  
          protected void writeObject(ObjectOutput out, Object obj, Map<Object, Integer> cache, byte version) throws IOException {
          	out.writeObject(obj);
          }
          protected Object readObject(ObjectInput in, List<Object> cache, byte version) throws IOException, ClassNotFoundException {
          	return in.readObject();
          }
          
          public boolean usesCache(byte version) {
          	return false;
          }
      }
  
      private static class IntColumnSerializer extends ColumnSerializer {
          @Override
  		protected void writeObject(ObjectOutput out, Object obj, Map<Object, Integer> cache, byte version) throws IOException {
              out.writeInt(((Integer)obj).intValue());
          }
          @Override
  		protected Object readObject(ObjectInput in, List<Object> cache, byte version) throws IOException {
              return Integer.valueOf(in.readInt());
          }
      }
  
      private static class LongColumnSerializer extends ColumnSerializer {
          @Override
  		protected void writeObject(ObjectOutput out, Object obj, Map<Object, Integer> cache, byte version) throws IOException {
              out.writeLong(((Long)obj).longValue());
          }
          @Override
  		protected Object readObject(ObjectInput in, List<Object> cache, byte version) throws IOException {
              return Long.valueOf(in.readLong());
          }
      }
  
      private static class FloatColumnSerializer extends ColumnSerializer {
          @Override
  		protected void writeObject(ObjectOutput out, Object obj, Map<Object, Integer> cache, byte version) throws IOException {
              out.writeFloat(((Float)obj).floatValue());
          }
          @Override
  		protected Object readObject(ObjectInput in, List<Object> cache, byte version) throws IOException {
              return new Float(in.readFloat());
          }
      }
  
      private static class DoubleColumnSerializer extends ColumnSerializer {
          @Override
  		protected void writeObject(ObjectOutput out, Object obj, Map<Object, Integer> cache, byte version) throws IOException {
              out.writeDouble(((Double)obj).doubleValue());
          }
          @Override
  		protected Object readObject(ObjectInput in, List<Object> cache, byte version) throws IOException {
              return new Double(in.readDouble());
          }
      }
  
      private static class ShortColumnSerializer extends ColumnSerializer {
          @Override
  		protected void writeObject(ObjectOutput out, Object obj, Map<Object, Integer> cache, byte version) throws IOException {
              out.writeShort(((Short)obj).shortValue());
          }
          @Override
  		protected Object readObject(ObjectInput in, List<Object> cache, byte version) throws IOException {
              return Short.valueOf(in.readShort());
          }
      }
  
      private static class BooleanColumnSerializer extends ColumnSerializer {
          /* This implementation compacts the isNull and boolean data for non-null values into a byte[]
           * by using a 8 bit mask that is bit-shifted to mask each value.
           */
      	@Override
          public void writeColumn(ObjectOutput out, int col, List<? extends List<?>> batch, Map<Object, Integer> cache, byte version) throws IOException {
              int currentByte = 0;
              int mask = 0x80;
              Object obj;
              for (int row = 0; row < batch.size(); row++) {
                  // Write the isNull value
                  obj = batch.get(row).get(col);
                  if (obj == null ) {
                      currentByte |= mask;
                  }
                  mask >>= 1; // Shift the mask to the next bit
                  if (mask == 0) {
                      // If the current byte has been used up, write it and reset.
                      out.write(currentByte);
                      currentByte = 0;
                      mask = 0x80;
                  }
                  if (obj != null) {
                      // Write the boolean value if it's not null
                      if (((Boolean)obj).booleanValue()) {
                          currentByte |= mask;
                      }
                      mask >>= 1;
                      if (mask == 0) {
                          out.write(currentByte);
                          currentByte = 0;
                          mask = 0x80;
                      }
                  }
              }
              // Invariant mask != 0
              // If we haven't reached the eight-row mark then the loop would not have written this byte
              // Write the final byte containing data for the extra rows, if it exists.
              if (mask != 0x80) {
                  out.write(currentByte);
              }
          }
      	
      	@Override
      	public void readColumn(ObjectInput in, int col,
      			List<List<Object>> batch, byte[] isNull, List<Object> cache, byte version) throws IOException,
          		ClassNotFoundException {
              int currentByte = 0, mask = 0; // Initialize the mask so that it is reset in the loop
              boolean isNullVal;
              for (int row = 0; row < batch.size(); row++) {
                  if (mask == 0) {
                      // If we used up the byte, read the next one, and reset the mask
  