Java 类com.facebook.presto.spi.RecordCursor 实例源码

@Test
public void testGetRecordSet()
        throws Exception
{
    ExampleRecordSetProvider recordSetProvider = new ExampleRecordSetProvider(new ExampleConnectorId("test"));
    RecordSet recordSet = recordSetProvider.getRecordSet(SESSION, new ExampleSplit("test", "schema", "table", dataUri), ImmutableList.of(
            new ExampleColumnHandle("test", "text", VARCHAR, 0),
            new ExampleColumnHandle("test", "value", BIGINT, 1)));
    assertNotNull(recordSet, "recordSet is null");

    RecordCursor cursor = recordSet.cursor();
    assertNotNull(cursor, "cursor is null");

    Map<String, Long> data = new LinkedHashMap<>();
    while (cursor.advanceNextPosition()) {
        data.put(cursor.getSlice(0).toStringUtf8(), cursor.getLong(1));
    }
    assertEquals(data, ImmutableMap.<String, Long>builder()
            .put("ten", 10L)
            .put("eleven", 11L)
            .put("twelve", 12L)
            .build());
}

@Test
public void testCursorSimple()
        throws Exception
{
    RecordSet recordSet = new ExampleRecordSet(new ExampleSplit("test", "schema", "table", dataUri), ImmutableList.of(
            new ExampleColumnHandle("test", "text", VARCHAR, 0),
            new ExampleColumnHandle("test", "value", BIGINT, 1)));
    RecordCursor cursor = recordSet.cursor();

    assertEquals(cursor.getType(0), VARCHAR);
    assertEquals(cursor.getType(1), BIGINT);

    Map<String, Long> data = new LinkedHashMap<>();
    while (cursor.advanceNextPosition()) {
        data.put(cursor.getSlice(0).toStringUtf8(), cursor.getLong(1));
        assertFalse(cursor.isNull(0));
        assertFalse(cursor.isNull(1));
    }
    assertEquals(data, ImmutableMap.<String, Long>builder()
            .put("ten", 10L)
            .put("eleven", 11L)
            .put("twelve", 12L)
            .build());
}

@Test
public void testCursorMixedOrder()
        throws Exception
{
    RecordSet recordSet = new ExampleRecordSet(new ExampleSplit("test", "schema", "table", dataUri), ImmutableList.of(
            new ExampleColumnHandle("test", "value", BIGINT, 1),
            new ExampleColumnHandle("test", "value", BIGINT, 1),
            new ExampleColumnHandle("test", "text", VARCHAR, 0)));
    RecordCursor cursor = recordSet.cursor();

    Map<String, Long> data = new LinkedHashMap<>();
    while (cursor.advanceNextPosition()) {
        assertEquals(cursor.getLong(0), cursor.getLong(1));
        data.put(cursor.getSlice(2).toStringUtf8(), cursor.getLong(0));
    }
    assertEquals(data, ImmutableMap.<String, Long>builder()
            .put("ten", 10L)
            .put("eleven", 11L)
            .put("twelve", 12L)
            .build());
}

private static IndexedTable indexTable(RecordSet recordSet, final List<String> outputColumns, List<String> keyColumns)
{
    List<Integer> keyPositions = FluentIterable.from(keyColumns)
            .transform(columnName -> {
                int position = outputColumns.indexOf(columnName);
                checkState(position != -1);
                return position;
            })
            .toList();

    ImmutableListMultimap.Builder<MaterializedTuple, MaterializedTuple> indexedValuesBuilder = ImmutableListMultimap.builder();

    List<Type> outputTypes = recordSet.getColumnTypes();
    List<Type> keyTypes = extractPositionValues(outputTypes, keyPositions);

    RecordCursor cursor = recordSet.cursor();
    while (cursor.advanceNextPosition()) {
        List<Object> values = extractValues(cursor, outputTypes);
        List<Object> keyValues = extractPositionValues(values, keyPositions);

        indexedValuesBuilder.put(new MaterializedTuple(keyValues), new MaterializedTuple(values));
    }

    return new IndexedTable(keyColumns, keyTypes, outputColumns, outputTypes, indexedValuesBuilder.build());
}

