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

@Override
public void createTable(ConnectorSession session, ConnectorTableMetadata tableMetadata)
{
    log.debug("Create table " + tableMetadata.getTable().getTableName());
    String tblName = tableMetadata.getTable().getTableName();
    String dbName = tableMetadata.getTable().getSchemaName();
    List<ColumnMetadata> columns = tableMetadata.getColumns();
    List<String> columnName = new LinkedList<>();
    List<String> dataType = new LinkedList<>();
    for (ColumnMetadata column : columns) {
        columnName.add(column.getName());
        dataType.add(column.getType().getDisplayName());
    }
    String userName = "";
    String storageFormatName = "";
    metaClient.createRegularTable(dbName, tblName, userName, storageFormatName, columnName, dataType);
}

@Override
public Map<SchemaTableName, List<ColumnMetadata>> listTableColumns(ConnectorSession session, SchemaTablePrefix prefix)
{
    requireNonNull(prefix, "prefix is null");

    ImmutableMap.Builder<SchemaTableName, List<ColumnMetadata>> columns = ImmutableMap.builder();

    List<SchemaTableName> tableNames = prefix.getSchemaName() == null ? listTables(session, null) : ImmutableList.of(new SchemaTableName(prefix.getSchemaName(), prefix.getTableName()));

    for (SchemaTableName tableName : tableNames) {
        ConnectorTableMetadata tableMetadata = getTableMetadata(tableName);
        // table can disappear during listing operation
        if (tableMetadata != null) {
            columns.put(tableName, tableMetadata.getColumns());
        }
    }
    return columns.build();
}

@Override
public Map<SchemaTableName, List<ColumnMetadata>> listTableColumns(ConnectorSession session, SchemaTablePrefix prefix)
{
    requireNonNull(prefix, "prefix is null");

    ImmutableMap.Builder<SchemaTableName, List<ColumnMetadata>> columns = ImmutableMap.builder();

    List<SchemaTableName> tableNames = prefix.getSchemaName() == null ? listTables(session, null) : ImmutableList.of(new SchemaTableName(prefix.getSchemaName(), prefix.getTableName()));

    for (SchemaTableName tableName : tableNames) {
        ConnectorTableMetadata tableMetadata = getTableMetadata(tableName);
        // table can disappear during listing operation
        if (tableMetadata != null) {
            columns.put(tableName, tableMetadata.getColumns());
        }
    }
    return columns.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);
}

@SuppressWarnings({"ValueOfIncrementOrDecrementUsed", "UnusedAssignment"})
@Test
public void testGetTableSchemaPartitionFormat()
        throws Exception
{
    ConnectorTableMetadata tableMetadata = metadata.getTableMetadata(newSession(), getTableHandle(tablePartitionFormat));
    Map<String, ColumnMetadata> map = uniqueIndex(tableMetadata.getColumns(), ColumnMetadata::getName);

    assertPrimitiveField(map, "t_string", VARCHAR, false);
    assertPrimitiveField(map, "t_tinyint", BIGINT, false);
    assertPrimitiveField(map, "t_smallint", BIGINT, false);
    assertPrimitiveField(map, "t_int", BIGINT, false);
    assertPrimitiveField(map, "t_bigint", BIGINT, false);
    assertPrimitiveField(map, "t_float", DOUBLE, false);
    assertPrimitiveField(map, "t_double", DOUBLE, false);
    assertPrimitiveField(map, "t_boolean", BOOLEAN, false);
    assertPrimitiveField(map, "ds", VARCHAR, true);
    assertPrimitiveField(map, "file_format", VARCHAR, true);
    assertPrimitiveField(map, "dummy", BIGINT, true);
}

@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 testCreateTableUnsupportedType()
{
    for (HiveStorageFormat storageFormat : createTableFormats) {
        try {
            ConnectorSession session = newSession();
            List<ColumnMetadata> columns = ImmutableList.of(new ColumnMetadata("dummy", HYPER_LOG_LOG, false));
            ConnectorTableMetadata tableMetadata = new ConnectorTableMetadata(invalidTable, columns, createTableProperties(storageFormat), session.getUser());
            metadata.beginCreateTable(session, tableMetadata);
            fail("create table with unsupported type should fail for storage format " + storageFormat);
        }
        catch (PrestoException e) {
            assertEquals(e.getErrorCode(), NOT_SUPPORTED.toErrorCode());
        }
    }
}

