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

@Override
public RecordSet getRecordSet(
        ConnectorTransactionHandle transaction,
        ConnectorSession session,
        ConnectorSplit split,
        List<? extends ColumnHandle> columns
) {
    EthereumSplit ethereumSplit = convertSplit(split);

    ImmutableList.Builder<EthereumColumnHandle> handleBuilder = ImmutableList.builder();

    for (ColumnHandle handle : columns) {
        EthereumColumnHandle columnHandle = convertColumnHandle(handle);
        handleBuilder.add(columnHandle);
    }

    return new EthereumRecordSet(web3j, handleBuilder.build(), ethereumSplit);
}

@Override
/**
 * @
 */
public RecordSet getRecordSet(ConnectorTransactionHandle transactionHandle, ConnectorSession session, ConnectorSplit split, List<? extends ColumnHandle> columns)
{
    requireNonNull(split, "split is null");
    KuduSplit kuduSplit = checkType(split, KuduSplit.class, "split");

    ImmutableList.Builder<KuduColumnHandle> handles = ImmutableList.builder();
    for (ColumnHandle handle : columns) {
        handles.add(checkType(handle, KuduColumnHandle.class, "handle"));
    }

    return new KuduRecordSet(kuduTable, kuduClientManager, kuduSplit, handles.build());
}

@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 testGetColumnTypes()
        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)));
    assertEquals(recordSet.getColumnTypes(), ImmutableList.of(VARCHAR, BIGINT));

    recordSet = new ExampleRecordSet(new ExampleSplit("test", "schema", "table", dataUri), ImmutableList.of(
            new ExampleColumnHandle("test", "value", BIGINT, 1),
            new ExampleColumnHandle("test", "text", VARCHAR, 0)));
    assertEquals(recordSet.getColumnTypes(), ImmutableList.of(BIGINT, VARCHAR));

    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)));
    assertEquals(recordSet.getColumnTypes(), ImmutableList.of(BIGINT, BIGINT, VARCHAR));

    recordSet = new ExampleRecordSet(new ExampleSplit("test", "schema", "table", dataUri), ImmutableList.<ExampleColumnHandle>of());
    assertEquals(recordSet.getColumnTypes(), ImmutableList.of());
}

@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 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 testGetColumnTypes()
        throws Exception
{
    RecordSet recordSet = new JdbcRecordSet(jdbcClient, split, ImmutableList.of(
            new JdbcColumnHandle("test", "text", VARCHAR),
            new JdbcColumnHandle("test", "value", BIGINT)));
    assertEquals(recordSet.getColumnTypes(), ImmutableList.of(VARCHAR, BIGINT));

    recordSet = new JdbcRecordSet(jdbcClient, split, ImmutableList.of(
            new JdbcColumnHandle("test", "value", BIGINT),
            new JdbcColumnHandle("test", "text", VARCHAR)));
    assertEquals(recordSet.getColumnTypes(), ImmutableList.of(BIGINT, VARCHAR));

    recordSet = new JdbcRecordSet(jdbcClient, split, ImmutableList.of(
            new JdbcColumnHandle("test", "value", BIGINT),
            new JdbcColumnHandle("test", "value", BIGINT),
            new JdbcColumnHandle("test", "text", VARCHAR)));
    assertEquals(recordSet.getColumnTypes(), ImmutableList.of(BIGINT, BIGINT, VARCHAR));

    recordSet = new JdbcRecordSet(jdbcClient, split, ImmutableList.<JdbcColumnHandle>of());
    assertEquals(recordSet.getColumnTypes(), ImmutableList.of());
}

@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());
}

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);
        }
    };
}

public FieldSetFilteringRecordSet(FunctionRegistry functionRegistry, RecordSet delegate, List<Set<Integer>> fieldSets)
{
    requireNonNull(functionRegistry, "functionRegistry is null");
    this.delegate = requireNonNull(delegate, "delegate is null");

    ImmutableList.Builder<Set<Field>> fieldSetsBuilder = ImmutableList.builder();
    List<Type> columnTypes = delegate.getColumnTypes();
    for (Set<Integer> fieldSet : requireNonNull(fieldSets, "fieldSets is null")) {
        ImmutableSet.Builder<Field> fieldSetBuilder = ImmutableSet.builder();
        for (int field : fieldSet) {
            fieldSetBuilder.add(new Field(
                    field,
                    functionRegistry.getScalarFunctionImplementation(internalOperator(OperatorType.EQUAL, BooleanType.BOOLEAN, ImmutableList.of(columnTypes.get(field), columnTypes.get(field)))).getMethodHandle()));
        }
        fieldSetsBuilder.add(fieldSetBuilder.build());
    }
    this.fieldSets = fieldSetsBuilder.build();
}