private static Object extractObject(RecordCursor cursor, int field, Type type)
{
    if (cursor.isNull(field)) {
        return null;
    }

    Class<?> javaType = type.getJavaType();
    if (javaType == boolean.class) {
        return cursor.getBoolean(field);
    }
    else if (javaType == long.class) {
        return cursor.getLong(field);
    }
    else if (javaType == double.class) {
        return cursor.getDouble(field);
    }
    else if (javaType == Slice.class) {
        return cursor.getSlice(field).toStringUtf8();
    }
    throw new AssertionError("Unsupported type: " + type);
}

private static Iterable<MaterializedTuple> tupleIterable(final RecordSet recordSet)
{
    return () -> new AbstractIterator<MaterializedTuple>()
    {
        private final RecordCursor cursor = recordSet.cursor();

        @Override
        protected MaterializedTuple computeNext()
        {
            if (!cursor.advanceNextPosition()) {
                return endOfData();
            }
            return new MaterializedTuple(extractValues(cursor, recordSet.getColumnTypes()));
        }
    };
}

@Test
public void testRecordSetProvider()
        throws Exception
{
    for (SchemaTableName schemaTableName : metadata.listTables(SESSION, "jmx")) {
        JmxTableHandle tableHandle = metadata.getTableHandle(SESSION, schemaTableName);
        List<ColumnHandle> columnHandles = ImmutableList.copyOf(metadata.getColumnHandles(SESSION, tableHandle).values());

        ConnectorTableLayoutHandle layout = new JmxTableLayoutHandle(tableHandle, TupleDomain.all());
        ConnectorSplitSource splitSource = splitManager.getSplits(JmxTransactionHandle.INSTANCE, SESSION, layout);
        List<ConnectorSplit> allSplits = getAllSplits(splitSource);
        assertEquals(allSplits.size(), nodes.size());
        ConnectorSplit split = allSplits.get(0);

        RecordSet recordSet = recordSetProvider.getRecordSet(JmxTransactionHandle.INSTANCE, SESSION, split, columnHandles);
        try (RecordCursor cursor = recordSet.cursor()) {
            while (cursor.advanceNextPosition()) {
                for (int i = 0; i < recordSet.getColumnTypes().size(); i++) {
                    cursor.isNull(i);
                }
            }
        }
    }
}

@Test
public void testGetRecordSet()
        throws Exception
{
    JdbcRecordSetProvider recordSetProvider = new JdbcRecordSetProvider(jdbcClient);
    RecordSet recordSet = recordSetProvider.getRecordSet(SESSION, split, ImmutableList.of(textColumn, valueColumn));
    assertNotNull(recordSet, "recordSet is null");

    RecordCursor cursor = recordSet.cursor();
    assertNotNull(cursor, "cursor is null");

    Map<String, Long> data = new LinkedHashMap<>();
    while (cursor.advanceNextPosition()) {
        data.put(cursor.getSlice(0).toStringUtf8(), cursor.getLong(1));
    }
    assertEquals(data, ImmutableMap.<String, Long>builder()
            .put("one", 1L)
            .put("two", 2L)
            .put("three", 3L)
            .put("ten", 10L)
            .put("eleven", 11L)
            .put("twelve", 12L)
            .build());
}

