import com.facebook.presto.spi.connector.ConnectorTransactionHandle; //导入依赖的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.connector.ConnectorTransactionHandle; //导入依赖的package包/类
@Override
public ConnectorPageSource createPageSource(ConnectorTransactionHandle transactionHandle, ConnectorSession session,
                                            ConnectorSplit split, List<ColumnHandle> columns)
{
    List<HDFSColumnHandle> hdfsColumns = columns.stream()
            .map(col -> (HDFSColumnHandle) col)
            .collect(Collectors.toList());
    HDFSSplit hdfsSplit = checkType(split, HDFSSplit.class, "hdfs split");
    Path path = new Path(hdfsSplit.getPath());

    Optional<ConnectorPageSource> pageSource = createHDFSPageSource(
            path,
            hdfsSplit.getStart(),
            hdfsSplit.getLen(),
            hdfsColumns);
    if (pageSource.isPresent()) {
        return pageSource.get();
    }
    throw new RuntimeException("Could not find a file reader for split " + hdfsSplit);
}
 

import com.facebook.presto.spi.connector.ConnectorTransactionHandle; //导入依赖的package包/类
@Override
public ConnectorSplitSource getSplits(ConnectorTransactionHandle handle, ConnectorSession session, ConnectorTableLayoutHandle layout)
{
    log.info("INFORMATION: AmpoolSplitManager getSplits() called.");

    AmpoolTableLayoutHandle layoutHandle = (AmpoolTableLayoutHandle) layout;
    AmpoolTableHandle tableHandle = layoutHandle.getTable();
    AmpoolTable table = new AmpoolTable(ampoolClient, tableHandle.getTableName());
    // this can happen if table is removed during a query
    checkState(table.getColumnsMetadata() != null, "Table %s.%s no longer exists", tableHandle.getSchemaName(), tableHandle.getTableName());

    List<ConnectorSplit> splits = new ArrayList<>();
    // TODO Pass here bucket id
    splits.add(new AmpoolSplit(connectorId, tableHandle.getSchemaName(), tableHandle.getTableName(),"" ,HostAddress.fromParts("localhost",0)));
    Collections.shuffle(splits);

    return new FixedSplitSource(splits);
}
 

import com.facebook.presto.spi.connector.ConnectorTransactionHandle; //导入依赖的package包/类
@Override
    public ConnectorSplitSource getSplits(ConnectorTransactionHandle transactionHandle, ConnectorSession session, ConnectorTableLayoutHandle layout)
    {
        KuduTableLayoutHandle layoutHandle = checkType(layout, KuduTableLayoutHandle.class, "layout");
        KuduTableHandle tableHandle = layoutHandle.getTable();
        KuduClient kuduClient = kuduClientManager.getClient();

        List<KuduScanToken> tokens = kuduClientManager.newScanTokenBuilder(kuduClient, tableHandle.getSchemaTableName().getTableName()).build();

        TupleDomain<KuduColumnHandle> effectivePredicate = layoutHandle.getConstraint()
                .transform(handle -> checkType(handle, KuduColumnHandle.class, "columnHandle"));

        ImmutableList.Builder<ConnectorSplit> builder = ImmutableList.builder();

        for (int i = 0; i < tokens.size(); i++) {
//            nodeManager.getWorkerNodes()
            List<HostAddress> hostAddresses = nodeManager.getWorkerNodes().stream()
                    .map(node -> node.getHostAndPort()).collect(Collectors.toList());
            ConnectorSplit split = new KuduSplit(hostAddresses, tableHandle.getSchemaTableName(), i, effectivePredicate);
            builder.add(split);
        }

        kuduClientManager.close(kuduClient);
        return new FixedSplitSource(builder.build());
    }
 

