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

@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 ConnectorTableMetadata getTableMetadata(ConnectorSession session, ConnectorTableHandle tableHandle)
{
    RaptorTableHandle handle = checkType(tableHandle, RaptorTableHandle.class, "tableHandle");
    SchemaTableName tableName = new SchemaTableName(handle.getSchemaName(), handle.getTableName());
    List<ColumnMetadata> columns = dao.getTableColumns(handle.getTableId()).stream()
            .map(TableColumn::toColumnMetadata)
            .filter(isSampleWeightColumn().negate())
            .collect(toCollection(ArrayList::new));
    if (columns.isEmpty()) {
        throw new PrestoException(RAPTOR_ERROR, "Table does not have any columns: " + tableName);
    }

    columns.add(hiddenColumn(SHARD_UUID_COLUMN_NAME, VARCHAR));
    return new ConnectorTableMetadata(tableName, columns);
}

@Override
public void addColumn(ConnectorSession session, ConnectorTableHandle tableHandle, ColumnMetadata column)
{
    RaptorTableHandle table = checkType(tableHandle, RaptorTableHandle.class, "tableHandle");

    // Always add new columns to the end.
    // TODO: This needs to be updated when we support dropping columns.
    List<TableColumn> existingColumns = dao.listTableColumns(table.getSchemaName(), table.getTableName());
    TableColumn lastColumn = existingColumns.get(existingColumns.size() - 1);
    long columnId = lastColumn.getColumnId() + 1;
    int ordinalPosition = existingColumns.size();

    String type = column.getType().getTypeSignature().toString();
    dao.insertColumn(table.getTableId(), columnId, column.getName(), ordinalPosition, type, null);
    shardManager.addColumn(table.getTableId(), new ColumnInfo(columnId, column.getType()));
}

@Override
public void finishDelete(ConnectorSession session, ConnectorTableHandle tableHandle, Collection<Slice> fragments)
{
    RaptorTableHandle table = checkType(tableHandle, RaptorTableHandle.class, "tableHandle");
    long transactionId = table.getTransactionId().getAsLong();
    long tableId = table.getTableId();

    List<ColumnInfo> columns = getColumnHandles(session, tableHandle).values().stream()
            .map(handle -> checkType(handle, RaptorColumnHandle.class, "columnHandle"))
            .map(ColumnInfo::fromHandle).collect(toList());

    ImmutableSet.Builder<UUID> oldShardUuids = ImmutableSet.builder();
    ImmutableList.Builder<ShardInfo> newShards = ImmutableList.builder();

    fragments.stream()
            .map(fragment -> shardDeltaCodec.fromJson(fragment.getBytes()))
            .forEach(delta -> {
                oldShardUuids.addAll(delta.getOldShardUuids());
                newShards.addAll(delta.getNewShards());
            });

    shardManager.replaceShardUuids(transactionId, tableId, columns, oldShardUuids.build(), newShards.build());

    clearRollback();
}

@Test
public void testRenameTable()
        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;
    SchemaTableName renamedTable = new SchemaTableName(raptorTableHandle.getSchemaName(), "orders_renamed");

    metadata.renameTable(SESSION, raptorTableHandle, renamedTable);
    assertNull(metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS));
    ConnectorTableHandle renamedTableHandle = metadata.getTableHandle(SESSION, renamedTable);
    assertNotNull(renamedTableHandle);
    assertEquals(((RaptorTableHandle) renamedTableHandle).getTableName(), renamedTable.getTableName());
}

@Test
public void testCreateTable()
{
    assertNull(metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS));

    metadata.createTable(SESSION, getOrdersTable());

    ConnectorTableHandle tableHandle = metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS);
    assertInstanceOf(tableHandle, RaptorTableHandle.class);
    assertEquals(((RaptorTableHandle) tableHandle).getTableId(), 1);

    ConnectorTableMetadata table = metadata.getTableMetadata(SESSION, tableHandle);
    assertTableEqual(table, getOrdersTable());

    ColumnHandle columnHandle = metadata.getColumnHandles(SESSION, tableHandle).get("orderkey");
    assertInstanceOf(columnHandle, RaptorColumnHandle.class);
    assertEquals(((RaptorColumnHandle) columnHandle).getColumnId(), 1);

    ColumnMetadata columnMetadata = metadata.getColumnMetadata(SESSION, tableHandle, columnHandle);
    assertNotNull(columnMetadata);
    assertEquals(columnMetadata.getName(), "orderkey");
    assertEquals(columnMetadata.getType(), BIGINT);
}