@Override
public RecordCursor cursor(ConnectorTransactionHandle transactionHandle, ConnectorSession connectorSession, TupleDomain<Integer> constraint)
{
    GlobalSystemTransactionHandle transaction = checkType(transactionHandle, GlobalSystemTransactionHandle.class, "transaction");
    Session session = toSession(transaction.getTransactionId(), connectorSession);
    Optional<String> catalogFilter = stringFilter(constraint, 0);
    Optional<String> schemaFilter = stringFilter(constraint, 1);
    Optional<String> tableFilter = stringFilter(constraint, 2);

    Builder table = InMemoryRecordSet.builder(METADATA);
    for (String catalog : filter(metadata.getCatalogNames().keySet(), catalogFilter)) {
        QualifiedTablePrefix prefix = FilterUtil.tablePrefix(catalog, schemaFilter, tableFilter);
        for (Entry<QualifiedObjectName, List<ColumnMetadata>> entry : metadata.listTableColumns(session, prefix).entrySet()) {
            addColumnRows(table, entry.getKey(), entry.getValue());
        }
    }
    return table.build().cursor();
}

@Override
public RecordCursor cursor(ConnectorTransactionHandle transactionHandle, ConnectorSession session, TupleDomain<Integer> constraint)
{
    Builder table = InMemoryRecordSet.builder(transactionsTable);
    for (TransactionInfo info : transactionManager.getAllTransactionInfos()) {
        table.addRow(
                info.getTransactionId().toString(),
                info.getIsolationLevel().toString(),
                info.isReadOnly(),
                info.isAutoCommitContext(),
                info.getCreateTime().getMillis(),
                (long) info.getIdleTime().getValue(TimeUnit.SECONDS),
                info.getWrittenConnectorId().orElse(null),
                createStringsBlock(info.getConnectorIds()));
    }
    return table.build().cursor();
}

private static RecordSet toRecordSet(ConnectorTransactionHandle sourceTransaction, SystemTable table, ConnectorSession session, TupleDomain<Integer> constraint)
{
    return new RecordSet()
    {
        private final List<Type> types = table.getTableMetadata().getColumns().stream()
                .map(ColumnMetadata::getType)
                .collect(toImmutableList());

        @Override
        public List<Type> getColumnTypes()
        {
            return types;
        }

        @Override
        public RecordCursor cursor()
        {
            return table.cursor(sourceTransaction, session, constraint);
        }
    };
}

@Override
public RecordCursor cursor(ConnectorTransactionHandle transactionHandle, ConnectorSession session, TupleDomain<Integer> constraint)
{
    Builder table = InMemoryRecordSet.builder(TABLE_PROPERTIES_TABLE);
    Map<String, Map<String, PropertyMetadata<?>>> tableProperties = new TreeMap<>(metadata.getTablePropertyManager().getAllTableProperties());
    for (Entry<String, Map<String, PropertyMetadata<?>>> entry : tableProperties.entrySet()) {
        String catalog = entry.getKey();
        Map<String, PropertyMetadata<?>> properties = new TreeMap<>(entry.getValue());
        for (PropertyMetadata<?> propertyMetadata : properties.values()) {
            table.addRow(
                    catalog,
                    propertyMetadata.getName(),
                    firstNonNull(propertyMetadata.getDefaultValue(), "").toString(),
                    propertyMetadata.getSqlType().toString(),
                    propertyMetadata.getDescription());
        }
    }
    return table.build().cursor();
}

@Override
public RecordCursor cursor(ConnectorTransactionHandle transactionHandle, ConnectorSession session, TupleDomain<Integer> constraint)
{
    Builder table = InMemoryRecordSet.builder(QUERY_TABLE);
    for (QueryInfo queryInfo : queryManager.getAllQueryInfo()) {
        QueryStats queryStats = queryInfo.getQueryStats();
        table.addRow(
                nodeId,
                queryInfo.getQueryId().toString(),
                queryInfo.getState().toString(),
                queryInfo.getSession().getUser(),
                queryInfo.getSession().getSource().orElse(null),
                queryInfo.getQuery(),

                toMillis(queryStats.getQueuedTime()),
                toMillis(queryStats.getAnalysisTime()),
                toMillis(queryStats.getDistributedPlanningTime()),

                toTimeStamp(queryStats.getCreateTime()),
                toTimeStamp(queryStats.getExecutionStartTime()),
                toTimeStamp(queryStats.getLastHeartbeat()),
                toTimeStamp(queryStats.getEndTime()));
    }
    return table.build().cursor();
}

