package com.facebook.presto.ingest;

import com.facebook.presto.TupleInfo;
import com.facebook.presto.hive.Record;
import com.facebook.presto.hive.RecordIterator;
import com.facebook.presto.hive.SchemaField;
import com.facebook.presto.slice.Slice;
import com.facebook.presto.slice.Slices;
import com.google.common.base.Charsets;

import java.util.List;

public class HiveTupleStream
        extends AbstractExternalTupleStream
{
    private final RecordIterator recordIterator;
    private final List<SchemaField> schemaFields;
    private Record record;

    public HiveTupleStream(RecordIterator recordIterator, List<SchemaField> schemaFields, TupleInfo tupleInfo)
    {
        super(tupleInfo);
        this.recordIterator = recordIterator;
        this.schemaFields = schemaFields;
    }

    @Override
    protected boolean computeNext()
    {
        if (!recordIterator.hasNext()) {
            return false;
        }
        record = recordIterator.next();
        return true;
    }

    @Override
    protected void buildTuple(TupleInfo.Builder builder)
    {
        int index = 0;
        for (TupleInfo.Type type : tupleInfo.getTypes()) {
            switch (type) {
                case FIXED_INT_64:
                    builder.append(getLong(index));
                    break;
                case DOUBLE:
                    builder.append(getDouble(index));
                    break;
                case VARIABLE_BINARY:
                    builder.append(getSlice(index));
                    break;
                default:
                    throw new AssertionError("unhandled type: " + type);
            }
            index++;
        }
    }

    @Override
    protected long getLong(int field)
    {
        return record.getLong(schemaFields.get(field).getFieldName());
    }

    @Override
    protected double getDouble(int field)
    {
        return record.getDouble(schemaFields.get(field).getFieldName());
    }

    @Override
    protected Slice getSlice(int field)
    {
        String s = record.getString(schemaFields.get(field).getFieldName());
        return Slices.wrappedBuffer(s.getBytes(Charsets.UTF_8));
    }
}

@@ -326,30 +326,203 @@
 ts;%0A
-%0Aimport java.util.List
+import com.google.common.collect.ImmutableList;%0A%0Aimport java.util.List;%0A%0Aimport static com.google.common.base.Preconditions.checkArgument;%0Aimport static com.google.common.base.Strings.nullToEmpty
 ;%0A%0Ap
@@ -814,29 +814,8 @@
 elds
-, TupleInfo tupleInfo
 )%0A  
@@ -832,25 +832,31 @@
   super(
-t
+createT
 upleInfo
 );%0A     
@@ -847,16 +847,30 @@
 upleInfo
+(schemaFields)
 );%0A     
@@ -1897,38 +1897,74 @@
 )%0A    %7B%0A        
-return
+// TODO: null support%0A        Long value =
  record.getLong(
@@ -1996,32 +1996,76 @@
 etFieldName());%0A
+        return (value == null) ? 0 : value;%0A
     %7D%0A%0A    @Over
@@ -2125,22 +2125,60 @@
         
-return
+// TODO: null support%0A        Double value =
  record.
@@ -2224,24 +2224,70 @@
 eldName());%0A
+        return (value == null) ? 0.0 : value;%0A
     %7D%0A%0A    @
@@ -2333,32 +2333,62 @@
 nt field)%0A    %7B%0A
+        // TODO: null support%0A
         String s
@@ -2484,17 +2484,30 @@
 dBuffer(
-s
+nullToEmpty(s)
 .getByte
@@ -2531,11 +2531,958 @@
 );%0A    %7D
+%0A%0A    private static TupleInfo createTupleInfo(List%3CSchemaField%3E schemaFields)%0A    %7B%0A        ImmutableList.Builder%3CTupleInfo.Type%3E list = ImmutableList.builder();%0A        for (SchemaField field : schemaFields) %7B%0A            checkArgument(field.getCategory() == SchemaField.Category.PRIMITIVE, %22Unhandled category: %25s%22, field.getCategory());%0A            SchemaField.Type type = field.getPrimitiveType();%0A            list.add(getTupleType(type));%0A        %7D%0A        return new TupleInfo(list.build());%0A    %7D%0A%0A    private static TupleInfo.Type getTupleType(SchemaField.Type type)%0A    %7B%0A        switch (type) %7B%0A            case LONG:%0A                return TupleInfo.Type.FIXED_INT_64;%0A            case DOUBLE:%0A                return TupleInfo.Type.DOUBLE;%0A            case STRING:%0A                return TupleInfo.Type.VARIABLE_BINARY;%0A            default:%0A                throw new IllegalArgumentException(%22Unhandled type: %22 + type);%0A        %7D%0A    %7D
 %0A%7D%0A

Add null handling and TupleInfo creation