Add getType(field) to RecordCursor

sjtufighter · Oct 29, 2013 · f58b438 · f58b438
1 parent d47d563
commit f58b438
Show file tree

Hide file tree

Showing 12 changed files with 87 additions and 74 deletions.
diff --git a/presto-hive/src/main/java/com/facebook/presto/hive/BytesHiveRecordCursor.java b/presto-hive/src/main/java/com/facebook/presto/hive/BytesHiveRecordCursor.java
@@ -201,6 +201,12 @@ public long getCompletedBytes()
         return completedBytes;
     }
 
+    @Override
+    public ColumnType getType(int field)
+    {
+        return types[field];
+    }
+
     @Override
     public boolean advanceNextPosition()
     {

diff --git a/presto-hive/src/main/java/com/facebook/presto/hive/GenericHiveRecordCursor.java b/presto-hive/src/main/java/com/facebook/presto/hive/GenericHiveRecordCursor.java
@@ -199,6 +199,12 @@ public long getCompletedBytes()
         return completedBytes;
     }
 
+    @Override
+    public ColumnType getType(int field)
+    {
+        return types[field];
+    }
+
     @Override
     public boolean advanceNextPosition()
     {

diff --git a/presto-main/src/main/java/com/facebook/presto/operator/FilterAndProjectOperator.java b/presto-main/src/main/java/com/facebook/presto/operator/FilterAndProjectOperator.java
@@ -35,12 +35,7 @@ public static class FilterAndProjectOperatorFactory
         private final List<TupleInfo> tupleInfos;
         private boolean closed;
 
-        public FilterAndProjectOperatorFactory(int operatorId, FilterFunction filterFunction, ProjectionFunction... projections)
-        {
-            this(operatorId, filterFunction, ImmutableList.copyOf(checkNotNull(projections, "projections is null")));
-        }
-
-        public FilterAndProjectOperatorFactory(int operatorId, FilterFunction filterFunction, List<ProjectionFunction> projections)
+        public FilterAndProjectOperatorFactory(int operatorId, FilterFunction filterFunction, Iterable<? extends ProjectionFunction> projections)
         {
             this.operatorId = operatorId;
             this.filterFunction = checkNotNull(filterFunction, "filterFunction is null");
@@ -72,12 +67,7 @@ public void close()
     private final FilterFunction filterFunction;
     private final List<ProjectionFunction> projections;
 
-    public FilterAndProjectOperator(OperatorContext operatorContext, FilterFunction filterFunction, ProjectionFunction... projections)
-    {
-        this(operatorContext, filterFunction, ImmutableList.copyOf(checkNotNull(projections, "projections is null")));
-    }
-
-    public FilterAndProjectOperator(OperatorContext operatorContext, FilterFunction filterFunction, List<ProjectionFunction> projections)
+    public FilterAndProjectOperator(OperatorContext operatorContext, FilterFunction filterFunction, Iterable<? extends ProjectionFunction> projections)
     {
         super(operatorContext, toTupleInfos(checkNotNull(projections, "projections is null")));
         this.filterFunction = checkNotNull(filterFunction, "filterFunction is null");
@@ -112,7 +102,7 @@ protected void filterAndProjectRowOriented(Block[] blocks, PageBuilder pageBuild
         }
     }
 
-    private static List<TupleInfo> toTupleInfos(List<ProjectionFunction> projections)
+    private static List<TupleInfo> toTupleInfos(Iterable<? extends ProjectionFunction> projections)
     {
         ImmutableList.Builder<TupleInfo> tupleInfos = ImmutableList.builder();
         for (ProjectionFunction projection : projections) {

diff --git a/presto-main/src/main/java/com/facebook/presto/operator/ScanFilterAndProjectOperator.java b/presto-main/src/main/java/com/facebook/presto/operator/ScanFilterAndProjectOperator.java
@@ -48,18 +48,7 @@ public ScanFilterAndProjectOperatorFactory(
                 DataStreamProvider dataStreamProvider,
                 Iterable<ColumnHandle> columns,
                 FilterFunction filterFunction,
-                ProjectionFunction... projections)
-        {
-            this(operatorId, sourceId, dataStreamProvider, columns, filterFunction, ImmutableList.copyOf(checkNotNull(projections, "projections is null")));
-        }
-
-        public ScanFilterAndProjectOperatorFactory(
-                int operatorId,
-                PlanNodeId sourceId,
-                DataStreamProvider dataStreamProvider,
-                Iterable<ColumnHandle> columns,
-                FilterFunction filterFunction,
-                Iterable<ProjectionFunction> projections)
+                Iterable<? extends ProjectionFunction> projections)
         {
             this.operatorId = operatorId;
             this.sourceId = checkNotNull(sourceId, "sourceId is null");
@@ -106,23 +95,7 @@ public ScanFilterAndProjectOperator(
             DataStreamProvider dataStreamProvider,
             Iterable<ColumnHandle> columns,
             FilterFunction filterFunction,
-            ProjectionFunction... projections)
-    {
-        this(operatorContext,
-                sourceId,
-                dataStreamProvider,
-                columns,
-                filterFunction,
-                ImmutableList.copyOf(checkNotNull(projections, "projections is null")));
-    }
-
-    public ScanFilterAndProjectOperator(
-            OperatorContext operatorContext,
-            PlanNodeId sourceId,
-            DataStreamProvider dataStreamProvider,
-            Iterable<ColumnHandle> columns,
-            FilterFunction filterFunction,
-            Iterable<ProjectionFunction> projections)
+            Iterable<? extends ProjectionFunction> projections)
     {
         super(operatorContext,
                 sourceId,

diff --git a/presto-main/src/main/java/com/facebook/presto/split/MappedRecordSet.java b/presto-main/src/main/java/com/facebook/presto/split/MappedRecordSet.java
@@ -80,6 +80,12 @@ public long getCompletedBytes()
             return delegate.getCompletedBytes();
         }
 
+        @Override
+        public ColumnType getType(int field)
+        {
+            return delegate.getType(field);
+        }
+
         @Override
         public boolean advanceNextPosition()
         {

diff --git a/presto-main/src/main/java/com/facebook/presto/sql/planner/TupleInputResolver.java b/presto-main/src/main/java/com/facebook/presto/sql/planner/TupleInputResolver.java
@@ -31,7 +31,6 @@ public class TupleInputResolver
     private TupleReadable[] inputs;
 
     private RecordCursor cursor;
-    private Map<Input, Type> inputTypes;
 
     public void setInputs(TupleReadable[] inputs)
     {
@@ -43,7 +42,6 @@ public void setCursor(RecordCursor cursor, Map<Input, Type> tupleInfo)
     {
         checkState(inputs == null, "%s already has inputs set", getClass().getName());
         this.cursor = checkNotNull(cursor, "cursor is null");
-        this.inputTypes = checkNotNull(tupleInfo, "tupleInfo is null");
     }
 
     @Override
@@ -78,14 +76,14 @@ else if (cursor != null) {
                 return null;
             }
 
-            switch (inputTypes.get(input)) {
+            switch (cursor.getType(input.getChannel())) {
                 case BOOLEAN:
                     return cursor.getBoolean(channel);
-                case BIGINT:
+                case LONG:
                     return cursor.getLong(channel);
                 case DOUBLE:
                     return cursor.getDouble(channel);
-                case VARCHAR:
+                case STRING:
                     return Slices.wrappedBuffer(cursor.getString(channel));
                 default:
                     throw new UnsupportedOperationException("not yet implemented");

diff --git a/presto-main/src/main/java/com/facebook/presto/util/DelimitedRecordSet.java b/presto-main/src/main/java/com/facebook/presto/util/DelimitedRecordSet.java
@@ -104,6 +104,12 @@ public long getCompletedBytes()
             return 0;
         }
 
+        @Override
+        public ColumnType getType(int field)
+        {
+            return columns.get(field).getType();
+        }
+
         @Override
         public boolean advanceNextPosition()
         {

diff --git a/presto-main/src/test/java/com/facebook/presto/benchmark/PredicateFilterBenchmark.java b/presto-main/src/test/java/com/facebook/presto/benchmark/PredicateFilterBenchmark.java
@@ -45,7 +45,10 @@ protected List<? extends OperatorFactory> createOperatorFactories()
     {
         BlockIterable blockIterable = getBlockIterable("orders", "totalprice", BlocksFileEncoding.RAW);
         AlignmentOperatorFactory alignmentOperator = new AlignmentOperatorFactory(0, blockIterable);
-        FilterAndProjectOperatorFactory filterAndProjectOperator = new FilterAndProjectOperatorFactory(1, new DoubleFilter(50000.00), singleColumn(Type.DOUBLE, 0, 0));
+        FilterAndProjectOperatorFactory filterAndProjectOperator = new FilterAndProjectOperatorFactory(
+                1,
+                new DoubleFilter(50000.00),
+                ImmutableList.of(singleColumn(Type.DOUBLE, 0, 0)));
 
         return ImmutableList.of(alignmentOperator, filterAndProjectOperator);
     }

diff --git a/presto-main/src/test/java/com/facebook/presto/operator/TestFilterAndProjectOperator.java b/presto-main/src/test/java/com/facebook/presto/operator/TestFilterAndProjectOperator.java
@@ -20,6 +20,7 @@
 import com.facebook.presto.tuple.TupleInfo;
 import com.facebook.presto.tuple.TupleReadable;
 import com.facebook.presto.util.MaterializedResult;
+import com.google.common.collect.ImmutableList;
 import org.testng.annotations.AfterMethod;
 import org.testng.annotations.BeforeMethod;
 import org.testng.annotations.Test;
@@ -28,6 +29,7 @@
 import java.util.concurrent.ExecutorService;
 
 import static com.facebook.presto.operator.OperatorAssertion.assertOperatorEquals;
+import static com.facebook.presto.operator.ProjectionFunctions.concat;
 import static com.facebook.presto.operator.ProjectionFunctions.singleColumn;
 import static com.facebook.presto.operator.RowPagesBuilder.rowPagesBuilder;
 import static com.facebook.presto.tuple.TupleInfo.SINGLE_LONG;
@@ -67,22 +69,25 @@ public void testAlignment()
                 .addSequencePage(100, 0, 0)
                 .build();
 
-        OperatorFactory operatorFactory = new FilterAndProjectOperatorFactory(0, new FilterFunction()
-        {
-            @Override
-            public boolean filter(TupleReadable... cursors)
-            {
-                long value = cursors[1].getLong(0);
-                return 10 <= value && value < 20;
-            }
+        OperatorFactory operatorFactory = new FilterAndProjectOperatorFactory(
+                0,
+                new FilterFunction()
+                {
+                    @Override
+                    public boolean filter(TupleReadable... cursors)
+                    {
+                        long value = cursors[1].getLong(0);
+                        return 10 <= value && value < 20;
+                    }
 
-            @Override
-            public boolean filter(RecordCursor cursor)
-            {
-                long value = cursor.getLong(0);
-                return 10 <= value && value < 20;
-            }
-        }, ProjectionFunctions.concat(singleColumn(VARIABLE_BINARY, 0, 0), singleColumn(FIXED_INT_64, 1, 0)));
+                    @Override
+                    public boolean filter(RecordCursor cursor)
+                    {
+                        long value = cursor.getLong(0);
+                        return 10 <= value && value < 20;
+                    }
+                },
+                ImmutableList.of(concat(singleColumn(VARIABLE_BINARY, 0, 0), singleColumn(FIXED_INT_64, 1, 0))));
 
         Operator operator = operatorFactory.createOperator(driverContext);
 

diff --git a/presto-main/src/test/java/com/facebook/presto/util/InfiniteRecordSet.java b/presto-main/src/test/java/com/facebook/presto/util/InfiniteRecordSet.java
@@ -17,10 +17,12 @@
 import com.facebook.presto.spi.RecordCursor;
 import com.facebook.presto.spi.RecordSet;
 import com.google.common.base.Charsets;
-import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
 
 import java.util.List;
 
+import static com.google.common.base.Preconditions.checkNotNull;
+
 public class InfiniteRecordSet
         implements RecordSet
 {
@@ -42,18 +44,19 @@ public List<ColumnType> getColumnTypes()
     @Override
     public RecordCursor cursor()
     {
-        return new InMemoryRecordCursor(record);
+        return new InMemoryRecordCursor(types, record);
     }
 
     private static class InMemoryRecordCursor
             implements RecordCursor
     {
+        private final List<ColumnType> types;
         private final List<?> record;
 
-        private InMemoryRecordCursor(List<?> record)
+        private InMemoryRecordCursor(List<ColumnType> types, List<?> record)
         {
-            Preconditions.checkNotNull(record, "record is null");
-            this.record = record;
+            this.types = checkNotNull(ImmutableList.copyOf(types), "types is null");
+            this.record = checkNotNull(ImmutableList.copyOf(record), "record is null");
         }
 
         @Override
@@ -74,32 +77,38 @@ public boolean advanceNextPosition()
             return true;
         }
 
+        @Override
+        public ColumnType getType(int field)
+        {
+            return types.get(field);
+        }
+
         @Override
         public boolean getBoolean(int field)
         {
-            Preconditions.checkNotNull(record.get(field), "value is null");
+            checkNotNull(record.get(field), "value is null");
             return (Boolean) record.get(field);
         }
 
         @Override
         public long getLong(int field)
         {
-            Preconditions.checkNotNull(record.get(field), "value is null");
+            checkNotNull(record.get(field), "value is null");
             return (Long) record.get(field);
         }
 
         @Override
         public double getDouble(int field)
         {
-            Preconditions.checkNotNull(record.get(field), "value is null");
+            checkNotNull(record.get(field), "value is null");
             return (Double) record.get(field);
         }
 
         @Override
         public byte[] getString(int field)
         {
             Object value = record.get(field);
-            Preconditions.checkNotNull(value, "value is null");
+            checkNotNull(value, "value is null");
             if (value instanceof byte[]) {
                 return (byte[]) value;
             }

diff --git a/presto-spi/src/main/java/com/facebook/presto/spi/InMemoryRecordSet.java b/presto-spi/src/main/java/com/facebook/presto/spi/InMemoryRecordSet.java
@@ -49,19 +49,22 @@ public List<ColumnType> getColumnTypes()
     @Override
     public RecordCursor cursor()
     {
-        return new InMemoryRecordCursor(records.iterator(), totalBytes);
+        return new InMemoryRecordCursor(types, records.iterator(), totalBytes);
     }
 
     private static class InMemoryRecordCursor
             implements RecordCursor
     {
+        private final List<ColumnType> types;
         private final Iterator<? extends List<?>> records;
         private final long totalBytes;
         private List<?> record;
         private long completedBytes;
 
-        private InMemoryRecordCursor(Iterator<? extends List<?>> records, long totalBytes)
+        private InMemoryRecordCursor(List<ColumnType> types, Iterator<? extends List<?>> records, long totalBytes)
         {
+            this.types = types;
+
             this.records = records;
 
             this.totalBytes = totalBytes;
@@ -79,6 +82,12 @@ public long getCompletedBytes()
             return completedBytes;
         }
 
+        @Override
+        public ColumnType getType(int field)
+        {
+            return types.get(field);
+        }
+
         @Override
         public boolean advanceNextPosition()
         {

diff --git a/presto-spi/src/main/java/com/facebook/presto/spi/RecordCursor.java b/presto-spi/src/main/java/com/facebook/presto/spi/RecordCursor.java
@@ -22,6 +22,8 @@ public interface RecordCursor
 
     long getCompletedBytes();
 
+    ColumnType getType(int field);
+
     boolean advanceNextPosition();
 
     boolean getBoolean(int field);