@Override
public int process(ConnectorSession session, RecordCursor cursor, int count, PageBuilder pageBuilder)
{
    checkArgument(!pageBuilder.isFull(), "page builder can't be full");
    checkArgument(count > 0, "count must be > 0");

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

        if (!cursor.advanceNextPosition()) {
            break;
        }

        if (filterFunction.filter(cursor)) {
            pageBuilder.declarePosition();
            for (int channel = 0; channel < projections.size(); channel++) {
                // todo: if the projection function increases the size of the data significantly, this could cause the servers to OOM
                projections.get(channel).project(cursor, pageBuilder.getBlockBuilder(channel));
            }
        }
    }
    return position;
}

private void generateProjectMethod(ClassDefinition classDefinition, CallSiteBinder callSiteBinder, CachedInstanceBinder cachedInstanceBinder, String methodName, RowExpression projection)
{
    Parameter session = arg("session", ConnectorSession.class);
    Parameter cursor = arg("cursor", RecordCursor.class);
    Parameter output = arg("output", BlockBuilder.class);
    MethodDefinition method = classDefinition.declareMethod(a(PUBLIC), methodName, type(void.class), session, cursor, output);

    method.comment("Projection: %s", projection.toString());

    Scope scope = method.getScope();
    Variable wasNullVariable = scope.declareVariable(type(boolean.class), "wasNull");

    BytecodeBlock body = method.getBody()
            .comment("boolean wasNull = false;")
            .putVariable(wasNullVariable, false);

    BytecodeExpressionVisitor visitor = new BytecodeExpressionVisitor(callSiteBinder, cachedInstanceBinder, fieldReferenceCompiler(cursor, wasNullVariable), metadata.getFunctionRegistry());

    body.getVariable(output)
            .comment("evaluate projection: " + projection.toString())
            .append(projection.accept(visitor, scope))
            .append(generateWrite(callSiteBinder, scope, wasNullVariable, projection.getType()))
            .ret();
}

@Test
public void test()
{
    ArrayType arrayOfBigintType = new ArrayType(BIGINT);
    FieldSetFilteringRecordSet fieldSetFilteringRecordSet = new FieldSetFilteringRecordSet(
            FUNCTION_REGISTRY,
            new InMemoryRecordSet(
                    ImmutableList.of(BIGINT, BIGINT, TIMESTAMP_WITH_TIME_ZONE, TIMESTAMP_WITH_TIME_ZONE, arrayOfBigintType, arrayOfBigintType),
                    ImmutableList.of(
                            ImmutableList.of(
                                    100L,
                                    100L,
                                    // test same time in different time zone to make sure equal check was done properly
                                    packDateTimeWithZone(100, getTimeZoneKeyForOffset(123)),
                                    packDateTimeWithZone(100, getTimeZoneKeyForOffset(234)),
                                    // test structural type
                                    arrayBlockOf(BIGINT, 12, 34, 56),
                                    arrayBlockOf(BIGINT, 12, 34, 56)
                            ))),
            ImmutableList.of(ImmutableSet.of(0, 1), ImmutableSet.of(2, 3), ImmutableSet.of(4, 5)));
    RecordCursor recordCursor = fieldSetFilteringRecordSet.cursor();
    assertTrue(recordCursor.advanceNextPosition());
}

@Override
public RecordCursor cursor() {
    EthBlock block = null;
    try {
        block = web3j.ethGetBlockByNumber(DefaultBlockParameter.valueOf(BigInteger.valueOf(split.getBlockId())), true).send();
    } catch (IOException e) {
        e.printStackTrace();
    }
    return new EthereumRecordCursor(columnHandles, block, split.getTable(), web3j);
}

