/*
 * Copyright 2004-present Facebook. All Rights Reserved.
 */
package com.facebook.presto.operator;

import com.facebook.presto.block.BlockBuilder;
import com.facebook.presto.tuple.TupleInfo;
import com.facebook.presto.tuple.TupleInfo.Type;
import com.facebook.presto.tuple.TupleReadable;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableList;

import java.util.List;

public class ProjectionFunctions
{
    public static ProjectionFunction singleColumn(Type columnType, int channelIndex, int fieldIndex)
    {
        return new SingleColumnProjection(columnType, channelIndex, fieldIndex);
    }

    private static class SingleColumnProjection implements ProjectionFunction
    {
        private final Type columnType;
        private final int channelIndex;
        private final int fieldIndex;
        private final TupleInfo info;

        public SingleColumnProjection(Type columnType, int channelIndex, int fieldIndex)
        {
            Preconditions.checkNotNull(columnType, "columnType is null");
            Preconditions.checkArgument(channelIndex >= 0, "channelIndex is negative");
            Preconditions.checkArgument(fieldIndex >= 0, "fieldIndex is negative");

            this.columnType = columnType;
            this.channelIndex = channelIndex;
            this.fieldIndex = fieldIndex;
            this.info = new TupleInfo(columnType);
        }

        @Override
        public TupleInfo getTupleInfo()
        {
            return info;
        }

        @Override
        public void project(TupleReadable[] cursors, BlockBuilder output)
        {
            switch (columnType) {
                case FIXED_INT_64:
                    output.append(cursors[channelIndex].getLong(fieldIndex));
                    return;
                case VARIABLE_BINARY:
                    output.append(cursors[channelIndex].getSlice(fieldIndex));
                    return;
                case DOUBLE:
                    output.append(cursors[channelIndex].getDouble(fieldIndex));
                    return;
            }
            throw new IllegalStateException("Unsupported type info " + info);
        }
    }

    public static ProjectionFunction concat(ProjectionFunction... projectionFunctions)
    {
        return concat(ImmutableList.copyOf(projectionFunctions));
    }

    public static ProjectionFunction concat(List<ProjectionFunction> projections)
    {
        return new ConcatProjection(projections);
    }

    private static class ConcatProjection implements ProjectionFunction
    {
        private final List<ProjectionFunction> projections;
        private final TupleInfo tupleInfo;

        private ConcatProjection(List<ProjectionFunction> projections)
        {
            this.projections = projections;

            ImmutableList.Builder<Type> builder = ImmutableList.builder();
            for (ProjectionFunction projection : projections) {
                builder.addAll(projection.getTupleInfo().getTypes());
            }
            this.tupleInfo = new TupleInfo(builder.build());
        }

        @Override
        public TupleInfo getTupleInfo()
        {
            return tupleInfo;
        }

        @Override
        public void project(TupleReadable[] cursors, BlockBuilder output)
        {
            for (ProjectionFunction projection : projections) {
                projection.project(cursors, output);
            }
        }
    }
}

@@ -1608,32 +1608,166 @@
 tput)%0A        %7B%0A
+            if (cursors%5BchannelIndex%5D.isNull(fieldIndex)) %7B%0A                output.appendNull();%0A            %7D%0A            else %7B%0A    
             swit
@@ -1796,24 +1796,28 @@
             
+    
 case FIXED_I
@@ -1823,16 +1823,20 @@
 INT_64:%0A
+    
         
@@ -1921,32 +1921,36 @@
             
+  
+  
 return;%0A
             
@@ -1929,32 +1929,36 @@
         return;%0A
+    
                 
@@ -1979,16 +1979,20 @@
 BINARY:%0A
+    
         
@@ -2078,32 +2078,36 @@
             
+  
+  
 return;%0A
             
@@ -2086,32 +2086,36 @@
         return;%0A
+    
                 
@@ -2127,16 +2127,20 @@
 DOUBLE:%0A
+    
         
@@ -2223,32 +2223,36 @@
                 
+  
+  
 return;%0A        
@@ -2247,34 +2247,42 @@
 rn;%0A            
-%7D%0A
+    %7D%0A    
             thro
@@ -2343,16 +2343,30 @@
  info);%0A
+            %7D%0A
         

Handle nulls