import com.facebook.presto.spi.connector.ConnectorTransactionHandle; //导入依赖的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.connector.ConnectorTransactionHandle; //导入依赖的package包/类
@Override
public ConnectorPageSource createPageSource(ConnectorTransactionHandle transactionHandle, ConnectorSession session, ConnectorSplit split, List<ColumnHandle> columns)
{
    RaptorSplit raptorSplit = checkType(split, RaptorSplit.class, "split");

    UUID shardUuid = raptorSplit.getShardUuid();
    List<RaptorColumnHandle> columnHandles = columns.stream().map(toRaptorColumnHandle()).collect(toList());
    List<Long> columnIds = columnHandles.stream().map(RaptorColumnHandle::getColumnId).collect(toList());
    List<Type> columnTypes = columnHandles.stream().map(RaptorColumnHandle::getColumnType).collect(toList());

    return storageManager.getPageSource(
            shardUuid,
            columnIds,
            columnTypes,
            raptorSplit.getEffectivePredicate(),
            ReaderAttributes.from(session),
            raptorSplit.getTransactionId());
}
 

import com.facebook.presto.spi.connector.ConnectorTransactionHandle; //导入依赖的package包/类
@Override
public ConnectorPageSink createPageSink(ConnectorTransactionHandle transactionHandle, ConnectorSession session, ConnectorInsertTableHandle tableHandle)
{
    RaptorInsertTableHandle handle = checkType(tableHandle, RaptorInsertTableHandle.class, "tableHandle");
    return new RaptorPageSink(
            pageSorter,
            storageManager,
            shardInfoCodec,
            handle.getTransactionId(),
            toColumnIds(handle.getColumnHandles()),
            handle.getColumnTypes(),
            Optional.empty(),
            toColumnIds(handle.getSortColumnHandles()),
            handle.getSortOrders(),
            maxBufferSize);
}
 

import com.facebook.presto.spi.connector.ConnectorTransactionHandle; //导入依赖的package包/类
@Override
public List<TableLayoutResult> getLayouts(Session session, TableHandle table, Constraint<ColumnHandle> constraint, Optional<Set<ColumnHandle>> desiredColumns)
{
    if (constraint.getSummary().isNone()) {
        return ImmutableList.of();
    }

    TupleDomain<ColumnHandle> summary = constraint.getSummary();
    String connectorId = table.getConnectorId();
    ConnectorTableHandle connectorTable = table.getConnectorHandle();
    Predicate<Map<ColumnHandle, NullableValue>> predicate = constraint.predicate();

    ConnectorEntry entry = getConnectorMetadata(connectorId);
    ConnectorMetadata metadata = entry.getMetadata(session);
    ConnectorTransactionHandle transaction = entry.getTransactionHandle(session);
    ConnectorSession connectorSession = session.toConnectorSession(entry.getCatalog());
    List<ConnectorTableLayoutResult> layouts = metadata.getTableLayouts(connectorSession, connectorTable, new Constraint<>(summary, predicate::test), desiredColumns);

    return layouts.stream()
            .map(layout -> new TableLayoutResult(fromConnectorLayout(connectorId, transaction, layout.getTableLayout()), layout.getUnenforcedConstraint()))
            .collect(toImmutableList());
}
 

import com.facebook.presto.spi.connector.ConnectorTransactionHandle; //导入依赖的package包/类
@Override
public ConnectorSplitSource getSplits(ConnectorTransactionHandle transaction, ConnectorSession session, ConnectorTableLayoutHandle layout)
{
    JmxTableLayoutHandle jmxLayout = checkType(layout, JmxTableLayoutHandle.class, "layout");
    JmxTableHandle tableHandle = jmxLayout.getTable();
    TupleDomain<ColumnHandle> predicate = jmxLayout.getConstraint();

    //TODO is there a better way to get the node column?
    JmxColumnHandle nodeColumnHandle = tableHandle.getColumns().get(0);

    List<ConnectorSplit> splits = nodeManager.getNodes(ACTIVE)
            .stream()
            .filter(node -> {
                NullableValue value = NullableValue.of(VARCHAR, utf8Slice(node.getNodeIdentifier()));
                return predicate.overlaps(fromFixedValues(ImmutableMap.of(nodeColumnHandle, value)));
            })
            .map(node -> new JmxSplit(tableHandle, ImmutableList.of(node.getHostAndPort())))
            .collect(toList());

    return new FixedSplitSource(connectorId, splits);
}
 