private static ConnectorTableMetadata getOrdersTable(Map<String, Object> properties)
{
    MetadataUtil.TableMetadataBuilder builder = tableMetadataBuilder(DEFAULT_TEST_ORDERS)
            .column("orderkey", BIGINT)
            .column("custkey", BIGINT)
            .column("totalprice", DOUBLE)
            .column("orderdate", DATE);

    if (!properties.isEmpty()) {
        for (Map.Entry<String, Object> entry : properties.entrySet()) {
            builder.property(entry.getKey(), entry.getValue());
        }
    }

    return builder.build();
}

private static void assertTableEqual(ConnectorTableMetadata actual, ConnectorTableMetadata expected)
{
    assertEquals(actual.getTable(), expected.getTable());

    List<ColumnMetadata> actualColumns = actual.getColumns().stream()
            .filter(columnMetadata -> !columnMetadata.isHidden())
            .collect(Collectors.toList());

    List<ColumnMetadata> expectedColumns = expected.getColumns();
    assertEquals(actualColumns.size(), expectedColumns.size());
    for (int i = 0; i < actualColumns.size(); i++) {
        ColumnMetadata actualColumn = actualColumns.get(i);
        ColumnMetadata expectedColumn = expectedColumns.get(i);
        assertEquals(actualColumn.getName(), expectedColumn.getName());
        assertEquals(actualColumn.getType(), expectedColumn.getType());
    }
}

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

public BlackHoleTableHandle(
        ConnectorTableMetadata tableMetadata,
        int splitCount,
        int pagesPerSplit,
        int rowsPerPage,
        int fieldsLength)
{
    this(tableMetadata.getTable().getSchemaName(),
            tableMetadata.getTable().getTableName(),
            tableMetadata.getColumns().stream()
                    .map(BlackHoleColumnHandle::new)
                    .collect(toList()),
            splitCount,
            pagesPerSplit,
            rowsPerPage,
            fieldsLength);
}

@Test
public void tableIsCreatedAfterCommits()
{
    assertThatNoTableIsCreated();

    SchemaTableName schemaTableName = new SchemaTableName("default", "temp_table");

    ConnectorOutputTableHandle table = metadata.beginCreateTable(
            SESSION,
            new ConnectorTableMetadata(
                    schemaTableName,
                    ImmutableList.of(),
                    tableProperties,
                    null));

    assertThatNoTableIsCreated();

    metadata.finishCreateTable(SESSION, table, ImmutableList.of());

    List<SchemaTableName> tables = metadata.listTables(SESSION, null);
    assertTrue(tables.size() == 1, "Expected only one table.");
    assertTrue(tables.get(0).getTableName().equals("temp_table"), "Expected table with name 'temp_table'");
}

@Test
public void getTableMetadata()
{
    // known table
    ConnectorTableMetadata tableMetadata = metadata.getTableMetadata(SESSION, tableHandle);
    assertEquals(tableMetadata.getTable(), new SchemaTableName("example", "numbers"));
    assertEquals(tableMetadata.getColumns(), ImmutableList.of(
            new ColumnMetadata("text", VARCHAR, false),
            new ColumnMetadata("value", BIGINT, false)));

    // escaping name patterns
    JdbcTableHandle specialTableHandle = metadata.getTableHandle(SESSION, new SchemaTableName("exa_ple", "num_ers"));
    ConnectorTableMetadata specialTableMetadata = metadata.getTableMetadata(SESSION, specialTableHandle);
    assertEquals(specialTableMetadata.getTable(), new SchemaTableName("exa_ple", "num_ers"));
    assertEquals(specialTableMetadata.getColumns(), ImmutableList.of(
            new ColumnMetadata("te_t", VARCHAR, false),
            new ColumnMetadata("va%ue", BIGINT, false)));

    // unknown tables should produce null
    unknownTableMetadata(new JdbcTableHandle(CONNECTOR_ID, new SchemaTableName("u", "numbers"), null, "unknown", "unknown"));
    unknownTableMetadata(new JdbcTableHandle(CONNECTOR_ID, new SchemaTableName("example", "numbers"), null, "example", "unknown"));
    unknownTableMetadata(new JdbcTableHandle(CONNECTOR_ID, new SchemaTableName("example", "numbers"), null, "unknown", "numbers"));
}

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