@Override
public Supplier<Optional<UpdatablePageSource>> addSplit(Split split)
{
    requireNonNull(split, "split is null");
    checkType(split.getConnectorSplit(), IndexSplit.class, "connectorSplit");
    checkState(source == null, "Index source split already set");

    IndexSplit indexSplit = (IndexSplit) split.getConnectorSplit();

    // Normalize the incoming RecordSet to something that can be consumed by the index
    RecordSet normalizedRecordSet = probeKeyNormalizer.apply(indexSplit.getKeyRecordSet());
    RecordSet result = index.lookup(normalizedRecordSet);
    source = new PageSourceOperator(new RecordPageSource(result), result.getColumnTypes(), operatorContext);

    operatorContext.setInfoSupplier(split::getInfo);

    return Optional::empty;
}

@Override
public ConnectorPageSource createPageSource(Session session, Split split, List<ColumnHandle> columns)
{
    assertInstanceOf(split.getConnectorSplit(), FunctionAssertions.TestSplit.class);
    FunctionAssertions.TestSplit testSplit = (FunctionAssertions.TestSplit) split.getConnectorSplit();
    if (testSplit.isRecordSet()) {
        RecordSet records = InMemoryRecordSet.builder(ImmutableList.<Type>of(BIGINT, VARCHAR, DOUBLE, BOOLEAN, BIGINT, VARCHAR, VARCHAR, TIMESTAMP_WITH_TIME_ZONE, VARBINARY))
                .addRow(
                        1234L,
                        "hello",
                        12.34,
                        true,
                        new DateTime(2001, 8, 22, 3, 4, 5, 321, DateTimeZone.UTC).getMillis(),
                        "%el%",
                        null,
                        packDateTimeWithZone(new DateTime(1970, 1, 1, 0, 1, 0, 999, DateTimeZone.UTC).getMillis(), TimeZoneKey.getTimeZoneKey("Z")),
                        Slices.wrappedBuffer((byte) 0xab))
                .build();
        return new RecordPageSource(records);
    }
    else {
        return new FixedPageSource(ImmutableList.of(SOURCE_PAGE));
    }
}

public <E extends TpchEntity> RecordSet getRecordSet(
        TpchTable<E> table,
        List<? extends ColumnHandle> columns,
        double scaleFactor,
        int partNumber,
        int totalParts)
{
    ImmutableList.Builder<TpchColumn<E>> builder = ImmutableList.builder();
    for (ColumnHandle column : columns) {
        String columnName = checkType(column, TpchColumnHandle.class, "column").getColumnName();
        if (columnName.equalsIgnoreCase(TpchMetadata.ROW_NUMBER_COLUMN_NAME)) {
            builder.add(new RowNumberTpchColumn<E>());
        }
        else {
            builder.add(table.getColumn(columnName));
        }
    }

    return createTpchRecordSet(table, builder.build(), scaleFactor, partNumber + 1, totalParts);
}

@Override
public RecordSet getRecordSet(ConnectorTransactionHandle transaction, ConnectorSession session, ConnectorSplit split, List<? extends ColumnHandle> columns)
{
    CassandraSplit cassandraSplit = checkType(split, CassandraSplit.class, "split");

    List<CassandraColumnHandle> cassandraColumns = columns.stream()
            .map(column -> checkType(column, CassandraColumnHandle.class, "columnHandle"))
            .collect(toList());

    String selectCql = CassandraCqlUtils.selectFrom(cassandraSplit.getCassandraTableHandle(), cassandraColumns).getQueryString();
    StringBuilder sb = new StringBuilder(selectCql);
    if (sb.charAt(sb.length() - 1) == ';') {
        sb.setLength(sb.length() - 1);
    }
    sb.append(cassandraSplit.getWhereClause());
    String cql = sb.toString();
    log.debug("Creating record set: %s", cql);

    return new CassandraRecordSet(cassandraSession, cassandraSplit.getSchema(), cql, cassandraColumns);
}