@Test
public void testTemporalColumn()
        throws Exception
{
    assertNull(metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS));

    ConnectorTableMetadata ordersTable = getOrdersTable(ImmutableMap.of(TEMPORAL_COLUMN_PROPERTY, "orderdate"));
    metadata.createTable(SESSION, ordersTable);

    ConnectorTableHandle tableHandle = metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS);
    assertInstanceOf(tableHandle, RaptorTableHandle.class);
    RaptorTableHandle raptorTableHandle = (RaptorTableHandle) tableHandle;
    assertEquals(raptorTableHandle.getTableId(), 1);

    long tableId = raptorTableHandle.getTableId();
    MetadataDao metadataDao = dbi.onDemand(MetadataDao.class);

    // verify sort columns are not set
    List<TableColumn> sortColumns = metadataDao.listSortColumns(tableId);
    assertEquals(sortColumns.size(), 0);
    assertEquals(sortColumns, ImmutableList.of());

    // verify temporal column is set
    assertEquals(metadataDao.getTemporalColumnId(tableId), Long.valueOf(4));
    metadata.dropTable(SESSION, tableHandle);
}

@Override
public HiveInsertTableHandle beginInsert(ConnectorSession session, ConnectorTableHandle tableHandle)
{
    verifyJvmTimeZone();

    SchemaTableName tableName = schemaTableName(tableHandle);
    Optional<Table> table = metastore.getTable(tableName.getSchemaName(), tableName.getTableName());
    if (!table.isPresent()) {
        throw new TableNotFoundException(tableName);
    }

    checkTableIsWritable(table.get());

    HiveStorageFormat hiveStorageFormat = extractHiveStorageFormat(table.get());

    List<HiveColumnHandle> handles = hiveColumnHandles(connectorId, table.get());

    return new HiveInsertTableHandle(
            connectorId,
            tableName.getSchemaName(),
            tableName.getTableName(),
            handles,
            session.getQueryId(),
            locationService.forExistingTable(session.getQueryId(), table.get()),
            hiveStorageFormat);
}

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

    ConnectorTableHandle tableHandle = getTableHandle(tableBucketedDoubleFloat);
    List<ColumnHandle> columnHandles = ImmutableList.copyOf(metadata.getColumnHandles(session, tableHandle).values());
    Map<String, Integer> columnIndex = indexColumns(columnHandles);

    assertTableIsBucketed(tableHandle);

    ImmutableMap<ColumnHandle, NullableValue> bindings = ImmutableMap.<ColumnHandle, NullableValue>builder()
            .put(columnHandles.get(columnIndex.get("t_float")), NullableValue.of(DOUBLE, 87.1))
            .put(columnHandles.get(columnIndex.get("t_double")), NullableValue.of(DOUBLE, 88.2))
            .build();

    // floats and doubles are not supported, so we should see all splits
    MaterializedResult result = readTable(tableHandle, columnHandles, session, TupleDomain.fromFixedValues(bindings), OptionalInt.of(32), Optional.empty());
    assertEquals(result.getRowCount(), 100);
}

@Override
public void dropTable(ConnectorSession session, ConnectorTableHandle tableHandle)
{
    HiveTableHandle handle = checkType(tableHandle, HiveTableHandle.class, "tableHandle");
    SchemaTableName tableName = schemaTableName(tableHandle);

    if (!allowDropTable) {
        throw new PrestoException(PERMISSION_DENIED, "DROP TABLE is disabled in this Hive catalog");
    }

    Optional<Table> target = metastore.getTable(handle.getSchemaName(), handle.getTableName());
    if (!target.isPresent()) {
        throw new TableNotFoundException(tableName);
    }
    Table table = target.get();

    if (!session.getUser().equals(table.getOwner())) {
        throw new PrestoException(PERMISSION_DENIED, format("Unable to drop table '%s': owner of the table is different from session user", table));
    }
    metastore.dropTable(handle.getSchemaName(), handle.getTableName());
}