@Override
    public void createTableWithFiber(ConnectorSession session, ConnectorTableMetadata tableMetadata, String fiberKey, String function, String timeKey)
    {
        log.debug("Create table with fiber " + tableMetadata.getTable().getTableName());
        // check fiberKey, function and timeKey
        List<ColumnMetadata> columns = tableMetadata.getColumns();
//        List<String> columnNames = columns.stream()
//                .map(ColumnMetadata::getName)
//                .collect(Collectors.toList());
        List<String> columnName = new LinkedList<>();
        List<String> dataType = new LinkedList<>();
        for (ColumnMetadata column : columns) {
            columnName.add(column.getName());
            dataType.add(column.getType().getDisplayName());
        }

        String tblName = tableMetadata.getTable().getTableName();
        String dbName = tableMetadata.getTable().getSchemaName();
        String storageFormatName = "";
        String userName = "";
        int fiberColIndex = Integer.parseInt(fiberKey);
        int timstampColIndex = Integer.parseInt(timeKey);
        // createTable
        metaClient.createFiberTable(dbName, tblName, userName,
                storageFormatName, fiberColIndex, function,
                timstampColIndex, columnName, dataType);
    }

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

@Override
public ConnectorTableMetadata getTableMetadata(SchemaTableName schemaTableName)
{
    return new ConnectorTableMetadata(
            schemaTableName,
            ImmutableList.of(
                    new ColumnMetadata("type", createUnboundedVarcharType()),
                    new ColumnMetadata("user", createUnboundedVarcharType()),
                    new ColumnMetadata("text", createUnboundedVarcharType())));
}

@Override
public ConnectorTableMetadata getTableMetadata(SchemaTableName schemaTableName)
{
    return new ConnectorTableMetadata(
            schemaTableName,
            ImmutableList.of(
                    new ColumnMetadata("number", BigintType.BIGINT),
                    new ColumnMetadata("state", createUnboundedVarcharType()),
                    new ColumnMetadata("user", createUnboundedVarcharType()),
                    new ColumnMetadata("title", createUnboundedVarcharType())));
}

@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 ConnectorTableMetadata getTableMetadata(SchemaTableName schemaTableName)
{
    return new ConnectorTableMetadata(
            schemaTableName,
            ImmutableList.of(
                    new ColumnMetadata("id", createUnboundedVarcharType()),
                    new ColumnMetadata("text", createUnboundedVarcharType()),
                    new ColumnMetadata("retweet_count", BigintType.BIGINT),
                    new ColumnMetadata("user_name", createUnboundedVarcharType()),
                    new ColumnMetadata("user_screen_name", createUnboundedVarcharType())));
}

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

@Override
public Map<SchemaTableName, List<ColumnMetadata>> listTableColumns(ConnectorSession session, SchemaTablePrefix prefix)
{
    requireNonNull(prefix, "prefix is null");
    ImmutableMap.Builder<SchemaTableName, List<ColumnMetadata>> columns = ImmutableMap.builder();
    for (SchemaTableName tableName : listTables(session, prefix)) {
        ConnectorTableMetadata tableMetadata = getTableMetadata(tableName);
        // table can disappear during listing operation
        if (tableMetadata != null) {
            columns.put(tableName, tableMetadata.getColumns());
        }
    }
    return columns.build();
}

@Override
public ConnectorTableMetadata getTableMetadata(ConnectorSession session, ConnectorTableHandle table) {
  SpreadsheetTableHandle baseTableHandle = (SpreadsheetTableHandle) table;
  Builder<ColumnMetadata> builder = ImmutableList.builder();
  Map<String, ColumnHandle> columnHandles = getColumnHandles(session, table);
  for (Entry<String, ColumnHandle> e : columnHandles.entrySet()) {
    ColumnMetadata columnMetadata = getColumnMetadata(session, table, e.getValue());
    builder.add(columnMetadata);
  }
  return new ConnectorTableMetadata(baseTableHandle.getTableName(), builder.build());
}

