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

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

@JsonCreator
public HDFSTableLayoutHandle(
        @JsonProperty("table") HDFSTableHandle table,
        @JsonProperty("fiberColumn") HDFSColumnHandle fiberColumn,
        @JsonProperty("timestampColumn") HDFSColumnHandle timestampColumn,
        @JsonProperty("fiberFunction") Function fiberFunction,
        @JsonProperty("storageFormat") StorageFormat storageFormat,
        @JsonProperty("predicates") Optional<TupleDomain<ColumnHandle>> predicates)
{
    this.table = requireNonNull(table, "table is null");
    this.fiberColumn = requireNonNull(fiberColumn, "fiberColumn is null");
    this.timestampColumn = requireNonNull(timestampColumn, "timestampColumn is null");
    this.fiberFunction = requireNonNull(fiberFunction, "fiberFunc is null");
    this.storageFormat = requireNonNull(storageFormat, "storageFormat is null");
    this.predicates = requireNonNull(predicates, "predicates is null");
}

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

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

@Override
public Map<String, ColumnHandle> getColumnHandles(ConnectorSession session, ConnectorTableHandle tableHandle)
{
    ExampleTableHandle exampleTableHandle = checkType(tableHandle, ExampleTableHandle.class, "tableHandle");
    checkArgument(exampleTableHandle.getConnectorId().equals(connectorId), "tableHandle is not for this connector");

    ExampleTable table = exampleClient.getTable(exampleTableHandle.getSchemaName(), exampleTableHandle.getTableName());
    if (table == null) {
        throw new TableNotFoundException(exampleTableHandle.toSchemaTableName());
    }

    ImmutableMap.Builder<String, ColumnHandle> columnHandles = ImmutableMap.builder();
    int index = 0;
    for (ColumnMetadata column : table.getColumnsMetadata()) {
        columnHandles.put(column.getName(), new ExampleColumnHandle(connectorId, column.getName(), column.getType(), index));
        index++;
    }
    return columnHandles.build();
}

@Override
public PlanNode visitIndexSource(IndexSourceNode node, RewriteContext<Set<Symbol>> context)
{
    List<Symbol> newOutputSymbols = node.getOutputSymbols().stream()
            .filter(context.get()::contains)
            .collect(toImmutableList());

    Set<Symbol> newLookupSymbols = node.getLookupSymbols().stream()
            .filter(context.get()::contains)
            .collect(toImmutableSet());

    Set<Symbol> requiredAssignmentSymbols = context.get();
    if (!node.getEffectiveTupleDomain().isNone()) {
        Set<Symbol> requiredSymbols = Maps.filterValues(node.getAssignments(), in(node.getEffectiveTupleDomain().getDomains().get().keySet())).keySet();
        requiredAssignmentSymbols = Sets.union(context.get(), requiredSymbols);
    }
    Map<Symbol, ColumnHandle> newAssignments = Maps.filterKeys(node.getAssignments(), in(requiredAssignmentSymbols));

    return new IndexSourceNode(node.getId(), node.getIndexHandle(), node.getTableHandle(), newLookupSymbols, newOutputSymbols, newAssignments, node.getEffectiveTupleDomain());
}

@Test
public void testTypesRcBinaryRecordCursor()
        throws Exception
{
    ConnectorSession session = newSession();

    if (metadata.getTableHandle(session, new SchemaTableName(database, "presto_test_types_rcbinary")) == null) {
        return;
    }

    ConnectorTableHandle tableHandle = getTableHandle(new SchemaTableName(database, "presto_test_types_rcbinary"));
    ConnectorTableMetadata tableMetadata = metadata.getTableMetadata(session, tableHandle);
    HiveSplit hiveSplit = getHiveSplit(tableHandle);
    List<ColumnHandle> columnHandles = ImmutableList.copyOf(metadata.getColumnHandles(session, tableHandle).values());

    ConnectorPageSourceProvider pageSourceProvider = new HivePageSourceProvider(
            new HiveClientConfig().setTimeZone(timeZone.getID()),
            hdfsEnvironment,
            ImmutableSet.<HiveRecordCursorProvider>of(new ColumnarBinaryHiveRecordCursorProvider()),
            ImmutableSet.<HivePageSourceFactory>of(),
            TYPE_MANAGER);

    ConnectorPageSource pageSource = pageSourceProvider.createPageSource(session, hiveSplit, columnHandles);
    assertGetRecords(RCBINARY, tableMetadata, hiveSplit, pageSource, columnHandles);
}

