import com.facebook.presto.spi.RecordSet; //导入依赖的package包/类
@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);
}
 

import com.facebook.presto.spi.RecordSet; //导入依赖的package包/类
@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());
}
 

import com.facebook.presto.spi.RecordSet; //导入依赖的package包/类
@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());
}
 

import com.facebook.presto.spi.RecordSet; //导入依赖的package包/类
@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());
}
 

import com.facebook.presto.spi.RecordSet; //导入依赖的package包/类
@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());
}
 

import com.facebook.presto.spi.RecordSet; //导入依赖的package包/类
@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());
}
 

import com.facebook.presto.spi.RecordSet; //导入依赖的package包/类
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());
}
 

import com.facebook.presto.spi.RecordSet; //导入依赖的package包/类
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()));
        }
    };
}
 

import com.facebook.presto.spi.RecordSet; //导入依赖的package包/类
@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);
                }
            }
        }
    }
}
 

import com.facebook.presto.spi.RecordSet; //导入依赖的package包/类
@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());
}
 

import com.facebook.presto.spi.RecordSet; //导入依赖的package包/类
@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());
}
 

import com.facebook.presto.spi.RecordSet; //导入依赖的package包/类
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);
        }
    };
}
 

import com.facebook.presto.spi.RecordSet; //导入依赖的package包/类
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();
}
 

import com.facebook.presto.spi.RecordSet; //导入依赖的package包/类
@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;
}
 

import com.facebook.presto.spi.RecordSet; //导入依赖的package包/类
@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));
    }
}
 

import com.facebook.presto.spi.RecordSet; //导入依赖的package包/类
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);
}
 

import com.facebook.presto.spi.RecordSet; //导入依赖的package包/类
@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);
}
 

import com.facebook.presto.spi.RecordSet; //导入依赖的package包/类
@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);
}
 

import com.facebook.presto.spi.RecordSet; //导入依赖的package包/类
@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);
}
 

import com.facebook.presto.spi.RecordSet; //导入依赖的package包/类
@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());
}