/**
 * Returns a table handle for the specified table name, or null if the connector does not contain the table.
 *
 * @param session session
 * @param tableName table name
 */
@Override
public ConnectorTableHandle getTableHandle(ConnectorSession session, SchemaTableName tableName)
{
    Optional<HDFSTableHandle> table = metaDataQuery.getTableHandle(connectorId, tableName.getSchemaName(), tableName.getTableName());
    return table.orElse(null);
}

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

/**
 * Return the metadata for the specified table handle.
 *
 * @param session session
 * @param table table
 * @throws RuntimeException if table handle is no longer valid
 */
@Override
public ConnectorTableMetadata getTableMetadata(ConnectorSession session, ConnectorTableHandle table)
{
    HDFSTableHandle hdfsTable = checkType(table, HDFSTableHandle.class, "table");
    SchemaTableName tableName = hdfsTable.getSchemaTableName();
    return getTableMetadata(tableName);
}

/**
 * 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 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 ConnectorTableHandle getTableHandle(ConnectorSession connectorSession, SchemaTableName schemaTableName)
{
    if (rest.listTables().contains(schemaTableName)) {
        return new RestTableHandle(schemaTableName);
    }
    return null;
}

@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 ConnectorTableHandle getTableHandle(ConnectorSession session, SchemaTableName tableName)
{
    requireNonNull(tableName, "tableName is null");
    KuduClient kuduClient = kuduClientManager.getClient();
    ConnectorTableHandle connectorTableHandle = kuduTables.getTables(kuduClient).get(tableName);
    kuduClientManager.close(kuduClient);
    return connectorTableHandle;
}

@Override
public ConnectorTableMetadata getTableMetadata(ConnectorSession session, ConnectorTableHandle table)
{
    KuduClient kuduClient = kuduClientManager.getClient();
    KuduTableHandle tableHandle = checkType(table, KuduTableHandle.class, "tableHandle");
    ConnectorTableMetadata connectorTableMetadata = new ConnectorTableMetadata(
            tableHandle.getSchemaTableName(), kuduTables.getColumns(kuduClient, tableHandle));
    kuduClientManager.close(kuduClient);
    return connectorTableMetadata;
}

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

@Override
public void dropTable(ConnectorSession session, ConnectorTableHandle tableHandle)
{
    KuduClient kuduClient = kuduClientManager.getClient();
    kuduClientManager.dropTable(kuduClient, ((KuduTableHandle) tableHandle).getSchemaTableName().getTableName());
    kuduClientManager.close(kuduClient);
}

@Override
public ConnectorResolvedIndex resolveIndex(
        ConnectorSession session,
        ConnectorTableHandle tableHandle,
        Set<ColumnHandle> indexableColumns,
        Set<ColumnHandle> outputColumns,
        TupleDomain<ColumnHandle> tupleDomain)
{
    try (ThreadContextClassLoader ignored = new ThreadContextClassLoader(classLoader)) {
        return delegate.resolveIndex(session, tableHandle, indexableColumns, outputColumns, tupleDomain);
    }
}

@Override
public ConnectorTableMetadata getTableMetadata(ConnectorSession connectorSession, ConnectorTableHandle tableHandle)
{
    KinesisTableHandle kinesisTableHandle = handleResolver.convertTableHandle(tableHandle);
    log.debug("Called getTableMetadata on %s.%s", kinesisTableHandle.getSchemaName(), kinesisTableHandle.getTableName());
    return getTableMetadata(kinesisTableHandle.toSchemaTableName());
}

@Override
public List<ConnectorTableLayoutResult> getTableLayouts(ConnectorSession session, ConnectorTableHandle table, Constraint<ColumnHandle> constraint, Optional<Set<ColumnHandle>> desiredColumns)
{
    KafkaTableHandle handle = convertTableHandle(table);
    ConnectorTableLayout layout = new ConnectorTableLayout(new KafkaTableLayoutHandle(handle));
    return ImmutableList.of(new ConnectorTableLayoutResult(layout, constraint.getSummary()));
}

@Override
public List<ConnectorTableLayoutResult> getTableLayouts(ConnectorSession session, ConnectorTableHandle table, Constraint<ColumnHandle> constraint, Optional<Set<ColumnHandle>> desiredColumns)
{
    ExampleTableHandle tableHandle = checkType(table, ExampleTableHandle.class, "table");
    ConnectorTableLayout layout = new ConnectorTableLayout(new ExampleTableLayoutHandle(tableHandle));
    return ImmutableList.of(new ConnectorTableLayoutResult(layout, constraint.getSummary()));
}

@Override
public ConnectorTableMetadata getTableMetadata(ConnectorSession session, ConnectorTableHandle table)
{
    ExampleTableHandle exampleTableHandle = checkType(table, ExampleTableHandle.class, "table");
    checkArgument(exampleTableHandle.getConnectorId().equals(connectorId), "tableHandle is not for this connector");
    SchemaTableName tableName = new SchemaTableName(exampleTableHandle.getSchemaName(), exampleTableHandle.getTableName());

    return getTableMetadata(tableName);
}

protected void doCreateTable(SchemaTableName tableName, HiveStorageFormat storageFormat)
        throws Exception
{
    ConnectorSession session = newSession();

    // begin creating the table
    ConnectorTableMetadata tableMetadata = new ConnectorTableMetadata(tableName, CREATE_TABLE_COLUMNS, createTableProperties(storageFormat), session.getUser());

    ConnectorOutputTableHandle outputHandle = metadata.beginCreateTable(session, tableMetadata);

    // write the data
    ConnectorPageSink sink = pageSinkProvider.createPageSink(session, outputHandle);
    sink.appendPage(CREATE_TABLE_DATA.toPage(), null);
    Collection<Slice> fragments = sink.finish();

    // verify all new files start with the unique prefix
    for (String filePath : listAllDataFiles(outputHandle)) {
        assertTrue(new Path(filePath).getName().startsWith(getFilePrefix(outputHandle)));
    }

    // commit the table
    metadata.commitCreateTable(session, outputHandle, fragments);

    // load the new table
    ConnectorTableHandle tableHandle = getTableHandle(tableName);
    List<ColumnHandle> columnHandles = ImmutableList.copyOf(metadata.getColumnHandles(session, tableHandle).values());

    // verify the metadata
    tableMetadata = metadata.getTableMetadata(session, getTableHandle(tableName));
    assertEquals(tableMetadata.getOwner(), session.getUser());
    assertEquals(tableMetadata.getColumns(), CREATE_TABLE_COLUMNS);

    // verify the data
    MaterializedResult result = readTable(tableHandle, columnHandles, session, TupleDomain.all(), OptionalInt.empty(), Optional.of(storageFormat));
    assertEqualsIgnoreOrder(result.getMaterializedRows(), CREATE_TABLE_DATA.getMaterializedRows());
}

private static SchemaTableName getTableName(ConnectorTableHandle tableHandle)
{
    requireNonNull(tableHandle, "tableHandle is null");
    checkArgument(tableHandle instanceof InMemoryTableHandle, "tableHandle is not an instance of InMemoryTableHandle");
    InMemoryTableHandle inMemoryTableHandle = (InMemoryTableHandle) tableHandle;
    return inMemoryTableHandle.getTableName();
}

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

@Override
public Map<String, ColumnHandle> getColumnHandles(ConnectorSession session, ConnectorTableHandle tableHandle)
{
    try (ThreadContextClassLoader ignored = new ThreadContextClassLoader(classLoader)) {
        return delegate.getColumnHandles(session, tableHandle);
    }
}

@Override
public void renameTable(ConnectorSession session, ConnectorTableHandle tableHandle, SchemaTableName newTableName)
{
    RaptorTableHandle table = checkType(tableHandle, RaptorTableHandle.class, "tableHandle");
    runTransaction(dbi, (handle, status) -> {
        MetadataDao dao = handle.attach(MetadataDao.class);
        dao.renameTable(table.getTableId(), newTableName.getSchemaName(), newTableName.getTableName());
        return null;
    });
}

@Override
public void renameColumn(ConnectorSession session, ConnectorTableHandle tableHandle, ColumnHandle source, String target)
{
    RaptorTableHandle table = checkType(tableHandle, RaptorTableHandle.class, "tableHandle");
    RaptorColumnHandle sourceColumn = checkType(source, RaptorColumnHandle.class, "columnHandle");
    dao.renameColumn(table.getTableId(), sourceColumn.getColumnId(), target);
}

@Override
public ConnectorInsertTableHandle beginInsert(ConnectorSession session, ConnectorTableHandle tableHandle)
{
    long tableId = checkType(tableHandle, RaptorTableHandle.class, "tableHandle").getTableId();

    ImmutableList.Builder<RaptorColumnHandle> columnHandles = ImmutableList.builder();
    ImmutableList.Builder<Type> columnTypes = ImmutableList.builder();
    for (TableColumn column : dao.getTableColumns(tableId)) {
        columnHandles.add(new RaptorColumnHandle(connectorId, column.getColumnName(), column.getColumnId(), column.getDataType()));
        columnTypes.add(column.getDataType());
    }

    long transactionId = shardManager.beginTransaction();

    setTransactionId(transactionId);

    Optional<String> externalBatchId = getExternalBatchId(session);
    List<RaptorColumnHandle> sortColumnHandles = getSortColumnHandles(tableId);
    return new RaptorInsertTableHandle(connectorId,
            transactionId,
            tableId,
            columnHandles.build(),
            columnTypes.build(),
            externalBatchId,
            sortColumnHandles,
            nCopies(sortColumnHandles.size(), ASC_NULLS_FIRST));
}

@Override
public ConnectorTableMetadata getTableMetadata(ConnectorSession session, ConnectorTableHandle tableHandle)
{
    requireNonNull(tableHandle, "tableHandle is null");
    SchemaTableName tableName = schemaTableName(tableHandle);
    return getTableMetadata(tableName);
}

@Override
public boolean supportsMetadataDelete(ConnectorSession session, ConnectorTableHandle tableHandle, ConnectorTableLayoutHandle tableLayoutHandle)
{
    try (ThreadContextClassLoader ignored = new ThreadContextClassLoader(classLoader)) {
        return delegate.supportsMetadataDelete(session, tableHandle, tableLayoutHandle);
    }
}

@Override
public List<ConnectorTableLayoutResult> getTableLayouts(ConnectorSession connectorSession, ConnectorTableHandle table,
                                                        Constraint<ColumnHandle> constraint, Optional<Set<ColumnHandle>> optional)
{
    KinesisTableHandle tblHandle = handleResolver.convertTableHandle(table);
    ConnectorTableLayout layout = new ConnectorTableLayout(new KinesisTableLayoutHandle(connectorId, tblHandle));
    return ImmutableList.of(new ConnectorTableLayoutResult(layout, constraint.getSummary()));
 }

@Override
public List<ConnectorTableLayoutResult> getTableLayouts(
        ConnectorSession session,
        ConnectorTableHandle table,
        Constraint<ColumnHandle> constraint,
        Optional<Set<ColumnHandle>> desiredColumns)
{
    RedisTableHandle tableHandle = convertTableHandle(table);

    ConnectorTableLayout layout = new ConnectorTableLayout(new RedisTableLayoutHandle(tableHandle));

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

@Override
public List<ConnectorTableLayoutResult> getTableLayouts(ConnectorSession session, ConnectorTableHandle table,
    Constraint<ColumnHandle> constraint, Optional<Set<ColumnHandle>> desiredColumns) {
  SpreadsheetTableHandle tableHandle = (SpreadsheetTableHandle) table;
  SpreadsheetTableLayoutHandle baseTableLayoutHandle = createTableLayoutHandle(tableHandle);
  ConnectorTableLayout layout = new ConnectorTableLayout(baseTableLayoutHandle);
  return ImmutableList.of(new ConnectorTableLayoutResult(layout, constraint.getSummary()));
}

@Override
public ConnectorTableMetadata getTableMetadata(ConnectorSession session, ConnectorTableHandle tableHandle)
{
    requireNonNull(tableHandle, "tableHandle is null");
    SchemaTableName tableName = getTableName(tableHandle);
    return getTableMetadata(session, tableName);
}