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

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

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

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

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

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

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

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

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

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

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

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