@Test
public void testRenameColumn()
        throws Exception
{
    assertNull(metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS));
    metadata.createTable(SESSION, getOrdersTable());
    ConnectorTableHandle tableHandle = metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS);
    assertInstanceOf(tableHandle, RaptorTableHandle.class);

    RaptorTableHandle raptorTableHandle = (RaptorTableHandle) tableHandle;
    ColumnHandle columnHandle = metadata.getColumnHandles(SESSION, tableHandle).get("orderkey");

    metadata.renameColumn(SESSION, raptorTableHandle, columnHandle, "orderkey_renamed");

    assertNull(metadata.getColumnHandles(SESSION, tableHandle).get("orderkey"));
    assertNotNull(metadata.getColumnHandles(SESSION, tableHandle).get("orderkey_renamed"));
}

@Test
public void testGetPartitionSplitsTableOfflinePartition()
        throws Exception
{
    ConnectorSession session = newSession();

    ConnectorTableHandle tableHandle = getTableHandle(tableOfflinePartition);
    assertNotNull(tableHandle);

    ColumnHandle dsColumn = metadata.getColumnHandles(session, tableHandle).get("ds");
    assertNotNull(dsColumn);

    Domain domain = Domain.singleValue(VARCHAR, utf8Slice("2012-12-30"));
    TupleDomain<ColumnHandle> tupleDomain = TupleDomain.withColumnDomains(ImmutableMap.of(dsColumn, domain));
    List<ConnectorTableLayoutResult> tableLayoutResults = metadata.getTableLayouts(session, tableHandle, new Constraint<>(tupleDomain, bindings -> true), Optional.empty());
    try {
        getSplitCount(splitManager.getSplits(session, getOnlyElement(tableLayoutResults).getTableLayout().getHandle()));
        fail("Expected PartitionOfflineException");
    }
    catch (PartitionOfflineException e) {
        assertEquals(e.getTableName(), tableOfflinePartition);
        assertEquals(e.getPartition(), "ds=2012-12-30");
    }
}

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 Map<String, ColumnHandle> getColumnHandles(ConnectorSession session, ConnectorTableHandle tableHandle)
{
    SchemaTableName tableName = schemaTableName(tableHandle);
    Optional<Table> table = metastore.getTable(tableName.getSchemaName(), tableName.getTableName());
    if (!table.isPresent()) {
        throw new TableNotFoundException(tableName);
    }
    ImmutableMap.Builder<String, ColumnHandle> columnHandles = ImmutableMap.builder();
    for (HiveColumnHandle columnHandle : hiveColumnHandles(connectorId, table.get())) {
        if (!columnHandle.getName().equals(SAMPLE_WEIGHT_COLUMN_NAME)) {
            columnHandles.put(columnHandle.getName(), columnHandle);
        }
    }
    return columnHandles.build();
}

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

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

@Test
public void testTypesRcTextRecordCursor()
        throws Exception
{
    ConnectorSession session = newSession();

    if (metadata.getTableHandle(session, new SchemaTableName(database, "presto_test_types_rctext")) == null) {
        return;
    }

    ConnectorTableHandle tableHandle = getTableHandle(new SchemaTableName(database, "presto_test_types_rctext"));
    ConnectorTableMetadata tableMetadata = metadata.getTableMetadata(session, tableHandle);
    HiveSplit hiveSplit = getHiveSplit(tableHandle);
    List<ColumnHandle> columnHandles = ImmutableList.copyOf(metadata.getColumnHandles(session, tableHandle).values());

    ConnectorPageSourceProvider pageSourceProvider = new HivePageSourceProvider(
            new HiveClientConfig().setTimeZone(timeZone.getID()),
            hdfsEnvironment,
            ImmutableSet.<HiveRecordCursorProvider>of(new ColumnarTextHiveRecordCursorProvider()),
            ImmutableSet.<HivePageSourceFactory>of(),
            TYPE_MANAGER);

    ConnectorPageSource pageSource = pageSourceProvider.createPageSource(session, hiveSplit, columnHandles);
    assertGetRecords(RCTEXT, tableMetadata, hiveSplit, pageSource, columnHandles);
}

@Override
public Void visitIndexSource(IndexSourceNode node, Void context)
{
    TableHandle tableHandle = node.getTableHandle();
    Optional<ColumnHandle> sampleWeightColumn = metadata.getSampleWeightColumnHandle(session, tableHandle);

    Set<Column> columns = new HashSet<>();
    for (ColumnHandle columnHandle : node.getAssignments().values()) {
        if (!columnHandle.equals(sampleWeightColumn.orElse(null))) {
            columns.add(createColumnEntry(metadata.getColumnMetadata(session, tableHandle, columnHandle)));
        }
    }

    inputs.put(createTableEntry(metadata.getTableMetadata(session, tableHandle)), columns);

    return null;
}

