package com.facebook.presto.metadata;

import com.facebook.presto.block.Block;
import com.facebook.presto.block.BlockIterable;
import com.facebook.presto.operator.Page;
import com.facebook.presto.operator.PageBuilder;
import com.facebook.presto.tuple.Tuple;
import com.facebook.presto.tuple.TupleInfo;
import com.google.common.collect.ImmutableList;

import java.util.ArrayList;
import java.util.List;

import static com.facebook.presto.block.BlockIterables.createBlockIterable;
import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkNotNull;

public class InternalTable
{
    private final List<BlockIterable> columns;

    public InternalTable(List<BlockIterable> columns)
    {
        this.columns = ImmutableList.copyOf(checkNotNull(columns, "columns is null"));
    }

    public BlockIterable getColumn(int index)
    {
        return columns.get(index);
    }

    public static Builder builder(TupleInfo tupleInfo)
    {
        return new Builder(tupleInfo);
    }

    public static class Builder
    {
        private final TupleInfo tupleInfo;
        private final List<TupleInfo> tupleInfos;
        private final List<List<Block>> columns;
        private PageBuilder pageBuilder;

        public Builder(TupleInfo tupleInfo)
        {
            this.tupleInfo = checkNotNull(tupleInfo, "tupleInfo is null");
            tupleInfos = getTupleInfos(tupleInfo);

            columns = new ArrayList<>();
            for (int i = 0; i < tupleInfo.getFieldCount(); i++) {
                columns.add(new ArrayList<Block>());
            }

            pageBuilder = new PageBuilder(tupleInfos);
        }

        public void add(Tuple tuple)
        {
            checkArgument(tuple.getTupleInfo().equals(tupleInfo), "tuple schema does not match builder");

            for (int i = 0; i < tupleInfo.getFieldCount(); i++) {
                pageBuilder.getBlockBuilder(i).append(tuple, i);
            }

            if (pageBuilder.isFull()) {
                flushPage();
                pageBuilder = new PageBuilder(tupleInfos);
            }
        }

        public InternalTable build()
        {
            flushPage();
            ImmutableList.Builder<BlockIterable> list = ImmutableList.builder();
            for (List<Block> column : columns) {
                list.add(createBlockIterable(column));
            }
            return new InternalTable(list.build());
        }

        private void flushPage()
        {
            Page page = pageBuilder.build();
            for (int i = 0; i < tupleInfo.getFieldCount(); i++) {
                columns.get(i).add(page.getBlock(i));
            }
        }

        private static List<TupleInfo> getTupleInfos(TupleInfo tupleInfo)
        {
            ImmutableList.Builder<TupleInfo> list = ImmutableList.builder();
            for (TupleInfo.Type type : tupleInfo.getTypes()) {
                list.add(new TupleInfo(type));
            }
            return list.build();
        }
    }
}

@@ -468,24 +468,95 @@
 ckIterable;%0A
+import static com.facebook.presto.block.BlockUtils.emptyBlockIterable;%0A
 import stati
@@ -2431,16 +2431,58 @@
 ist.add(
+column.isEmpty() ? emptyBlockIterable() : 
 createBl
@@ -2499,24 +2499,24 @@
 e(column));%0A
-
             
@@ -2615,32 +2615,78 @@
 age()%0A        %7B%0A
+            if (!pageBuilder.isEmpty()) %7B%0A    
             Page
@@ -2706,32 +2706,36 @@
 uilder.build();%0A
+    
             for 
@@ -2776,32 +2776,36 @@
 Count(); i++) %7B%0A
+    
                 
@@ -2838,24 +2838,42 @@
 tBlock(i));%0A
+                %7D%0A
             

Support building empty internal tables