import com.facebook.presto.spi.HostAddress; //导入依赖的package包/类
@JsonCreator
public HDFSSplit(
        @JsonProperty("connectorId") HDFSConnectorId connectorId,
        @JsonProperty("table") SchemaTableName table,
        @JsonProperty("path") String path,
        @JsonProperty("start") long start,
        @JsonProperty("len") long len,
        @JsonProperty("addresses") List<HostAddress> addresses
        )
{
    this.connectorId = requireNonNull(connectorId, "connectorId is null");
    this.table = requireNonNull(table, "table is null");
    this.path = requireNonNull(path, "path is null");
    this.start = requireNonNull(start);
    this.len = requireNonNull(len);
    this.addresses = ImmutableList.copyOf(requireNonNull(addresses, "addresses is null"));
}
 

import com.facebook.presto.spi.HostAddress; //导入依赖的package包/类
public List<HostAddress> getBlockLocations(Path file, long start, long len)
{
    Set<HostAddress> addresses = new HashSet<>();
    if (!getFS().isPresent()) {
        throw new FileSystemNotFoundException("");
    }
    BlockLocation[] locations = new BlockLocation[0];
    try {
        locations = getFS().get().getFileBlockLocations(file, start, len);
    }
    catch (IOException e) {
        log.error(e);
    }
    assert locations.length <= 1;
    for (BlockLocation location : locations) {
        try {
            addresses.addAll(toHostAddress(location.getHosts()));
        }
        catch (IOException e) {
            log.error(e);
        }
    }
    return new ArrayList<>(addresses);
}
 

import com.facebook.presto.spi.HostAddress; //导入依赖的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.HostAddress; //导入依赖的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.HostAddress; //导入依赖的package包/类
@JsonCreator
public KafkaSplit(
        @JsonProperty("connectorId") String connectorId,
        @JsonProperty("topicName") String topicName,
        @JsonProperty("keyDataFormat") String keyDataFormat,
        @JsonProperty("messageDataFormat") String messageDataFormat,
        @JsonProperty("partitionId") int partitionId,
        @JsonProperty("start") long start,
        @JsonProperty("end") long end,
        @JsonProperty("nodes") List<HostAddress> nodes)
{
    this.connectorId = requireNonNull(connectorId, "connector id is null");
    this.topicName = requireNonNull(topicName, "topicName is null");
    this.keyDataFormat = requireNonNull(keyDataFormat, "dataFormat is null");
    this.messageDataFormat = requireNonNull(messageDataFormat, "messageDataFormat is null");
    this.partitionId = partitionId;
    this.start = start;
    this.end = end;
    this.nodes = ImmutableList.copyOf(requireNonNull(nodes, "addresses is null"));
}
 

import com.facebook.presto.spi.HostAddress; //导入依赖的package包/类
@JsonCreator
    public ExampleSplit(
            @JsonProperty("connectorId") String connectorId,
            @JsonProperty("schemaName") String schemaName,
            @JsonProperty("tableName") String tableName,
            @JsonProperty("uri") URI uri)
    {
        this.schemaName = requireNonNull(schemaName, "schema name is null");
        this.connectorId = requireNonNull(connectorId, "connector id is null");
        this.tableName = requireNonNull(tableName, "table name is null");
        this.uri = requireNonNull(uri, "uri is null");

//        if ("http".equalsIgnoreCase(uri.getScheme()) || "https".equalsIgnoreCase(uri.getScheme())) {
        remotelyAccessible = true;
        addresses = ImmutableList.of(HostAddress.fromUri(uri));
    }
 

import com.facebook.presto.spi.HostAddress; //导入依赖的package包/类
@Test
public void testAddresses()
{
    // http split with default port
    ExampleSplit httpSplit = new ExampleSplit("connectorId", "schemaName", "tableName", URI.create("http://example.com/example"));
    assertEquals(httpSplit.getAddresses(), ImmutableList.of(HostAddress.fromString("example.com")));
    assertEquals(httpSplit.isRemotelyAccessible(), true);

    // http split with custom port
    httpSplit = new ExampleSplit("connectorId", "schemaName", "tableName", URI.create("http://example.com:8080/example"));
    assertEquals(httpSplit.getAddresses(), ImmutableList.of(HostAddress.fromParts("example.com", 8080)));
    assertEquals(httpSplit.isRemotelyAccessible(), true);

    // http split with default port
    ExampleSplit httpsSplit = new ExampleSplit("connectorId", "schemaName", "tableName", URI.create("https://example.com/example"));
    assertEquals(httpsSplit.getAddresses(), ImmutableList.of(HostAddress.fromString("example.com")));
    assertEquals(httpsSplit.isRemotelyAccessible(), true);

    // http split with custom port
    httpsSplit = new ExampleSplit("connectorId", "schemaName", "tableName", URI.create("https://example.com:8443/example"));
    assertEquals(httpsSplit.getAddresses(), ImmutableList.of(HostAddress.fromParts("example.com", 8443)));
    assertEquals(httpsSplit.isRemotelyAccessible(), true);
}
 

