package com.facebook.presto;

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;

public class BlockBuilder
{
    private static final DataSize DEFAULT_MAX_BLOCK_SIZE = new DataSize(64, Unit.KILOBYTE);

    private final long startPosition;
    private final TupleInfo tupleInfo;
    private final int maxBlockSize;
    private final DynamicSliceOutput sliceOutput;
    private int count;

    private TupleInfo.Builder tupleBuilder;

    public BlockBuilder(long startPosition, TupleInfo tupleInfo)
    {
        this(startPosition, tupleInfo, DEFAULT_MAX_BLOCK_SIZE);
    }

    public BlockBuilder(long startPosition, TupleInfo tupleInfo, DataSize blockSize)
    {
        checkArgument(startPosition >= 0, "startPosition is negative");
        checkNotNull(blockSize, "blockSize is null");

        this.startPosition = startPosition;
        this.tupleInfo = tupleInfo;
        maxBlockSize = (int) blockSize.toBytes();
        sliceOutput = new DynamicSliceOutput((int) blockSize.toBytes());

        tupleBuilder = tupleInfo.builder(sliceOutput);
    }

    public boolean isFull()
    {
        return sliceOutput.size() > maxBlockSize;
    }

    public void append(long value)
    {
        flushTupleIfNecessary();

        tupleBuilder.append(value);
    }

    public void append(byte[] value)
    {
        flushTupleIfNecessary();

        tupleBuilder.append(Slices.wrappedBuffer(value));
    }

    public void append(Slice value)
    {
        flushTupleIfNecessary();

        tupleBuilder.append(value);
    }

    public void append(Tuple tuple)
    {
        flushTupleIfNecessary();

        tupleBuilder.append(tuple);
    }

    private void flushTupleIfNecessary()
    {
        if (tupleBuilder.isComplete()) {
            tupleBuilder.finish();
            count++;
        }
    }

    public ValueBlock build()
    {
        flushTupleIfNecessary();

        Preconditions.checkState(count > 0, "Cannot build an empty block");

        return new UncompressedValueBlock(Range.create(startPosition, startPosition + count - 1), tupleInfo, sliceOutput.slice());
    }
}

@@ -1347,36 +1347,44 @@
   %7D%0A%0A    public 
-void
+BlockBuilder
  append(long val
@@ -1454,32 +1454,54 @@
 r.append(value);
+%0A%0A        return this;
 %0A    %7D%0A%0A    publ
@@ -1495,36 +1495,44 @@
   %7D%0A%0A    public 
-void
+BlockBuilder
  append(byte%5B%5D v
@@ -1634,16 +1634,38 @@
 value));
+%0A%0A        return this;
 %0A    %7D%0A%0A
@@ -1667,36 +1667,44 @@
   %7D%0A%0A    public 
-void
+BlockBuilder
  append(Slice va
@@ -1783,16 +1783,38 @@
 (value);
+%0A%0A        return this;
 %0A    %7D%0A%0A
@@ -1824,20 +1824,28 @@
  public 
-void
+BlockBuilder
  append(
@@ -1932,16 +1932,38 @@
 (tuple);
+%0A%0A        return this;
 %0A    %7D%0A%0A

Make BlockBuilder append chainable