@Override
public PlanNode visitTableScan(TableScanNode node, RewriteContext<Set<Symbol>> context)
{
    Set<Symbol> requiredTableScanOutputs = context.get().stream()
            .filter(node.getOutputSymbols()::contains)
            .collect(toImmutableSet());

    List<Symbol> newOutputSymbols = node.getOutputSymbols().stream()
            .filter(requiredTableScanOutputs::contains)
            .collect(toImmutableList());

    Map<Symbol, ColumnHandle> newAssignments = Maps.filterKeys(node.getAssignments(), in(requiredTableScanOutputs));

    return new TableScanNode(
            node.getId(),
            node.getTable(),
            newOutputSymbols,
            newAssignments,
            node.getLayout(),
            node.getCurrentConstraint(),
            node.getOriginalConstraint());
}

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

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

/**
 * Return a list of table layouts that satisfy the given constraint.
 * <p>
 * For each layout, connectors must return an "unenforced constraint" representing the part of the constraint summary that isn't guaranteed by the layout.
 *
 * @param session session
 * @param table table
 * @param constraint constraint
 * @param desiredColumns desired columns
 */
@Override
public List<ConnectorTableLayoutResult> getTableLayouts(ConnectorSession session, ConnectorTableHandle table, Constraint<ColumnHandle> constraint, Optional<Set<ColumnHandle>> desiredColumns)
{
    // get table name from ConnectorTableHandle
    HDFSTableHandle hdfsTable = checkType(table, HDFSTableHandle.class, "table");
    SchemaTableName tableName = hdfsTable.getSchemaTableName();
    // create HDFSTableLayoutHandle
    HDFSTableLayoutHandle tableLayout = metaDataQuery.getTableLayout(connectorId, tableName.getSchemaName(), tableName.getTableName()).orElse(null);
    tableLayout.setPredicates(constraint.getSummary() != null ? Optional.of(constraint.getSummary()) : Optional.empty());
    // ConnectorTableLayout layout = new ConnectorTableLayout(HDFSTableLayoutHandle)
    ConnectorTableLayout layout = getTableLayout(session, tableLayout);

    return ImmutableList.of(new ConnectorTableLayoutResult(layout, constraint.getSummary()));
}

/**
 * Gets all of the columns on the specified table, or an empty map if the columns can not be enumerated.
 *
 * @param session session
 * @param tableHandle table handle
 * @throws RuntimeException if table handle is no longer valid
 */
@Override
public Map<String, ColumnHandle> getColumnHandles(ConnectorSession session, ConnectorTableHandle tableHandle)
{
    HDFSTableHandle table = checkType(tableHandle, HDFSTableHandle.class, "table");
    List<HDFSColumnHandle> cols = metaDataQuery.getTableColumnHandle(connectorId, table.getSchemaName(), table.getTableName())
            .orElse(new ArrayList<>());
    Map<String, ColumnHandle> columnMap = new HashMap<>();
    for (HDFSColumnHandle col : cols) {
        columnMap.putIfAbsent(col.getName(), col);
    }
    return columnMap;
}

@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 ConnectorHandleResolver getHandleResolver()
{
    return new ConnectorHandleResolver()
    {
        public Class<? extends ConnectorTableHandle> getTableHandleClass()
        {
            return RestTableHandle.class;
        }

        public Class<? extends ColumnHandle> getColumnHandleClass()
        {
            return RestColumnHandle.class;
        }

        public Class<? extends ConnectorSplit> getSplitClass()
        {
            return RestConnectorSplit.class;
        }

        public Class<? extends ConnectorTableLayoutHandle> getTableLayoutHandleClass()
        {
            return RestConnectorTableLayoutHandle.class;
        }

        @Override
        public Class<? extends ConnectorTransactionHandle> getTransactionHandleClass()
        {
            return RestTransactionHandle.class;
        }

        @Override
        public Class<? extends ConnectorInsertTableHandle> getInsertTableHandleClass()
        {
            return RestInsertTableHandle.class;
        }
    };
}

@Override
public List<ConnectorTableLayoutResult> getTableLayouts(ConnectorSession connectorSession, ConnectorTableHandle connectorTableHandle, Constraint<ColumnHandle> constraint, Optional<Set<ColumnHandle>> optional)
{
    RestTableHandle tableHandle = Types.checkType(connectorTableHandle, RestTableHandle.class, "tableHandle");
    return ImmutableList.of(
            new ConnectorTableLayoutResult(
                    getTableLayout(connectorSession, new RestConnectorTableLayoutHandle(tableHandle)),
                    TupleDomain.all()));
}