private static List<MaterializedRow> getMaterializedResults(RecordCursor cursor, List<ColumnMetadata> columns)
{
    List<Type> types = columns.stream().map(ColumnMetadata::getType).collect(toList());

    ImmutableList.Builder<MaterializedRow> rowBuilder = ImmutableList.builder();
    for (int i = 0; i < types.size(); i++) {
        assertEquals(cursor.getType(i), types.get(i));
    }

    while (cursor.advanceNextPosition()) {
        List<Object> values = new ArrayList<>(types.size());
        for (int i = 0; i < columns.size(); i++) {
            Type type = columns.get(i).getType();
            Class<?> javaType = type.getJavaType();
            if (cursor.isNull(i)) {
                continue;
            }
            if (javaType == boolean.class) {
                values.add(i, cursor.getBoolean(i));
            }
            else if (javaType == long.class) {
                values.add(i, cursor.getLong(i));
            }
            else if (javaType == double.class) {
                values.add(i, cursor.getDouble(i));
            }
            else if (javaType == Slice.class) {
                values.add(i, cursor.getSlice(i));
            }
        }
        rowBuilder.add(new MaterializedRow(DEFAULT_PRECISION, values));
    }
    return rowBuilder.build();
}

private AppendingRecordCursor(RecordCursor delegate, int delegateFieldCount, List<Object> appendedValues, List<Type> appendedTypes)
{
    this.delegate = requireNonNull(delegate, "delegate is null");
    this.delegateFieldCount = delegateFieldCount;
    checkArgument(delegateFieldCount >= 0, "delegateFieldCount must be greater than or equal to zero");
    this.appendedValues = requireNonNull(appendedValues, "appendedValues is null"); // May contain null elements
    this.appendedTypes = ImmutableList.copyOf(requireNonNull(appendedTypes, "appendedTypes is null"));
    checkArgument(appendedValues.size() == appendedTypes.size(), "appendedValues must have the same size as appendedTypes");
}

@Override
public RecordCursor cursor()
{
    // NOTE: the ConcatRecordCursor implementation relies on the fact that the
    // cursor creation in the Iterable is lazy so DO NOT materialize this into
    // an ImmutableList
    Iterable<RecordCursor> recordCursors = transform(recordSets, RecordSet::cursor);
    return new ConcatRecordCursor(recordCursors.iterator(), types);
}

private ConcatRecordCursor(Iterator<RecordCursor> iterator, List<Type> types)
{
    // NOTE: this cursor implementation relies on the fact that the
    // cursor creation in the Iterable is lazy so DO NOT materialize this into
    // an ImmutableList
    this.iterator = requireNonNull(iterator, "iterator is null");
    this.types = ImmutableList.copyOf(requireNonNull(types, "types is null"));
}

private static List<Object> extractValues(RecordCursor cursor, List<Type> types)
{
    List<Object> list = new ArrayList<>(types.size());
    for (int i = 0; i < types.size(); i++) {
        list.add(extractObject(cursor, i, types.get(i)));
    }
    return list;
}

@Test
public void testCursorSimple()
        throws Exception
{
    RecordSet recordSet = new JdbcRecordSet(jdbcClient, split, ImmutableList.of(
            columnHandles.get("text"),
            columnHandles.get("value")));

    try (RecordCursor cursor = recordSet.cursor()) {
        assertEquals(cursor.getType(0), VARCHAR);
        assertEquals(cursor.getType(1), BIGINT);

        Map<String, Long> data = new LinkedHashMap<>();
        while (cursor.advanceNextPosition()) {
            data.put(cursor.getSlice(0).toStringUtf8(), cursor.getLong(1));
            assertFalse(cursor.isNull(0));
            assertFalse(cursor.isNull(1));
        }

        assertEquals(data, ImmutableMap.<String, Long>builder()
                .put("one", 1L)
                .put("two", 2L)
                .put("three", 3L)
                .put("ten", 10L)
                .put("eleven", 11L)
                .put("twelve", 12L)
                .build());
    }
}

