embulk · dmikurube · Aug 9, 2016 · Nov 20, 2017
@@ -4,6  4,7 @@
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import org.embulk.spi.type.Type;
 import org.embulk.spi.type.BinaryType;
 import org.embulk.spi.type.BooleanType;
 import org.embulk.spi.type.DoubleType;
 import org.embulk.spi.type.LongType;
@@ -48,7  49,9 @@ public Type getType()
 
     public void visit(ColumnVisitor visitor)
     {
-        if (type instanceof BooleanType) {
         if (type instanceof BinaryType) {
             visitor.binaryColumn(this);
         } else if (type instanceof BooleanType) {
             visitor.booleanColumn(this);
         } else if (type instanceof LongType) {
             visitor.longColumn(this);

@@ -2,6  2,12 @@
 
 public interface ColumnVisitor
 {
     void binaryColumn(Column column);  // TODO: Add default implementation
     /*
     default void binaryColumn(Column column) {
     }
     */
 
     void booleanColumn(Column column);
 
     void longColumn(Column column);

@@ -1,11  1,13 @@
 package org.embulk.spi;
 
 import java.nio.ByteBuffer;
 import java.util.List;
 import org.msgpack.value.ImmutableValue;
 
 public class Page
 {
     private final Buffer buffer;
     private List<ByteBuffer> binaryReferences;
     private List<String> stringReferences;
     private List<ImmutableValue> valueReferences;
 
@@ -24,6  26,12 @@ public static Page wrap(Buffer buffer)
         return new Page(buffer);
     }
 
     public Page setBinaryReferences(List<ByteBuffer> binaryReferences)
     {
         this.binaryReferences = binaryReferences;
         return this;
     }
 
     public Page setStringReferences(List<String> values)
     {
         this.stringReferences = values;
@@ -36,6  44,12 @@ public Page setValueReferences(List<ImmutableValue> values)
         return this;
     }
 
     public List<ByteBuffer> getBinaryReferences()
     {
         // TODO used by mapreduce executor
         return binaryReferences;
     }
 
     public List<String> getStringReferences()
     {
         // TODO used by mapreduce executor
@@ -48,6  62,11 @@ public List<ImmutableValue> getValueReferences()
         return valueReferences;
     }
 
     public ByteBuffer getBinaryReference(int index)
     {
         return binaryReferences.get(index).asReadOnlyBuffer();
     }
 
     public String getStringReference(int index)
     {
         return stringReferences.get(index);

@@ -1,6  1,7 @@
 package org.embulk.spi;
 
 import java.io.Serializable;
 import java.nio.ByteBuffer;
 import java.util.Map;
 import java.util.List;
 import java.util.Arrays;
@@ -30,6  31,7 @@ public class PageBuilder
     private int count;
     private int position;
     private final byte[] nullBitSet;
     private final BiMap<ByteBuffer, Integer> binaryReferences = HashBiMap.create();
     private final BiMap<String, Integer> stringReferences = HashBiMap.create();
     private List<ImmutableValue> valueReferences = new ArrayList<>();
     private int referenceSize;
@@ -54,6  56,7 @@ private void newBuffer()
         this.bufferSlice = Slices.wrappedBuffer(buffer.array(), buffer.offset(), buffer.capacity());
         this.count = 0;
         this.position = PageFormat.PAGE_HEADER_SIZE;
         this.binaryReferences.clear();
         this.stringReferences.clear();
         this.valueReferences = new ArrayList<>();
         this.referenceSize = 0;
@@ -79,6  82,25 @@ private void clearNull(int columnIndex)
         nullBitSet[columnIndex >>> 3] &= ~(1 << (columnIndex & 7));
     }
 
     public void setBinary(Column column, ByteBuffer value)
     {
         setBinary(column.getIndex(), value);
     }
 
     public void setBinary(int columnIndex, ByteBuffer value)
     {
         Integer reuseIndex = binaryReferences.get(value.asReadOnlyBuffer());
         if (reuseIndex != null) {
             bufferSlice.setInt(getOffset(columnIndex), reuseIndex);
         } else {
             int index = binaryReferences.size();
             binaryReferences.put(value, index);
             bufferSlice.setInt(getOffset(columnIndex), index);
             referenceSize  = value.length * 2   4;  // assuming size of char = size of byte * 2   length
         }
         clearNull(columnIndex);
     }
 
     public void setBoolean(Column column, boolean value)
     {
         // TODO check type?
@@ -169,6  191,33 @@ private int getOffset(int columnIndex)
         return position   columnOffsets[columnIndex];
     }
 
     private static class BinaryReferenceSortComparator
             implements Comparator<Map.Entry<ByteBuffer, Integer>>, Serializable
     {
         @Override
         public int compare(Map.Entry<ByteBuffer, Integer> e1, Map.Entry<ByteBuffer, Integer> e2)
         {
             return e1.getValue().compareTo(e2.getValue());
         }
 
         @Override
         public boolean equals(Object obj)
         {
             return obj instanceof BinaryReferenceSortComparator;
         }
     }
 
     private List<byte[]> getSortedBinaryReferences()
     {
         ArrayList<Map.Entry<ByteBuffer, Integer>> s = new ArrayList<>(binaryReferences.entrySet());
         Collections.sort(s, new BinaryReferenceSortComparator());
         ByteBuffer[] array = new ByteBuffer[s.size()];
         for (int i=0; i < array.length; i  ) {
             array[i] = s.get(i).getKey();
         }
         return Arrays.asList(array);
     }
 
     private static class StringReferenceSortComparator
             implements Comparator<Map.Entry<String, Integer>>, Serializable
     {
@@ -222,6  271,7 @@ private void doFlush()
 
             // flush page
             Page page = Page.wrap(buffer)
                 .setBinaryReferences(getSortedBinaryReferences())
                 .setStringReferences(getSortedStringReferences())
                 .setValueReferences(valueReferences);
             buffer = null;

@@ -66,6  66,17 @@ public boolean isNull(int columnIndex)
         return (nullBitSet[columnIndex >>> 3] & (1 << (columnIndex & 7))) != 0;
     }
 
     public ByteBuffer getBinary(Column column)
     {
         return getBinary(column.getIndex());
     }
 
     public ByteBuffer getBinary(int columnIndex)
     {
         int index = pageSlice.getInt(getOffset(columnIndex));
         return page.getBinaryReference(index).asReadOnlyBuffer();
     }
 
     public boolean getBoolean(Column column)
     {
         // TODO check type?

@@ -0,0  1,14 @@
 package org.embulk.spi.type;
 
 import java.nio.ByteBuffer;
 
 public class BinaryType
         extends AbstractType
 {
     static final BinaryType BINARY = new BinaryType();
 
     private BinaryType()
     {
         super("binary", ByteBuffer.class, 4);
     }
 }
@@ -15,6  15,7 @@ public class TypeDeserializer
 
     static {
         ImmutableMap.Builder<String, Type> builder = ImmutableMap.builder();
         builder.put(BinaryType.BINARY.getName(), BinaryType.BINARY);
         builder.put(BooleanType.BOOLEAN.getName(), BooleanType.BOOLEAN);
         builder.put(LongType.LONG.getName(), LongType.LONG);
         builder.put(DoubleType.DOUBLE.getName(), DoubleType.DOUBLE);

@@ -2,6  2,8 @@
 
 public class Types
 {
     public static final BinaryType BINARY = BinaryType.BINARY;
 
     public static final BooleanType BOOLEAN = BooleanType.BOOLEAN;
 
     public static final LongType LONG = LongType.LONG;

@@ -1,14  1,17 @@
 package org.embulk.spi.util;
 
 import java.nio.ByteBuffer;
 import org.joda.time.DateTimeZone;
 import org.embulk.config.ConfigSource;
 import org.embulk.spi.type.Type;
 import org.embulk.spi.type.BinaryType;
 import org.embulk.spi.type.BooleanType;
 import org.embulk.spi.type.LongType;
 import org.embulk.spi.type.DoubleType;
 import org.embulk.spi.type.StringType;
 import org.embulk.spi.type.TimestampType;
 import org.embulk.spi.type.JsonType;
 import org.embulk.spi.util.dynamic.BinaryColumnSetter;
 import org.embulk.spi.util.dynamic.BooleanColumnSetter;
 import org.embulk.spi.util.dynamic.LongColumnSetter;
 import org.embulk.spi.util.dynamic.DoubleColumnSetter;
@@ -45,7  48,11 @@ public static DefaultValueSetter nullDefaultValue()
     public DynamicColumnSetter newColumnSetter(PageBuilder pageBuilder, Column column)
     {
         Type type = column.getType();
-        if (type instanceof BooleanType) {
         if (type instanceof BinaryType) {
             final TimestampFormatter formatter = new TimestampFormatter(
                 getTimestampFormat(column).getFormat(), getTimeZone(column));
             return new BinaryColumnSetter(pageBuilder, column, defaultValue, formatter);
         } else if (type instanceof BooleanType) {
             return new BooleanColumnSetter(pageBuilder, column, defaultValue);
         } else if (type instanceof LongType) {
             return new LongColumnSetter(pageBuilder, column, defaultValue);

@@ -1,5  1,6 @@
 package org.embulk.spi.util;
 
 import java.nio.ByteBuffer;
 import java.util.List;
 import java.util.ArrayList;
 import org.embulk.spi.time.Timestamp;
@@ -74,6  75,20 @@ public ToStringColumnVisitor(PageReader reader)
             this.reader = reader;
         }
 
         @Override
         public void binaryColumn(Column column)
         {
             final StringBuilder stringBuilder = new StringBuilder();
             final ByteBuffer duplicatedBuffer = reader.getBinary(column).duplicate().rewind();
             while (duplicatedBuffer.hasRemaining()) {
                 if (stringBuilder.length() > 0) {
                     stringBuilder.append(":");
                 }
                 stringBuilder.append(String.format("X", duplicatedBuffer.get()));
             }
             string = stringBuilder.toString();
         }
 
         public void booleanColumn(Column column)
         {
             string = Boolean.toString(reader.getBoolean(column));

@@ -58,6  58,16 @@ public ObjectColumnVisitor(PageReader record)
 
         public abstract void visit(Column column, Object obj);
 
         @Override
         public void binaryColumn(Column column)
         {
             if (record.isNull(column)) {
                 visit(column, null);
             } else {
                 visit(column, record.getBinary(column));
             }
         }
 
         @Override
         public void booleanColumn(Column column)
         {

@@ -14,6  14,7 @@
 import org.embulk.spi.time.Timestamp;
 import org.embulk.spi.json.RubyValueApi;
 import org.msgpack.value.Value;
 import java.nio.ByteBuffer;
 
 public abstract class AbstractDynamicColumnSetter
         implements DynamicColumnSetter
@@ -40,6  41,11 @@ protected AbstractDynamicColumnSetter(PageBuilder pageBuilder, Column column,
 
     public abstract void set(String value);
 
     public void set(ByteBuffer value)
     {
         // Default implementation for older plugins
     }
 
     public abstract void set(Timestamp value);
 
     public abstract void set(Value value);

@@ -0,0  1,73 @@
 package org.embulk.spi.util.dynamic;
 
 import com.google.common.collect.ImmutableSet;
 import java.nio.ByteBuffer;
 import org.embulk.spi.Column;
 import org.embulk.spi.PageBuilder;
 import org.embulk.spi.time.Timestamp;
 import org.embulk.spi.time.TimestampFormatter;
 import org.msgpack.value.Value;
 
 import java.nio.charset.Charset;
 
 public class BinaryColumnSetter
         extends AbstractDynamicColumnSetter
 {
     public BinaryColumnSetter(PageBuilder pageBuilder, Column column,
                               DefaultValueSetter defaultValue,
                               TimestampFormatter timestampFormatter)
     {
         super(pageBuilder, column, defaultValue);
         this.timestampFormatter = timestampFormatter;
     }
 
     @Override
     public void setNull()
     {
         pageBuilder.setNull(column);
     }
 
     @Override
     public void set(boolean v)
     {
         pageBuilder.setBinary(column, Boolean.toString(v).getBytes(Charset.defaultCharset()));
     }
 
     @Override
     public void set(long v)
     {
         pageBuilder.setBinary(column, Long.toString(v).getBytes(Charset.defaultCharset()));
     }
 
     @Override
     public void set(double v)
     {
         pageBuilder.setBinary(column, Double.toString(v).getBytes(Charset.defaultCharset()));
     }
 
     @Override
     public void set(String v)
     {
         pageBuilder.setBinary(column, v.getBytes(Charset.defaultCharset()));
     }
 
     @Override
     public void set(ByteBuffer v)
     {
         pageBuilder.setBinary(column, v.asReadOnlyBuffer());
     }
 
     @Override
     public void set(Timestamp v)
     {
         pageBuilder.setBinary(column, timestampFormatter.format(v).getBytes(Charset.defaultCharset()));
     }
 
     @Override
     public void set(Value v)
     {
         pageBuilder.setBinary(column, v.toJson().getBytes(Charset.defaultCharset()));
     }
 
     private final TimestampFormatter timestampFormatter;
 }