import com.facebook.presto.spi.connector.ConnectorTransactionHandle; //导入依赖的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.connector.ConnectorTransactionHandle; //导入依赖的package包/类
@Override
public ConnectorSplitSource getSplits(ConnectorTransactionHandle transaction, ConnectorSession session, ConnectorTableLayoutHandle layout)
{
    CassandraTableLayoutHandle layoutHandle = checkType(layout, CassandraTableLayoutHandle.class, "layout");
    CassandraTableHandle cassandraTableHandle = layoutHandle.getTable();
    List<CassandraPartition> partitions = layoutHandle.getPartitions().get();

    requireNonNull(partitions, "partitions is null");
    if (partitions.isEmpty()) {
        return new FixedSplitSource(connectorId, ImmutableList.<ConnectorSplit>of());
    }

    // if this is an unpartitioned table, split into equal ranges
    if (partitions.size() == 1) {
        CassandraPartition cassandraPartition = partitions.get(0);
        if (cassandraPartition.isUnpartitioned() || cassandraPartition.isIndexedColumnPredicatePushdown()) {
            CassandraTable table = schemaProvider.getTable(cassandraTableHandle);
            List<ConnectorSplit> splits = getSplitsByTokenRange(table, cassandraPartition.getPartitionId());
            return new FixedSplitSource(connectorId, splits);
        }
    }

    return new FixedSplitSource(connectorId, getSplitsForPartitions(cassandraTableHandle, partitions));
}
 

import com.facebook.presto.spi.connector.ConnectorTransactionHandle; //导入依赖的package包/类
@Override
public ConnectorSplitSource getSplits(ConnectorTransactionHandle transaction, ConnectorSession session, ConnectorTableLayoutHandle layout)
{
    TpchTableHandle tableHandle = checkType(layout, TpchTableLayoutHandle.class, "layout").getTable();

    Set<Node> nodes = nodeManager.getActiveDatasourceNodes(connectorId);
    checkState(!nodes.isEmpty(), "No TPCH nodes available");

    int totalParts = nodes.size() * splitsPerNode;
    int partNumber = 0;

    // Split the data using split and skew by the number of nodes available.
    ImmutableList.Builder<ConnectorSplit> splits = ImmutableList.builder();
    for (Node node : nodes) {
        for (int i = 0; i < splitsPerNode; i++) {
            splits.add(new TpchSplit(tableHandle, partNumber, totalParts, ImmutableList.of(node.getHostAndPort())));
            partNumber++;
        }
    }
    return new FixedSplitSource(connectorId, splits.build());
}
 

import com.facebook.presto.spi.connector.ConnectorTransactionHandle; //导入依赖的package包/类
@Override
public ConnectorSplitSource getSplits(ConnectorTransactionHandle transactionHandle, ConnectorSession session, ConnectorTableLayoutHandle layout)
{
    KinesisTableLayoutHandle kinesislayout = handleResolver.convertLayout(layout);
    KinesisTableHandle kinesisTableHandle = kinesislayout.getTable();

    InternalStreamDescription desc = this.getStreamDescription(kinesisTableHandle.getStreamName());

    ImmutableList.Builder<ConnectorSplit> builder = ImmutableList.builder();
    for (Shard shard : desc.getShards()) {
        KinesisSplit split = new KinesisSplit(connectorId,
                kinesisTableHandle.getStreamName(),
                kinesisTableHandle.getMessageDataFormat(),
                shard.getShardId(),
                shard.getSequenceNumberRange().getStartingSequenceNumber(),
                shard.getSequenceNumberRange().getEndingSequenceNumber());
        builder.add(split);
    }

    return new FixedSplitSource(builder.build());
}
 