import com.facebook.presto.spi.HostAddress; //导入依赖的package包/类
private ConnectorSplit createSplit(ShardNodes shard)
{
    UUID shardId = shard.getShardUuid();
    Collection<String> nodeIds = shard.getNodeIdentifiers();

    List<HostAddress> addresses = getAddressesForNodes(nodesById, nodeIds);

    if (addresses.isEmpty()) {
        if (!backupAvailable) {
            throw new PrestoException(RAPTOR_NO_HOST_FOR_SHARD, format("No host for shard %s found: %s", shardId, nodeIds));
        }

        // Pick a random node and optimistically assign the shard to it.
        // That node will restore the shard from the backup location.
        Set<Node> availableNodes = nodeSupplier.getWorkerNodes();
        if (availableNodes.isEmpty()) {
            throw new PrestoException(NO_NODES_AVAILABLE, "No nodes available to run query");
        }
        Node node = selectRandom(availableNodes);
        shardManager.assignShard(tableId, shardId, node.getNodeIdentifier());
        addresses = ImmutableList.of(node.getHostAndPort());
    }

    return new RaptorSplit(connectorId, shardId, addresses, effectivePredicate, transactionId);
}
 

import com.facebook.presto.spi.HostAddress; //导入依赖的package包/类
@JsonCreator
public RedisSplit(
        @JsonProperty("connectorId") String connectorId,
        @JsonProperty("schemaName") String schemaName,
        @JsonProperty("tableName") String tableName,
        @JsonProperty("keyDataFormat") String keyDataFormat,
        @JsonProperty("valueDataFormat") String valueDataFormat,
        @JsonProperty("keyName") String keyName,
        @JsonProperty("start") long start,
        @JsonProperty("end") long end,
        @JsonProperty("nodes") List<HostAddress> nodes)
{
    this.connectorId = requireNonNull(connectorId, "connector id is null");
    this.schemaName = requireNonNull(schemaName, "schemaName is null");
    this.tableName = requireNonNull(tableName, "dataFormat is null");
    this.keyDataFormat = requireNonNull(keyDataFormat, "KeydataFormat is null");
    this.valueDataFormat = requireNonNull(valueDataFormat, "valueDataFormat is null");
    this.keyName = keyName;
    this.nodes = ImmutableList.copyOf(requireNonNull(nodes, "addresses is null"));
    this.start = start;
    this.end = end;
    this.valueDataType = toRedisDataType(valueDataFormat);
    this.keyDataType = toRedisDataType(keyDataFormat);
}
 

import com.facebook.presto.spi.HostAddress; //导入依赖的package包/类
@Test
public void testNoPredicate()
        throws Exception
{
    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());

    Set<String> actualNodes = nodes.stream().map(Node::getNodeIdentifier).collect(toSet());
    Set<String> expectedNodes = new HashSet<>();
    for (ConnectorSplit split : allSplits) {
        List<HostAddress> addresses = split.getAddresses();
        assertEquals(addresses.size(), 1);
        expectedNodes.add(addresses.get(0).getHostText());
    }
    assertEquals(actualNodes, expectedNodes);
}
 

import com.facebook.presto.spi.HostAddress; //导入依赖的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.HostAddress; //导入依赖的package包/类
public NetworkLocationCache(NetworkTopology networkTopology)
{
    this.networkTopology = requireNonNull(networkTopology, "networkTopology is null");

    this.cache = CacheBuilder.newBuilder()
            .expireAfterWrite(1, DAYS)
            .refreshAfterWrite(12, HOURS)
            .build(asyncReloading(new CacheLoader<HostAddress, NetworkLocation>()
            {
                @Override
                public NetworkLocation load(HostAddress host)
                        throws Exception
                {
                    return locate(host);
                }
            }, executor));

    this.negativeCache = CacheBuilder.newBuilder()
            .expireAfterWrite(NEGATIVE_CACHE_DURATION.toMillis(), MILLISECONDS)
            .build();
}
 

import com.facebook.presto.spi.HostAddress; //导入依赖的package包/类
@Test
public void testSerialization()
        throws Exception
{
    String connectorId = "testid";
    SystemTableHandle tableHandle = new SystemTableHandle(connectorId, "xyz", "foo");
    SystemSplit expected = new SystemSplit(connectorId, tableHandle, HostAddress.fromParts("127.0.0.1", 0), TupleDomain.all());

    JsonCodec<SystemSplit> codec = jsonCodec(SystemSplit.class);
    SystemSplit actual = codec.fromJson(codec.toJson(expected));

    assertEquals(actual.getConnectorId(), expected.getConnectorId());
    assertEquals(actual.getTableHandle(), expected.getTableHandle());
    assertEquals(actual.getAddresses(), expected.getAddresses());
    assertEquals(actual.getConstraint(), expected.getConstraint());
}
 

