Rewrite ColumnProcessor and data import

src/main/java/com/facebook/presto/AbstractColumnProcessor.java

@@ -0,0 +1,43 @@
+package com.facebook.presto;
+
+import java.io.IOException;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkElementIndex;
+import static com.google.common.base.Preconditions.checkNotNull;
+import static com.google.common.base.Preconditions.checkState;
+
+public abstract class AbstractColumnProcessor
+        implements ColumnProcessor
+{
+    protected final TupleInfo.Type type;
+    protected final int index;
+    protected final Cursor cursor;
+
+    private boolean finished = false;
+
+    protected AbstractColumnProcessor(TupleInfo.Type type, int index, Cursor cursor)
+    {
+        checkNotNull(type, "type is null");
+        checkNotNull(cursor, "cursor is null");
+        checkElementIndex(index, cursor.getTupleInfo().getFieldCount());
+        TupleInfo.Type cursorType = cursor.getTupleInfo().getTypes().get(index);
+        checkArgument(type == cursorType, "type (%s) does not match cursor type (%s) at index (%s)", type, cursorType, index);
+
+        this.type = type;
+        this.index = index;
+        this.cursor = cursor;
+    }
+
+    @Override
+    public final void finish()
+            throws IOException
+    {
+        checkState(!finished, "finish called twice");
+        finished = true;
+        finished();
+    }
+
+    protected abstract void finished()
+            throws IOException;
+}

src/main/java/com/facebook/presto/BlockBuilder.java

@@ -3,12 +3,12 @@
 import com.facebook.presto.slice.DynamicSliceOutput;
 import com.facebook.presto.slice.Slice;
 import com.facebook.presto.slice.Slices;
-import com.google.common.base.Preconditions;
 import io.airlift.units.DataSize;
 import io.airlift.units.DataSize.Unit;
 
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkNotNull;
+import static com.google.common.base.Preconditions.checkState;
 
 public class BlockBuilder
 {
@@ -40,6 +40,11 @@ public BlockBuilder(long startPosition, TupleInfo tupleInfo, DataSize blockSize)
         tupleBuilder = tupleInfo.builder(sliceOutput);
     }
 
+    public boolean isEmpty()
+    {
+        return count == 0;
+    }
+
     public boolean isFull()
     {
         return sliceOutput.size() > maxBlockSize;
@@ -93,7 +98,8 @@ public UncompressedValueBlock build()
     {
         flushTupleIfNecessary();
 
-        Preconditions.checkState(count > 0, "Cannot build an empty block");
+        checkState(!tupleBuilder.isPartial(), "Tuple is not complete");
+        checkState(!isEmpty(), "Cannot build an empty block");
 
         return new UncompressedValueBlock(Range.create(startPosition, startPosition + count - 1), tupleInfo, sliceOutput.slice());
     }

src/main/java/com/facebook/presto/ColumnProcessor.java

@@ -3,13 +3,13 @@
  */
 package com.facebook.presto;
 
-import com.facebook.presto.TupleInfo.Type;
+import java.io.IOException;
 
 public interface ColumnProcessor
 {
-    Type getColumnType();
+    boolean processPositions(long end)
+            throws IOException;
 
-    void processBlock(ValueBlock block);
-
-    void finish();
+    void finish()
+            throws IOException;
 }

src/main/java/com/facebook/presto/ColumnProcessors.java

@@ -0,0 +1,36 @@
+package com.facebook.presto;
+
+import java.io.IOException;
+import java.util.List;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+import static com.google.common.base.Preconditions.checkState;
+
+public class ColumnProcessors
+{
+    private static final int DEFAULT_BATCH_SIZE = 10_000;
+
+    public static void process(List<? extends ColumnProcessor> processors)
+            throws IOException
+    {
+        process(processors, DEFAULT_BATCH_SIZE);
+    }
+
+    public static void process(List<? extends ColumnProcessor> processors, int batchSize)
+            throws IOException
+    {
+        checkNotNull(processors, "processors is null");
+        long end = 0;
+        while (true) {
+            end = Math.min(end + batchSize, Integer.MAX_VALUE);
+            boolean moreData = false;
+            for (ColumnProcessor processor : processors) {
+                moreData |= processor.processPositions(end);
+            }
+            if (!moreData) {
+                return;
+            }
+            checkState(end < Integer.MAX_VALUE, "processor should be complete");
+        }
+    }
+}