@Override
public RecordSet getRecordSet(
        ConnectorTransactionHandle connectorTransactionHandle,
        ConnectorSession connectorSession,
        ConnectorSplit connectorSplit,
        List<? extends ColumnHandle> list)
{
    RestConnectorSplit split = Types.checkType(connectorSplit, RestConnectorSplit.class, "split");
    // TODO fix below cast
    List<RestColumnHandle> restColumnHandles = (List<RestColumnHandle>) list;

    SchemaTableName schemaTableName = split.getTableHandle().getSchemaTableName();
    Collection<? extends List<?>> rows = rest.getRows(schemaTableName);
    ConnectorTableMetadata tableMetadata = rest.getTableMetadata(schemaTableName);

    List<Integer> columnIndexes = restColumnHandles.stream()
            .map(column -> {
                int index = 0;
                for (ColumnMetadata columnMetadata : tableMetadata.getColumns()) {
                    if (columnMetadata.getName().equalsIgnoreCase(column.getName())) {
                        return index;
                    }
                    index++;
                }
                throw new IllegalStateException("Unknown column: " + column.getName());
            })
            .collect(toList());

    Collection<? extends List<?>> mappedRows = rows.stream()
            .map(row -> columnIndexes.stream()
                    .map(index -> row.get(index))
                    .collect(toList()))
            .collect(toList());

    List<Type> mappedTypes = restColumnHandles.stream()
            .map(RestColumnHandle::getType)
            .collect(toList());
    return new InMemoryRecordSet(mappedTypes, mappedRows);
}

@Override
public RecordSet getRecordSet(ConnectorTransactionHandle connectorTransactionHandle, ConnectorSession connectorSession, ConnectorSplit connectorSplit, List<? extends ColumnHandle> list)
{
    log.info("INFORMATION: AmpoolRecordSetProvider getRecordSet() called.");

    requireNonNull(connectorSplit, "split is null");
    AmpoolSplit ampoolSplit = (AmpoolSplit) connectorSplit;
    checkArgument(ampoolSplit.getConnectorId().equals(connectorId), "split is not for this connector");

    ImmutableList.Builder<AmpoolColumnHandle> handles = ImmutableList.builder();
    for (ColumnHandle handle : list)
    {
        handles.add((AmpoolColumnHandle) handle);
    }


    // TODO: Projections and filters on Ampool side
    Iterator<Row> iterator;
    if (ampoolClient.existsFTable(ampoolSplit.getTableName()))
        iterator = ampoolClient.getFTable(ampoolSplit.getTableName()).getScanner(new Scan()).iterator();
    else if (ampoolClient.existsMTable(ampoolSplit.getTableName()))
        iterator = ampoolClient.getMTable(ampoolSplit.getTableName()).getScanner(new Scan()).iterator();
    else
        iterator = null;

    return new AmpoolRecordSet(ampoolSplit, handles.build(), iterator);
}

@Override
public RecordSet getRecordSet(ConnectorSession session, ConnectorSplit split, List<? extends ColumnHandle> columns)
{
    requireNonNull(split, "partitionChunk is null");
    ExampleSplit exampleSplit = checkType(split, ExampleSplit.class, "split");
    checkArgument(exampleSplit.getConnectorId().equals(connectorId), "split is not for this connector");

    ImmutableList.Builder<ExampleColumnHandle> handles = ImmutableList.builder();
    for (ColumnHandle handle : columns) {
        handles.add(checkType(handle, ExampleColumnHandle.class, "handle"));
    }

    return new ExampleRecordSet(exampleSplit, handles.build());
}

public AppendingRecordSet(RecordSet delegate, List<Object> appendedValues, List<Type> appendedTypes)
{
    this.delegate = requireNonNull(delegate, "delegate is null");
    this.appendedValues = new ArrayList<>(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");
    for (int i = 0; i < appendedValues.size(); i++) {
        Object value = appendedValues.get(i);
        if (value != null) {
            checkArgument(appendedTypes.get(i).getJavaType().isInstance(value), "Object value does not match declared type");
        }
    }
}

public ConcatRecordSet(Iterable<RecordSet> recordSets, List<Type> types)
{
    this.recordSets = requireNonNull(recordSets, "recordSets is null");
    for (RecordSet recordSet : recordSets) {
        checkState(recordSet.getColumnTypes().equals(types), "RecordSet types do not match declared types");
    }
    this.types = ImmutableList.copyOf(requireNonNull(types, "types is null"));
}

@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);
}