import com.facebook.presto.spi.HostAddress; //导入依赖的package包/类
@JsonCreator
public CassandraSplit(
        @JsonProperty("connectorId") String connectorId,
        @JsonProperty("schema") String schema,
        @JsonProperty("table") String table,
        @JsonProperty("partitionId") String partitionId,
        @JsonProperty("splitCondition") String splitCondition,
        @JsonProperty("addresses") List<HostAddress> addresses)
{
    requireNonNull(connectorId, "connectorId is null");
    requireNonNull(schema, "schema is null");
    requireNonNull(table, "table is null");
    requireNonNull(partitionId, "partitionName is null");
    requireNonNull(addresses, "addresses is null");

    this.connectorId = connectorId;
    this.schema = schema;
    this.table = table;
    this.partitionId = partitionId;
    this.addresses = ImmutableList.copyOf(addresses);
    this.splitCondition = splitCondition;
}
 

import com.facebook.presto.spi.HostAddress; //导入依赖的package包/类
private List<ConnectorSplit> getSplitsByTokenRange(CassandraTable table, String partitionId)
{
    String schema = table.getTableHandle().getSchemaName();
    String tableName = table.getTableHandle().getTableName();
    String tokenExpression = table.getTokenExpression();

    ImmutableList.Builder<ConnectorSplit> builder = ImmutableList.builder();
    List<CassandraTokenSplitManager.TokenSplit> tokenSplits;
    try {
        tokenSplits = tokenSplitMgr.getSplits(schema, tableName);
    }
    catch (IOException e) {
        throw new RuntimeException(e);
    }
    for (CassandraTokenSplitManager.TokenSplit tokenSplit : tokenSplits) {
        String condition = buildTokenCondition(tokenExpression, tokenSplit.getStartToken(), tokenSplit.getEndToken());
        List<HostAddress> addresses = new HostAddressFactory().AddressNamesToHostAddressList(tokenSplit.getHosts());
        CassandraSplit split = new CassandraSplit(connectorId, schema, tableName, partitionId, condition, addresses);
        builder.add(split);
    }

    return builder.build();
}
 

import com.facebook.presto.spi.HostAddress; //导入依赖的package包/类
@Test
public void testToHostAddressList()
        throws Exception
{
    Set<Host> hosts = ImmutableSet.<Host>of(
            new TestHost(
                new InetSocketAddress(
                    InetAddress.getByAddress(new byte[] {
                        1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16
                    }),
                    3000)),
            new TestHost(new InetSocketAddress(InetAddress.getByAddress(new byte[] {1, 2, 3, 4}), 3000)));

    HostAddressFactory hostAddressFactory = new HostAddressFactory();
    List<HostAddress> list = hostAddressFactory.toHostAddressList(hosts);

    assertEquals(list.toString(), "[[102:304:506:708:90a:b0c:d0e:f10], 1.2.3.4]");
}
 

import com.facebook.presto.spi.HostAddress; //导入依赖的package包/类
private List<HostAddress> toHostAddress(String[] hosts)
{
    ImmutableList.Builder<HostAddress> builder = ImmutableList.builder();
    for (String host : hosts) {
        builder.add(HostAddress.fromString(host));
    }
    return builder.build();
}
 

import com.facebook.presto.spi.HostAddress; //导入依赖的package包/类
@JsonCreator
public RestConnectorSplit(
        @JsonProperty("tableHandle") RestTableHandle tableHandle,
        @JsonProperty("addresses") List<HostAddress> addresses)
{
    this.tableHandle = tableHandle;
    this.addresses = addresses;
}
 

import com.facebook.presto.spi.HostAddress; //导入依赖的package包/类
@Override
public ConnectorSplitSource getSplits(ConnectorTransactionHandle transactionHandle, ConnectorSession session, ConnectorTableLayoutHandle layout)
{
    RestConnectorTableLayoutHandle layoutHandle = Types.checkType(layout, RestConnectorTableLayoutHandle.class, "layout");

    List<HostAddress> addresses = nodeManager.getRequiredWorkerNodes().stream()
            .map(Node::getHostAndPort)
            .collect(toList());

    return new FixedSplitSource(ImmutableList.of(
            new RestConnectorSplit(layoutHandle.getTableHandle(), addresses)));
}
 

import com.facebook.presto.spi.HostAddress; //导入依赖的package包/类
@JsonCreator
public AmpoolSplit(@JsonProperty("connectorId") String connectorId,
                   @JsonProperty("schemaName") String schemaName,
                   @JsonProperty("tableName") String tableName,
                   @JsonProperty("bucketId") String bucketId,
                   @JsonProperty("address") HostAddress address)
{
    this.schemaName = requireNonNull(schemaName, "schema name is null");
    this.connectorId = requireNonNull(connectorId, "connector id is null");
    this.tableName = requireNonNull(tableName, "table name is null");
    this.bucketId = requireNonNull(bucketId, "bucket id is null");
    this.address = requireNonNull(address, "address is null");
    log.info("INFORMATION: AmpoolSplit created successfully.");
}