/*
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.facebook.presto.operator;

import com.facebook.presto.block.BlockBuilder;
import com.facebook.presto.spi.ColumnType;
import com.facebook.presto.spi.RecordCursor;
import com.facebook.presto.spi.RecordSet;
import com.facebook.presto.tuple.TupleInfo;
import com.facebook.presto.tuple.TupleInfo.Type;
import com.google.common.collect.ImmutableList;
import com.google.common.util.concurrent.ListenableFuture;
import io.airlift.units.DataSize;

import java.util.List;

import static com.google.common.base.Preconditions.checkNotNull;
import static io.airlift.units.DataSize.Unit.BYTE;

public class RecordProjectOperator
        implements Operator
{
    private static final int ROWS_PER_REQUEST = 16384;
    private final OperatorContext operatorContext;
    private final RecordSet recordSet;
    private final List<TupleInfo> tupleInfos;
    private final PageBuilder pageBuilder;
    private RecordCursor cursor;
    private boolean finishing;
    private long completedBytes;

    public RecordProjectOperator(OperatorContext operatorContext, RecordSet recordSet)
    {
        this.operatorContext = checkNotNull(operatorContext, "operatorContext is null");
        this.recordSet = checkNotNull(recordSet, "recordSet is null");

        // project each field into a separate channel
        ImmutableList.Builder<TupleInfo> tupleInfos = ImmutableList.builder();
        for (ColumnType columnType : recordSet.getColumnTypes()) {
            tupleInfos.add(new TupleInfo(Type.fromColumnType(columnType)));
        }
        this.tupleInfos = tupleInfos.build();

        pageBuilder = new PageBuilder(getTupleInfos());
    }

    @Override
    public OperatorContext getOperatorContext()
    {
        return operatorContext;
    }

    public RecordCursor getCursor()
    {
        if (cursor == null) {
            cursor = recordSet.cursor();
        }
        return cursor;
    }

    @Override
    public List<TupleInfo> getTupleInfos()
    {
        return tupleInfos;
    }

    @Override
    public void finish()
    {
        finishing = true;
        if (cursor != null) {
            cursor.close();
        }
    }

    @Override
    public boolean isFinished()
    {
        return finishing && pageBuilder.isEmpty();
    }

    @Override
    public ListenableFuture<?> isBlocked()
    {
        return NOT_BLOCKED;
    }

    @Override
    public boolean needsInput()
    {
        return false;
    }

    @Override
    public void addInput(Page page)
    {
        throw new UnsupportedOperationException(getClass().getName() + " can not take input");
    }

    @Override
    public Page getOutput()
    {
        if (!finishing) {
            RecordCursor cursor = getCursor();

            int i = 0;
            for (; i < ROWS_PER_REQUEST; i++) {
                if (pageBuilder.isFull()) {
                    break;
                }

                if (!cursor.advanceNextPosition()) {
                    finishing = true;
                    break;
                }

                for (int column = 0; column < tupleInfos.size(); column++) {
                    BlockBuilder output = pageBuilder.getBlockBuilder(column);
                    if (cursor.isNull(column)) {
                        output.appendNull();
                    }
                    else {
                        Type type = getTupleInfos().get(column).getTypes().get(0);
                        switch (type) {
                            case BOOLEAN:
                                output.append(cursor.getBoolean(column));
                                break;
                            case FIXED_INT_64:
                                output.append(cursor.getLong(column));
                                break;
                            case DOUBLE:
                                output.append(cursor.getDouble(column));
                                break;
                            case VARIABLE_BINARY:
                                output.append(cursor.getString(column));
                                break;
                            default:
                                throw new AssertionError("unimplemented type: " + type);
                        }
                    }
                }
            }

            long bytesProcessed = cursor.getCompletedBytes() - completedBytes;
            operatorContext.recordGeneratedInput(new DataSize(bytesProcessed, BYTE), i);
            completedBytes += bytesProcessed;
        }

        // only return a full page is buffer is full or we are finishing
        if (pageBuilder.isEmpty() || (!finishing && !pageBuilder.isFull())) {
            return null;
        }

        Page page = pageBuilder.build();
        pageBuilder.reset();
        return page;
    }
}

@@ -1342,29 +1342,29 @@
 l Record
-Set recordSet
+Cursor cursor
 ;%0A    pr
@@ -1450,41 +1450,8 @@
 er;%0A
-    private RecordCursor cursor;%0A
     
@@ -1596,32 +1596,241 @@
 ecordSet)%0A    %7B%0A
+        this(operatorContext, recordSet.getColumnTypes(), recordSet.cursor());%0A    %7D%0A%0A    public RecordProjectOperator(OperatorContext operatorContext, List%3CColumnType%3E columnTypes, RecordCursor cursor)%0A    %7B%0A
         this.ope
@@ -1915,25 +1915,22 @@
    this.
-recordSet
+cursor
  = check
@@ -1941,29 +1941,23 @@
 ull(
-recordSet, %22recordSet
+cursor, %22cursor
  is 
@@ -2135,30 +2135,17 @@
 nType : 
-recordSet.getC
+c
 olumnTyp
@@ -2146,18 +2146,16 @@
 umnTypes
-()
 ) %7B%0A    
@@ -2499,89 +2499,8 @@
   %7B%0A
-        if (cursor == null) %7B%0A            cursor = recordSet.cursor();%0A        %7D%0A
     
@@ -2697,42 +2697,8 @@
 ue;%0A
-        if (cursor != null) %7B%0A    
     
@@ -2717,26 +2717,16 @@
 lose();%0A
-        %7D%0A
     %7D%0A%0A 
@@ -3211,32 +3211,32 @@
 tOutput()%0A    %7B%0A
+
         if (!fin
@@ -3249,56 +3249,8 @@
 ) %7B%0A
-            RecordCursor cursor = getCursor();%0A%0A
     

Revert "Lazily load RecordCursor in RecordProjectOpereator"