@Test
public void getTableMetadata()
{
    // known table
    ConnectorTableMetadata tableMetadata = metadata.getTableMetadata(SESSION, NUMBERS_TABLE_HANDLE);
    assertEquals(tableMetadata.getTable(), new SchemaTableName("example", "numbers"));
    assertEquals(tableMetadata.getColumns(), ImmutableList.of(
            new ColumnMetadata("text", VARCHAR, false),
            new ColumnMetadata("value", BIGINT, false)));

    // unknown tables should produce null
    assertNull(metadata.getTableMetadata(SESSION, new ExampleTableHandle(CONNECTOR_ID, "unknown", "unknown")));
    assertNull(metadata.getTableMetadata(SESSION, new ExampleTableHandle(CONNECTOR_ID, "example", "unknown")));
    assertNull(metadata.getTableMetadata(SESSION, new ExampleTableHandle(CONNECTOR_ID, "unknown", "numbers")));
}

@Test(expectedExceptions = PrestoException.class)
public void testCreateTable()
{
    metadata.createTable(SESSION, new ConnectorTableMetadata(
            new SchemaTableName("example", "foo"),
            ImmutableList.of(new ColumnMetadata("text", VARCHAR, false))));
}

private static int getColumnIndex(ConnectorTableMetadata tableMetadata, String columnName)
{
    List<ColumnMetadata> columns = tableMetadata.getColumns();
    for (int i = 0; i < columns.size(); i++) {
        if (columns.get(i).getName().equals(columnName)) {
            return i;
        }
    }
    throw new IllegalArgumentException(format("Column %s not found", columnName));
}

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

    ConnectorTableMetadata ordersTable = getOrdersTable(ImmutableMap.of(
            ORDERING_PROPERTY, ImmutableList.of("orderdate", "custkey"),
            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
    List<TableColumn> sortColumns = metadataDao.listSortColumns(tableId);
    assertEquals(sortColumns.size(), 2);
    assertEquals(sortColumns, ImmutableList.of(
            new TableColumn(DEFAULT_TEST_ORDERS, "orderdate", DATE, 4),
            new TableColumn(DEFAULT_TEST_ORDERS, "custkey", BIGINT, 2)));

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

@Test(expectedExceptions = PrestoException.class, expectedExceptionsMessageRegExp = "Ordering column does not exist: orderdatefoo")
public void testInvalidOrderingColumns()
        throws Exception
{
    assertNull(metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS));

    ConnectorTableMetadata ordersTable = getOrdersTable(ImmutableMap.of(ORDERING_PROPERTY, ImmutableList.of("orderdatefoo")));
    metadata.createTable(SESSION, ordersTable);
    fail("Expected createTable to fail");
}

@Test(expectedExceptions = PrestoException.class, expectedExceptionsMessageRegExp = "Temporal column does not exist: foo")
public void testInvalidTemporalColumn()
        throws Exception
{
    assertNull(metadata.getTableHandle(SESSION, DEFAULT_TEST_ORDERS));

    ConnectorTableMetadata ordersTable = getOrdersTable(ImmutableMap.of(TEMPORAL_COLUMN_PROPERTY, "foo"));
    metadata.createTable(SESSION, ordersTable);
    fail("Expected createTable to fail");
}

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

    ConnectorTableMetadata ordersTable = getOrdersTable(ImmutableMap.of(ORDERING_PROPERTY, ImmutableList.of("orderdate", "custkey")));
    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
    List<TableColumn> sortColumns = metadataDao.listSortColumns(tableId);
    assertEquals(sortColumns.size(), 2);
    assertEquals(sortColumns, ImmutableList.of(
            new TableColumn(DEFAULT_TEST_ORDERS, "orderdate", DATE, 4),
            new TableColumn(DEFAULT_TEST_ORDERS, "custkey", BIGINT, 2)));

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

private void createDummyTable(SchemaTableName tableName)
{
    ConnectorSession session = newSession();
    List<ColumnMetadata> columns = ImmutableList.of(new ColumnMetadata("dummy", VARCHAR, false));
    ConnectorTableMetadata tableMetadata = new ConnectorTableMetadata(tableName, columns, createTableProperties(TEXTFILE), session.getUser());
    ConnectorOutputTableHandle handle = metadata.beginCreateTable(session, tableMetadata);
    metadata.commitCreateTable(session, handle, ImmutableList.of());
}

@Override
public ConnectorOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata)
{
    int splitCount = (Integer) tableMetadata.getProperties().get(SPLIT_COUNT_PROPERTY);
    int pagesPerSplit = (Integer) tableMetadata.getProperties().get(PAGES_PER_SPLIT_PROPERTY);
    int rowsPerPage = (Integer) tableMetadata.getProperties().get(ROWS_PER_PAGE_PROPERTY);
    int fieldsLength = (Integer) tableMetadata.getProperties().get(FIELD_LENGTH_PROPERTY);

    if (splitCount < 0) {
        throw new PrestoException(INVALID_TABLE_PROPERTY, SPLIT_COUNT_PROPERTY + " property is negative");
    }
    if (pagesPerSplit < 0) {
        throw new PrestoException(INVALID_TABLE_PROPERTY, PAGES_PER_SPLIT_PROPERTY + " property is negative");
    }
    if (rowsPerPage < 0) {
        throw new PrestoException(INVALID_TABLE_PROPERTY, ROWS_PER_PAGE_PROPERTY + " property is negative");
    }

    if (((splitCount > 0) || (pagesPerSplit > 0) || (rowsPerPage > 0)) &&
            ((splitCount == 0) || (pagesPerSplit == 0) || (rowsPerPage == 0))) {
        throw new PrestoException(INVALID_TABLE_PROPERTY, format("All properties [%s, %s, %s] must be set if any are set",
                SPLIT_COUNT_PROPERTY, PAGES_PER_SPLIT_PROPERTY, ROWS_PER_PAGE_PROPERTY));
    }

    return new BlackHoleOutputTableHandle(new BlackHoleTableHandle(
            tableMetadata,
            splitCount,
            pagesPerSplit,
            rowsPerPage,
            fieldsLength));
}

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