@Test
public void testCursorMixedOrder()
        throws Exception
{
    RecordSet recordSet = new JdbcRecordSet(jdbcClient, split, ImmutableList.of(
            columnHandles.get("value"),
            columnHandles.get("value"),
            columnHandles.get("text")));

    try (RecordCursor cursor = recordSet.cursor()) {
        assertEquals(cursor.getType(0), BIGINT);
        assertEquals(cursor.getType(1), BIGINT);
        assertEquals(cursor.getType(2), VARCHAR);

        Map<String, Long> data = new LinkedHashMap<>();
        while (cursor.advanceNextPosition()) {
            assertEquals(cursor.getLong(0), cursor.getLong(1));
            data.put(cursor.getSlice(2).toStringUtf8(), cursor.getLong(0));
        }

        assertEquals(data, ImmutableMap.<String, Long>builder()
                .put("one", 1L)
                .put("two", 2L)
                .put("three", 3L)
                .put("ten", 10L)
                .put("eleven", 11L)
                .put("twelve", 12L)
                .build());
    }
}

@Test
public void testIdempotentClose()
{
    RecordSet recordSet = new JdbcRecordSet(jdbcClient, split, ImmutableList.of(
            columnHandles.get("value"),
            columnHandles.get("value"),
            columnHandles.get("text")));

    RecordCursor cursor = recordSet.cursor();
    cursor.close();
    cursor.close();
}

private RecordCursor getCursor(JdbcTableHandle jdbcTableHandle, List<JdbcColumnHandle> columns, TupleDomain<ColumnHandle> domain)
        throws InterruptedException
{
    JdbcTableLayoutHandle layoutHandle = new JdbcTableLayoutHandle(jdbcTableHandle, domain);
    ConnectorSplitSource splits = jdbcClient.getSplits(layoutHandle);
    JdbcSplit split = (JdbcSplit) getOnlyElement(getFutureValue(splits.getNextBatch(1000)));

    JdbcRecordSetProvider recordSetProvider = new JdbcRecordSetProvider(jdbcClient);
    RecordSet recordSet = recordSetProvider.getRecordSet(SESSION, split, columns);

    return recordSet.cursor();
}

@Override
public RecordCursor cursor(ConnectorTransactionHandle transactionHandle, ConnectorSession session, TupleDomain<Integer> constraint)
{
    Builder table = InMemoryRecordSet.builder(METADATA);
    for (String name : metadata.getCatalogNames().keySet()) {
        table.addRow(name);
    }
    return table.build().cursor();
}

@Override
public RecordCursor cursor(ConnectorTransactionHandle transactionHandle, ConnectorSession connectorSession, TupleDomain<Integer> constraint)
{
    GlobalSystemTransactionHandle transaction = checkType(transactionHandle, GlobalSystemTransactionHandle.class, "transaction");
    Session session = toSession(transaction.getTransactionId(), connectorSession);
    Optional<String> catalogFilter = FilterUtil.stringFilter(constraint, 1);

    Builder table = InMemoryRecordSet.builder(METADATA);
    for (String catalog : filter(metadata.getCatalogNames().keySet(), catalogFilter)) {
        for (String schema : metadata.listSchemaNames(session, catalog)) {
            table.addRow(schema, catalog);
        }
    }
    return table.build().cursor();
}

@Override
public RecordCursor cursor(ConnectorTransactionHandle transactionHandle, ConnectorSession session, TupleDomain<Integer> constraint)
{
    return InMemoryRecordSet.builder(METADATA)
            .addRow("TABLE")
            .addRow("VIEW")
            .build().cursor();
}

@Override
public RecordCursor cursor(ConnectorTransactionHandle transactionHandle, ConnectorSession session, TupleDomain<Integer> constraint)
{
    Builder table = InMemoryRecordSet.builder(CATALOG_TABLE);
    for (Map.Entry<String, String> entry : metadata.getCatalogNames().entrySet()) {
        table.addRow(entry.getKey(), entry.getValue());
    }
    return table.build().cursor();
}