@Override
public ColumnMetadata getColumnMetadata(
        ConnectorSession session,
        ConnectorTableHandle tableHandle,
        ColumnHandle columnHandle
) {
    convertTableHandle(tableHandle);
    return convertColumnHandle(columnHandle).getColumnMetadata();
}

@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 List<ConnectorTableLayoutResult> getTableLayouts(ConnectorSession session, ConnectorTableHandle table, Constraint<ColumnHandle> constraint, Optional<Set<ColumnHandle>> desiredColumns)
{
    KuduTableHandle tableHandle = checkType(table, KuduTableHandle.class, "tableHandle");
    ConnectorTableLayout layout = new ConnectorTableLayout(new KuduTableLayoutHandle(tableHandle, constraint.getSummary()));
    return ImmutableList.of(new ConnectorTableLayoutResult(layout, constraint.getSummary()));
}

private Map<String, ColumnHandle> getColumnHandles(KuduTableHandle tableHandle)
{
    ImmutableMap.Builder<String, ColumnHandle> columnHandles = ImmutableMap.builder();
    int index = 0;
    KuduClient kuduClient = kuduClientManager.getClient();

    for (ColumnMetadata column : kuduTables.getColumns(kuduClient, tableHandle)) {
        int ordinalPosition;
        ordinalPosition = index;
        index++;
        columnHandles.put(column.getName(), new KuduColumnHandle(column.getName(), column.getType(), ordinalPosition));
    }
    kuduClientManager.close(kuduClient);
    return columnHandles.build();
}

@JsonCreator
public KuduTableLayoutHandle(
        @JsonProperty("table") KuduTableHandle table,
        @JsonProperty("constraint") TupleDomain<ColumnHandle> constraint)
{
    this.table = requireNonNull(table, "table is null");
    this.constraint = requireNonNull(constraint, "constraint is null");
}

@Override
public ConnectorPageSource createPageSource(Session session, Split split, List<ColumnHandle> columns)
{
    requireNonNull(split, "split is null");
    requireNonNull(columns, "columns is null");

    // assumes connectorId and catalog are the same
    ConnectorSession connectorSession = session.toConnectorSession(split.getConnectorId());
    return getPageSourceProvider(split).createPageSource(split.getTransactionHandle(), connectorSession, split.getConnectorSplit(), columns);
}

@Override
public Map<String, ColumnHandle> getColumnHandles(ConnectorSession session, ConnectorTableHandle tableHandle)
{
    ImmutableMap.Builder<String, ColumnHandle> builder = ImmutableMap.builder();
    int index = 0;
    for (ColumnMetadata columnMetadata : getTableMetadata(session, tableHandle).getColumns()) {
        builder.put(columnMetadata.getName(), new InMemoryColumnHandle(columnMetadata.getName(), index, columnMetadata.getType()));
        index++;
    }
    return builder.build();
}

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

@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 Map<String, ColumnHandle> getColumnHandles(ConnectorSession session, ConnectorTableHandle tableHandle)
{
    RaptorTableHandle raptorTableHandle = checkType(tableHandle, RaptorTableHandle.class, "tableHandle");
    ImmutableMap.Builder<String, ColumnHandle> builder = ImmutableMap.builder();
    for (TableColumn tableColumn : dao.listTableColumns(raptorTableHandle.getTableId())) {
        if (tableColumn.getColumnName().equals(SAMPLE_WEIGHT_COLUMN_NAME)) {
            continue;
        }
        builder.put(tableColumn.getColumnName(), getRaptorColumnHandle(tableColumn));
    }
    RaptorColumnHandle uuidColumn = shardUuidColumnHandle(connectorId);
    builder.put(uuidColumn.getColumnName(), uuidColumn);
    return builder.build();
}

private List<String> getFilteredPartitionNames(HiveMetastore metastore, SchemaTableName tableName, List<HiveColumnHandle> partitionKeys, TupleDomain<ColumnHandle> effectivePredicate)
{
    checkArgument(effectivePredicate.getDomains().isPresent());

    List<String> filter = new ArrayList<>();
    for (HiveColumnHandle partitionKey : partitionKeys) {
        Domain domain = effectivePredicate.getDomains().get().get(partitionKey);
        if (domain != null && domain.isNullableSingleValue()) {
            Object value = domain.getNullableSingleValue();
            if (value == null) {
                filter.add(HivePartitionKey.HIVE_DEFAULT_DYNAMIC_PARTITION);
            }
            else if (value instanceof Slice) {
                filter.add(((Slice) value).toStringUtf8());
            }
            else if ((value instanceof Boolean) || (value instanceof Double) || (value instanceof Long)) {
                if (assumeCanonicalPartitionKeys) {
                    filter.add(value.toString());
                }
                else {
                    // Hive treats '0', 'false', and 'False' the same. However, the metastore differentiates between these.
                    filter.add(PARTITION_VALUE_WILDCARD);
                }
            }
            else {
                throw new PrestoException(NOT_SUPPORTED, "Only Boolean, Double and Long partition keys are supported");
            }
        }
        else {
            filter.add(PARTITION_VALUE_WILDCARD);
        }
    }

    // fetch the partition names
    return metastore.getPartitionNamesByParts(tableName.getSchemaName(), tableName.getTableName(), filter)
            .orElseThrow(() -> new TableNotFoundException(tableName));
}