@Override
public RecordSet lookup(RecordSet rawInputRecordSet)
{
    // convert the input record set from the column ordering in the query to
    // match the column ordering of the index
    RecordSet inputRecordSet = keyFormatter.apply(rawInputRecordSet);

    // lookup the values in the index
    RecordSet rawOutputRecordSet = indexedTable.lookupKeys(inputRecordSet);

    // convert the output record set of the index into the column ordering
    // expect by the query
    return outputFormatter.apply(rawOutputRecordSet);
}

public TpchIndexedData(String connectorId, TpchIndexSpec tpchIndexSpec)
{
    requireNonNull(connectorId, "connectorId is null");
    requireNonNull(tpchIndexSpec, "tpchIndexSpec is null");

    TpchMetadata tpchMetadata = new TpchMetadata(connectorId);
    TpchRecordSetProvider tpchRecordSetProvider = new TpchRecordSetProvider();

    ImmutableMap.Builder<Set<TpchScaledColumn>, IndexedTable> indexedTablesBuilder = ImmutableMap.builder();

    Set<TpchScaledTable> tables = tpchIndexSpec.listIndexedTables();
    for (TpchScaledTable table : tables) {
        SchemaTableName tableName = new SchemaTableName("sf" + table.getScaleFactor(), table.getTableName());
        TpchTableHandle tableHandle = tpchMetadata.getTableHandle(null, tableName);
        Map<String, ColumnHandle> columnHandles = new LinkedHashMap<>(tpchMetadata.getColumnHandles(null, tableHandle));
        for (Set<String> columnNames : tpchIndexSpec.getColumnIndexes(table)) {
            List<String> keyColumnNames = ImmutableList.copyOf(columnNames); // Finalize the key order
            Set<TpchScaledColumn> keyColumns = FluentIterable.from(keyColumnNames)
                    .transform(TpchScaledColumn.columnFunction(table))
                    .toSet();

            TpchTable<?> tpchTable = TpchTable.getTable(table.getTableName());
            RecordSet recordSet = tpchRecordSetProvider.getRecordSet(tpchTable, ImmutableList.copyOf(columnHandles.values()), table.getScaleFactor(), 0, 1);
            IndexedTable indexedTable = indexTable(recordSet, ImmutableList.copyOf(columnHandles.keySet()), keyColumnNames);
            indexedTablesBuilder.put(keyColumns, indexedTable);
        }
    }

    indexedTables = indexedTablesBuilder.build();
}

public RecordSet lookupKeys(RecordSet recordSet)
{
    checkArgument(recordSet.getColumnTypes().equals(keyTypes), "Input RecordSet keys do not match expected key type");

    Iterable<RecordSet> outputRecordSets = Iterables.transform(tupleIterable(recordSet), key -> {
        for (Object value : key.getValues()) {
            if (value == null) {
                throw new IllegalArgumentException("TPCH index does not support null values");
            }
        }
        return lookupKey(key);
    });

    return new ConcatRecordSet(outputRecordSets, outputTypes);
}

@Override
public RecordSet getRecordSet(ConnectorSession session, ConnectorSplit split, List<? extends ColumnHandle> columns)
{
    JdbcSplit jdbcSplit = checkType(split, JdbcSplit.class, "split");

    ImmutableList.Builder<JdbcColumnHandle> handles = ImmutableList.builder();
    for (ColumnHandle handle : columns) {
        handles.add(checkType(handle, JdbcColumnHandle.class, "columnHandle"));
    }

    return new JdbcRecordSet(jdbcClient, jdbcSplit, handles.build());
}

@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();
}

public IndexSourceOperatorFactory(
        int operatorId,
        PlanNodeId sourceId,
        ConnectorIndex index,
        List<Type> types,
        Function<RecordSet, RecordSet> probeKeyNormalizer)
{
    this.operatorId = operatorId;
    this.sourceId = requireNonNull(sourceId, "sourceId is null");
    this.index = requireNonNull(index, "index is null");
    this.types = requireNonNull(types, "types is null");
    this.probeKeyNormalizer = requireNonNull(probeKeyNormalizer, "probeKeyNormalizer is null");
}