@Override
public RecordCursor cursor(ConnectorTransactionHandle transactionHandle, ConnectorSession session, TupleDomain<Integer> constraint)
{
    Builder table = InMemoryRecordSet.builder(TASK_TABLE);
    for (TaskInfo taskInfo : taskManager.getAllTaskInfo()) {
        TaskStats stats = taskInfo.getStats();
        table.addRow(
                nodeId,

                taskInfo.getTaskId().toString(),
                taskInfo.getTaskId().getStageId().toString(),
                taskInfo.getTaskId().getQueryId().toString(),
                taskInfo.getState().toString(),

                (long) stats.getTotalDrivers(),
                (long) stats.getQueuedDrivers(),
                (long) stats.getRunningDrivers(),
                (long) stats.getCompletedDrivers(),

                toMillis(stats.getTotalScheduledTime()),
                toMillis(stats.getTotalCpuTime()),
                toMillis(stats.getTotalUserTime()),
                toMillis(stats.getTotalBlockedTime()),

                toBytes(stats.getRawInputDataSize()),
                stats.getRawInputPositions(),

                toBytes(stats.getProcessedInputDataSize()),
                stats.getProcessedInputPositions(),

                toBytes(stats.getOutputDataSize()),
                stats.getOutputPositions(),

                toTimeStamp(stats.getCreateTime()),
                toTimeStamp(stats.getFirstStartTime()),
                toTimeStamp(taskInfo.getLastHeartbeat()),
                toTimeStamp(stats.getEndTime()));
    }
    return table.build().cursor();
}

@Override
public RecordCursor cursor(ConnectorTransactionHandle transactionHandle, ConnectorSession session, TupleDomain<Integer> constraint)
{
    Builder table = InMemoryRecordSet.builder(NODES_TABLE);
    AllNodes allNodes = nodeManager.getAllNodes();
    addRows(table, allNodes.getActiveNodes(), ACTIVE);
    addRows(table, allNodes.getInactiveNodes(), INACTIVE);
    addRows(table, allNodes.getShuttingDownNodes(), SHUTTING_DOWN);
    return table.build().cursor();
}

@Override
public void project(RecordCursor cursor, BlockBuilder output)
{
    // record cursors have each value in a separate field
    if (cursor.isNull(channelIndex)) {
        output.appendNull();
    }
    else {
        Class<?> javaType = columnType.getJavaType();
        if (javaType == boolean.class) {
            columnType.writeBoolean(output, cursor.getBoolean(channelIndex));
        }
        else if (javaType == long.class) {
            columnType.writeLong(output, cursor.getLong(channelIndex));
        }
        else if (javaType == double.class) {
            columnType.writeDouble(output, cursor.getDouble(channelIndex));
        }
        else if (javaType == Slice.class) {
            Slice slice = cursor.getSlice(channelIndex);
            columnType.writeSlice(output, slice, 0, slice.length());
        }
        else {
            throw new UnsupportedOperationException("not yet implemented: " + javaType);
        }
    }
}

public RecordProjectOperator(OperatorContext operatorContext, List<Type> columnTypes, RecordCursor cursor)
{
    this.operatorContext = requireNonNull(operatorContext, "operatorContext is null");
    this.cursor = requireNonNull(cursor, "cursor is null");

    ImmutableList.Builder<Type> types = ImmutableList.builder();
    for (Type columnType : columnTypes) {
        types.add(columnType);
    }
    this.types = types.build();

    pageBuilder = new PageBuilder(getTypes());
}

private static boolean fieldSetsEqual(RecordCursor cursor, List<Set<Field>> fieldSets)
{
    for (Set<Field> fieldSet : fieldSets) {
        if (!fieldsEquals(cursor, fieldSet)) {
            return false;
        }
    }
    return true;
}