@Test
public void testNone()
        throws Exception
{
    Assert.assertTrue(TupleDomain.none().isNone());
    Assert.assertEquals(TupleDomain.<ColumnHandle>none(),
            TupleDomain.withColumnDomains(ImmutableMap.of(
                    A, Domain.none(BIGINT))));
    Assert.assertEquals(TupleDomain.<ColumnHandle>none(),
            TupleDomain.withColumnDomains(ImmutableMap.of(
                    A, Domain.all(BIGINT),
                    B, Domain.none(VARCHAR))));
}

@Override
public ConnectorIndex getIndex(ConnectorSession session, ConnectorIndexHandle indexHandle, List<ColumnHandle> lookupSchema, List<ColumnHandle> outputSchema)
{
    try (ThreadContextClassLoader ignored = new ThreadContextClassLoader(classLoader)) {
        return delegate.getIndex(session, indexHandle, lookupSchema, outputSchema);
    }
}

@Override
public Map<String, ColumnHandle> getColumnHandles(Session session, TableHandle tableHandle)
{
    ConnectorEntry entry = lookupConnectorFor(tableHandle);
    ConnectorMetadata metadata = entry.getMetadata(session);
    return metadata.getColumnHandles(session.toConnectorSession(entry.getCatalog()), tableHandle.getConnectorHandle());
}

@Test
public void testIntersection()
        throws Exception
{
    TupleDomain<ColumnHandle> tupleDomain1 = TupleDomain.withColumnDomains(
            ImmutableMap.<ColumnHandle, Domain>builder()
                    .put(A, Domain.all(VARCHAR))
                    .put(B, Domain.notNull(DOUBLE))
                    .put(C, Domain.singleValue(BIGINT, 1L))
                    .put(D, Domain.create(ValueSet.ofRanges(Range.greaterThanOrEqual(DOUBLE, 0.0)), true))
                    .build());

    TupleDomain<ColumnHandle> tupleDomain2 = TupleDomain.withColumnDomains(
            ImmutableMap.<ColumnHandle, Domain>builder()
                    .put(A, Domain.singleValue(VARCHAR, utf8Slice("value")))
                    .put(B, Domain.singleValue(DOUBLE, 0.0))
                    .put(C, Domain.singleValue(BIGINT, 1L))
                    .put(D, Domain.create(ValueSet.ofRanges(Range.lessThan(DOUBLE, 10.0)), false))
                    .build());

    TupleDomain<ColumnHandle> expectedTupleDomain = TupleDomain.withColumnDomains(
            ImmutableMap.<ColumnHandle, Domain>builder()
                    .put(A, Domain.singleValue(VARCHAR, utf8Slice("value")))
                    .put(B, Domain.singleValue(DOUBLE, 0.0))
                    .put(C, Domain.singleValue(BIGINT, 1L))
                    .put(D, Domain.create(ValueSet.ofRanges(Range.range(DOUBLE, 0.0, true, 10.0, false)), false))
                    .build());

    Assert.assertEquals(tupleDomain1.intersect(tupleDomain2), expectedTupleDomain);
}

@Override
public ConnectorTableLayout getTableLayout(ConnectorSession session, ConnectorTableLayoutHandle handle)
{
    RedisTableLayoutHandle layout = convertLayout(handle);

    // tables in this connector have a single layout
    return getTableLayouts(session, layout.getTable(), Constraint.<ColumnHandle>alwaysTrue(), Optional.empty())
            .get(0)
            .getTableLayout();
}

@Override
public List<ConnectorTableLayoutResult> getTableLayouts(
        ConnectorSession session,
        ConnectorTableHandle table,
        Constraint<ColumnHandle> constraint,
        Optional<Set<ColumnHandle>> desiredColumns)
{
    try (ThreadContextClassLoader ignored = new ThreadContextClassLoader(classLoader)) {
        return delegate.getTableLayouts(session, table, constraint, desiredColumns);
    }
}