public IndexSourceOperator(
        OperatorContext operatorContext,
        PlanNodeId planNodeId,
        ConnectorIndex index,
        List<Type> types,
        Function<RecordSet, RecordSet> probeKeyNormalizer)
{
    this.operatorContext = requireNonNull(operatorContext, "operatorContext is null");
    this.planNodeId = requireNonNull(planNodeId, "planNodeId is null");
    this.index = requireNonNull(index, "index is null");
    this.types = ImmutableList.copyOf(requireNonNull(types, "types is null"));
    this.probeKeyNormalizer = requireNonNull(probeKeyNormalizer, "probeKeyNormalizer is null");
}

public MappedRecordSet(RecordSet delegate, List<Integer> delegateFieldIndex)
{
    this.delegate = requireNonNull(delegate, "delegate is null");
    this.delegateFieldIndex = Ints.toArray(requireNonNull(delegateFieldIndex, "delegateFieldIndex is null"));

    List<Type> types = delegate.getColumnTypes();
    this.columnTypes = delegateFieldIndex.stream().map(types::get).collect(toImmutableList());
}

@Override
public RecordSet getRecordSet(ConnectorTransactionHandle transaction, ConnectorSession session, ConnectorSplit split, List<? extends ColumnHandle> columns)
{
    TpchSplit tpchSplit = checkType(split, TpchSplit.class, "split");

    String tableName = tpchSplit.getTableHandle().getTableName();

    TpchTable<?> tpchTable = TpchTable.getTable(tableName);

    return getRecordSet(tpchTable, columns, tpchSplit.getTableHandle().getScaleFactor(), tpchSplit.getPartNumber(), tpchSplit.getTotalParts());
}

@Override
public RecordSet getRecordSet(ConnectorSession session, ConnectorSplit split, List<? extends ColumnHandle> columns)
{
    try (ThreadContextClassLoader ignored = new ThreadContextClassLoader(classLoader)) {
        return new ClassLoaderSafeRecordSet(delegate.getRecordSet(session, split, columns), classLoader);
    }
}

@Override
public RecordSet getRecordSet(ConnectorTransactionHandle transactionHandle, ConnectorSession session,
    ConnectorSplit split, List<? extends ColumnHandle> columns) {
  SpreadsheetSplit spreadsheetSplit = (SpreadsheetSplit) split;
  SpreadsheetTableHandle spreadsheetTableHandle = spreadsheetSplit.getTableHandle();
  SchemaTableName schemaTableName = spreadsheetTableHandle.getTableName();
  UserGroupInformation proxy = SpreadsheetMetadata.getUgi(session, _proxyUser, _ugi);
  SpreadsheetReader spreadSheetHelper = SpreadsheetMetadata.getSpreadSheetHelper(proxy, session,
      spreadsheetTableHandle, _configuration, _useFileCache);
  return new SpreadsheetRecordSet(schemaTableName.getTableName(), spreadSheetHelper, columns);
}

@Override
public RecordSet getRecordSet(ConnectorTransactionHandle transactionHandle, ConnectorSession session,
                              ConnectorSplit split, List<? extends ColumnHandle> columns)
{
    KinesisSplit kinesisSplit = handleResolver.convertSplit(split);

    ImmutableList.Builder<KinesisColumnHandle> handleBuilder = ImmutableList.builder();
    ImmutableMap.Builder<KinesisColumnHandle, KinesisFieldDecoder<?>> messageFieldDecoderBuilder = ImmutableMap.builder();

    KinesisRowDecoder messageDecoder = registry.getRowDecoder(kinesisSplit.getMessageDataFormat());

    for (ColumnHandle handle : columns) {
        KinesisColumnHandle columnHandle = handleResolver.convertColumnHandle(handle);
        handleBuilder.add(columnHandle);

        if (!columnHandle.isInternal()) {
            KinesisFieldDecoder<?> fieldDecoder = registry.getFieldDecoder(kinesisSplit.getMessageDataFormat(),
                    columnHandle.getType().getJavaType(),
                    columnHandle.getDataFormat());

            messageFieldDecoderBuilder.put(columnHandle, fieldDecoder);
        }
    }

    ImmutableList<KinesisColumnHandle> handles = handleBuilder.build();
    ImmutableMap<KinesisColumnHandle, KinesisFieldDecoder<?>> messageFieldDecoders = messageFieldDecoderBuilder.build();

    return new KinesisRecordSet(kinesisSplit, session, clientManager, handles, messageDecoder, messageFieldDecoders, kinesisConnectorConfig);
}