private static boolean fieldsEquals(RecordCursor cursor, Set<Field> fields)
{
    if (fields.size() < 2) {
        return true; // Nothing to compare
    }
    Iterator<Field> fieldIterator = fields.iterator();
    Field firstField = fieldIterator.next();
    while (fieldIterator.hasNext()) {
        if (!fieldEquals(cursor, firstField, fieldIterator.next())) {
            return false;
        }
    }
    return true;
}

private static boolean fieldEquals(RecordCursor cursor, Field field1, Field field2)
{
    checkArgument(cursor.getType(field1.getField()).equals(cursor.getType(field2.getField())), "Should only be comparing fields of the same type");

    if (cursor.isNull(field1.getField()) || cursor.isNull(field2.getField())) {
        return false;
    }

    Class<?> javaType = cursor.getType(field1.getField()).getJavaType();
    try {
        if (javaType == long.class) {
            return (boolean) field1.getEqualsMethodHandle().invokeExact(cursor.getLong(field1.getField()), cursor.getLong(field2.getField()));
        }
        else if (javaType == double.class) {
            return (boolean) field1.getEqualsMethodHandle().invokeExact(cursor.getDouble(field1.getField()), cursor.getDouble(field2.getField()));
        }
        else if (javaType == boolean.class) {
            return (boolean) field1.getEqualsMethodHandle().invokeExact(cursor.getBoolean(field1.getField()), cursor.getBoolean(field2.getField()));
        }
        else if (javaType == Slice.class) {
            return (boolean) field1.getEqualsMethodHandle().invokeExact(cursor.getSlice(field1.getField()), cursor.getSlice(field2.getField()));
        }
        else {
            return (boolean) field1.getEqualsMethodHandle().invoke(cursor.getObject(field1.getField()), cursor.getObject(field2.getField()));
        }
    }
    catch (Throwable throwable) {
        throw Throwables.propagate(throwable);
    }
}

private void generateFilterMethod(ClassDefinition classDefinition, CallSiteBinder callSiteBinder, CachedInstanceBinder cachedInstanceBinder, RowExpression filter)
{
    Parameter session = arg("session", ConnectorSession.class);
    Parameter cursor = arg("cursor", RecordCursor.class);
    MethodDefinition method = classDefinition.declareMethod(a(PUBLIC), "filter", type(boolean.class), session, cursor);

    method.comment("Filter: %s", filter);

    Scope scope = method.getScope();
    Variable wasNullVariable = scope.declareVariable(type(boolean.class), "wasNull");

    BytecodeExpressionVisitor visitor = new BytecodeExpressionVisitor(callSiteBinder, cachedInstanceBinder, fieldReferenceCompiler(cursor, wasNullVariable), metadata.getFunctionRegistry());

    LabelNode end = new LabelNode("end");
    method.getBody()
            .comment("boolean wasNull = false;")
            .putVariable(wasNullVariable, false)
            .comment("evaluate filter: " + filter)
            .append(filter.accept(visitor, scope))
            .comment("if (wasNull) return false;")
            .getVariable(wasNullVariable)
            .ifFalseGoto(end)
            .pop(boolean.class)
            .push(false)
            .visitLabel(end)
            .retBoolean();
}

@Override
public void project(RecordCursor cursor, BlockBuilder output)
{
    if (cursor.isNull(channelIndex)) {
        output.appendNull();
    }
    else {
        BIGINT.writeLong(output, cursor.getLong(channelIndex) + 5);
    }
}

private static Object evaluate(Expression expression)
{
    IdentityHashMap<Expression, Type> expressionTypes = getExpressionTypes(TEST_SESSION, METADATA, SQL_PARSER, SYMBOL_TYPES, expression);
    ExpressionInterpreter interpreter = expressionInterpreter(expression, METADATA, TEST_SESSION, expressionTypes);

    return interpreter.evaluate((RecordCursor) null);
}