@Override
public ConnectorTableMetadata getTableMetadata(ConnectorSession session, ConnectorTableHandle table)
{
    JdbcTableHandle handle = checkType(table, JdbcTableHandle.class, "tableHandle");

    ImmutableList.Builder<ColumnMetadata> columnMetadata = ImmutableList.builder();
    for (JdbcColumnHandle column : jdbcClient.getColumns(handle)) {
        columnMetadata.add(column.getColumnMetadata());
    }
    return new ConnectorTableMetadata(handle.getSchemaTableName(), columnMetadata.build());
}

@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 ConnectorOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata)
{
    checkState(rollbackAction.get() == null, "Cannot begin a new write while in an existing one");
    ConnectorOutputTableHandle outputTableHandle = metadata.beginCreateTable(session, tableMetadata);
    setRollback(() -> metadata.rollbackCreateTable(session, outputTableHandle));
    return outputTableHandle;
}

@Override
public ColumnMetadata getColumnMetadata(ConnectorSession session, ConnectorTableHandle tableHandle, ColumnHandle columnHandle)
{
    InformationSchemaTableHandle informationSchemaTableHandle = checkTableHandle(tableHandle);
    ConnectorTableMetadata tableMetadata = TABLES.get(informationSchemaTableHandle.getSchemaTableName());

    String columnName = checkType(columnHandle, InformationSchemaColumnHandle.class, "columnHandle").getColumnName();

    ColumnMetadata columnMetadata = findColumnMetadata(tableMetadata, columnName);
    checkArgument(columnMetadata != null, "Column %s on table %s does not exist", columnName, tableMetadata.getTable());
    return columnMetadata;
}

@Override
public Map<String, ColumnHandle> getColumnHandles(ConnectorSession session, ConnectorTableHandle tableHandle)
{
    InformationSchemaTableHandle informationSchemaTableHandle = checkTableHandle(tableHandle);

    ConnectorTableMetadata tableMetadata = TABLES.get(informationSchemaTableHandle.getSchemaTableName());

    return tableMetadata.getColumns().stream()
            .map(ColumnMetadata::getName)
            .collect(toMap(identity(), InformationSchemaColumnHandle::new));
}

@Override
public Map<SchemaTableName, List<ColumnMetadata>> listTableColumns(ConnectorSession session, SchemaTablePrefix prefix)
{
    requireNonNull(prefix, "prefix is null");
    ImmutableMap.Builder<SchemaTableName, List<ColumnMetadata>> builder = ImmutableMap.builder();
    for (Entry<SchemaTableName, ConnectorTableMetadata> entry : TABLES.entrySet()) {
        if (prefix.matches(entry.getKey())) {
            builder.put(entry.getKey(), entry.getValue().getColumns());
        }
    }
    return builder.build();
}