import com.facebook.presto.spi.connector.ConnectorTransactionHandle; //导入依赖的package包/类
@Override
public ConnectorSplitSource getSplits(ConnectorTransactionHandle transaction, ConnectorSession session, ConnectorTableLayoutHandle layout)
{
    InformationSchemaTableLayoutHandle handle = checkType(layout, InformationSchemaTableLayoutHandle.class, "layout");
    Map<ColumnHandle, NullableValue> bindings = extractFixedValues(handle.getConstraint()).orElse(ImmutableMap.of());

    List<HostAddress> localAddress = ImmutableList.of(nodeManager.getCurrentNode().getHostAndPort());

    Map<String, NullableValue> filters = bindings.entrySet().stream().collect(toMap(
            entry -> checkType(entry.getKey(), InformationSchemaColumnHandle.class, "column").getColumnName(),
            Entry::getValue));

    ConnectorSplit split = new InformationSchemaSplit(handle.getTable(), filters, localAddress);

    return new FixedSplitSource(null, ImmutableList.of(split));
}
 

import com.facebook.presto.spi.connector.ConnectorTransactionHandle; //导入依赖的package包/类
@Override
public ConnectorPageSource createPageSource(ConnectorTransactionHandle transactionHandle, ConnectorSession session, ConnectorSplit split, List<ColumnHandle> columns)
{
    InternalTable table = getInternalTable(transactionHandle, session, split, columns);

    List<Integer> channels = new ArrayList<>();
    for (ColumnHandle column : columns) {
        String columnName = checkType(column, InformationSchemaColumnHandle.class, "column").getColumnName();
        int columnIndex = table.getColumnIndex(columnName);
        channels.add(columnIndex);
    }

    ImmutableList.Builder<Page> pages = ImmutableList.builder();
    for (Page page : table.getPages()) {
        Block[] blocks = new Block[channels.size()];
        for (int index = 0; index < blocks.length; index++) {
            blocks[index] = page.getBlock(channels.get(index));
        }
        pages.add(new Page(page.getPositionCount(), blocks));
    }
    return new FixedPageSource(pages.build());
}
 

import com.facebook.presto.spi.connector.ConnectorTransactionHandle; //导入依赖的package包/类
private InternalTable getInternalTable(ConnectorTransactionHandle transactionHandle, ConnectorSession connectorSession, ConnectorSplit connectorSplit, List<ColumnHandle> columns)
{
    InformationSchemaTransactionHandle transaction = checkType(transactionHandle, InformationSchemaTransactionHandle.class, "transaction");
    InformationSchemaSplit split = checkType(connectorSplit, InformationSchemaSplit.class, "split");

    requireNonNull(columns, "columns is null");

    InformationSchemaTableHandle handle = split.getTableHandle();
    Map<String, NullableValue> filters = split.getFilters();

    Session session = Session.builder(metadata.getSessionPropertyManager())
            .setTransactionId(transaction.getTransactionId())
            .setQueryId(new QueryId(connectorSession.getQueryId()))
            .setIdentity(connectorSession.getIdentity())
            .setSource("information_schema")
            .setCatalog("") // default catalog is not be used
            .setSchema("") // default schema is not be used
            .setTimeZoneKey(connectorSession.getTimeZoneKey())
            .setLocale(connectorSession.getLocale())
            .setStartTime(connectorSession.getStartTime())
            .build();

    return getInformationSchemaTable(session, handle.getCatalogName(), handle.getSchemaTableName(), filters);
}
 

import com.facebook.presto.spi.connector.ConnectorTransactionHandle; //导入依赖的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.connector.ConnectorTransactionHandle; //导入依赖的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.connector.ConnectorTransactionHandle; //导入依赖的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.connector.ConnectorTransactionHandle; //导入依赖的package包/类
@Override
public ConnectorTransactionHandle beginTransaction(IsolationLevel isolationLevel, boolean readOnly)
{
    checkConnectorSupports(READ_UNCOMMITTED, isolationLevel);
    HDFSTransactionHandle transaction = new HDFSTransactionHandle();
    transactions.putIfAbsent(transaction, hdfsMetadataFactory.create());
    return transaction;
}