@Override
public RecordSet getRecordSet(ConnectorSplit split,
                              List<? extends ColumnHandle> columns) {
    checkNotNull(split, "partitionChunk is null");
    checkArgument(split instanceof CoverageSplit);
    //log.debug("getRecordSet");

    CoverageSplit coverageSplit = (CoverageSplit) split;
    checkArgument(coverageSplit.getTableHandle().getConnectorId()
            .equals(connectorId));

    PRTable table = coverageSplit.getTable();

    ImmutableList.Builder<RiakColumnHandle> handles = ImmutableList.builder();
    for (ColumnHandle handle : columns) {
        checkArgument(handle instanceof RiakColumnHandle);
        RiakColumnHandle riakColumnHandle = (RiakColumnHandle) handle;
        boolean has2i = false;//TODO: table.hasIndex(riakColumnHandle.getColumn().getName());
        riakColumnHandle.getColumn().setIndex(has2i);
        handles.add(riakColumnHandle);
    }

    //log.debug("supplying CoverageRecordSet");
    return new CoverageRecordSet(coverageSplit,
            handles.build(),
            riakConfig,
            coverageSplit.getTupleDomain(),
            directConnection);
}

@Override
public ConnectorIndex getIndex(
        ConnectorTransactionHandle transaction,
        ConnectorSession session,
        ConnectorIndexHandle indexHandle,
        List<ColumnHandle> lookupSchema,
        List<ColumnHandle> outputSchema)
{
    TpchIndexHandle tpchIndexHandle = checkType(indexHandle, TpchIndexHandle.class, "indexHandle");

    Map<ColumnHandle, NullableValue> fixedValues = TupleDomain.extractFixedValues(tpchIndexHandle.getFixedValues()).get();
    checkArgument(lookupSchema.stream().noneMatch(handle -> fixedValues.keySet().contains(handle)),
            "Lookup columnHandles are not expected to overlap with the fixed value predicates");

    // Establish an order for the fixedValues
    List<ColumnHandle> fixedValueColumns = ImmutableList.copyOf(fixedValues.keySet());

    // Extract the fixedValues as their raw values and types
    List<Object> rawFixedValues = new ArrayList<>(fixedValueColumns.size());
    List<Type> rawFixedTypes = new ArrayList<>(fixedValueColumns.size());
    for (ColumnHandle fixedValueColumn : fixedValueColumns) {
        rawFixedValues.add(fixedValues.get(fixedValueColumn).getValue());
        rawFixedTypes.add(((TpchColumnHandle) fixedValueColumn).getType());
    }

    // Establish the schema after we append the fixed values to the lookup keys.
    List<ColumnHandle> finalLookupSchema = ImmutableList.<ColumnHandle>builder()
            .addAll(lookupSchema)
            .addAll(fixedValueColumns)
            .build();

    Optional<TpchIndexedData.IndexedTable> indexedTable = indexedData.getIndexedTable(tpchIndexHandle.getTableName(), tpchIndexHandle.getScaleFactor(), tpchIndexHandle.getIndexColumnNames());
    checkState(indexedTable.isPresent());
    TpchIndexedData.IndexedTable table = indexedTable.get();

    // Compute how to map from the final lookup schema to the table index key order
    List<Integer> keyRemap = computeRemap(handleToNames(finalLookupSchema), table.getKeyColumns());
    Function<RecordSet, RecordSet> keyFormatter = key -> new MappedRecordSet(new AppendingRecordSet(key, rawFixedValues, rawFixedTypes), keyRemap);

    // Compute how to map from the output of the indexed data to the expected output schema
    List<Integer> outputRemap = computeRemap(table.getOutputColumns(), handleToNames(outputSchema));
    Function<RecordSet, RecordSet> outputFormatter = output -> new MappedRecordSet(output, outputRemap);

    return new TpchConnectorIndex(keyFormatter, outputFormatter, table);
}