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

public Optional<List<ColumnMetadata>> getTableColMetadata(String connectorId, String dbName, String tblName)
{
    log.debug("Get list of column metadata of table " + formName(dbName, tblName));
    List<ColumnMetadata> colMetadatas = new ArrayList<>();
    MetaProto.StringListType dataTypeList = metaClient.listColumnsDataType(dbName, tblName);
    MetaProto.StringListType colNameList = metaClient.listColumns(dbName, tblName);
    if (dataTypeList.getIsEmpty() || colNameList.getIsEmpty()) {
        log.warn("No col matches!");
        return Optional.empty();
    }
    for (int i = 0; i < dataTypeList.getStrCount(); i++) {
        String dataType = dataTypeList.getStr(i);
        Type type = getType(dataType);
        ColumnMetadata metadata = new ColumnMetadata(
                colNameList.getStr(i),
                type,
                "",
                false);
        colMetadatas.add(metadata);
    }
    return Optional.of(colMetadatas);
}

@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");
    KuduClient kuduClient = kuduClientManager.getClient();

    ImmutableMap.Builder<SchemaTableName, List<ColumnMetadata>> columns = ImmutableMap.builder();
    for (SchemaTableName tableName : listTables(session, prefix)) {
        KuduTableHandle tableHandle = kuduTables.getTables(kuduClient).get(tableName);
        if (tableHandle != null) {
            columns.put(tableName, kuduTables.getColumns(kuduClient, tableHandle));
        }
    }
    kuduClientManager.close(kuduClient);
    return columns.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 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()));
}

@SuppressWarnings("ValueOfIncrementOrDecrementUsed")
private ConnectorTableMetadata getTableMetadata(SchemaTableName schemaTableName)
{
    RedisTableDescription table = getDefinedTables().get(schemaTableName);
    if (table == null) {
        throw new TableNotFoundException(schemaTableName);
    }

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

    appendFields(builder, table.getKey());
    appendFields(builder, table.getValue());

    for (RedisInternalFieldDescription fieldDescription : internalFieldDescriptions) {
        builder.add(fieldDescription.getColumnMetadata(hideInternalColumns));
    }

    return new ConnectorTableMetadata(schemaTableName, builder.build());
}

@Override
public Map<SchemaTableName, List<ColumnMetadata>> listTableColumns(ConnectorSession session, SchemaTablePrefix prefix)
{
    requireNonNull(prefix, "prefix is null");
    if (prefix.getSchemaName() != null && !prefix.getSchemaName().equals(SCHEMA_NAME)) {
        return ImmutableMap.of();
    }

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

    List<SchemaTableName> tableNames;
    if (prefix.getTableName() == null) {
        tableNames = listTables(session, prefix.getSchemaName());
    }
    else {
        tableNames = ImmutableList.of(new SchemaTableName(prefix.getSchemaName(), prefix.getTableName()));
    }

    for (SchemaTableName tableName : tableNames) {
        JmxTableHandle tableHandle = getTableHandle(session, tableName);
        columns.put(tableName, tableHandle.getTableMetadata().getColumns());
    }
    return columns.build();
}

@Override
public void addColumn(ConnectorSession session, ConnectorTableHandle tableHandle, ColumnMetadata column)
{
    if (!allowAddColumn) {
        throw new PrestoException(PERMISSION_DENIED, "Adding Columns is disabled in this Hive catalog");
    }

    HiveTableHandle handle = checkType(tableHandle, HiveTableHandle.class, "tableHandle");
    Optional<Table> tableMetadata = metastore.getTable(handle.getSchemaName(), handle.getTableName());
    if (!tableMetadata.isPresent()) {
        throw new TableNotFoundException(handle.getSchemaTableName());
    }
    Table table = tableMetadata.get();
    StorageDescriptor sd = table.getSd();

    ImmutableList.Builder<FieldSchema> columns = ImmutableList.builder();
    columns.addAll(sd.getCols());
    columns.add(new FieldSchema(column.getName(), toHiveType(column.getType()).getHiveTypeName(), column.getComment()));
    sd.setCols(columns.build());

    table.setSd(sd);
    metastore.alterTable(handle.getSchemaName(), handle.getTableName(), table);
}

@Override
public Map<SchemaTableName, List<ColumnMetadata>> listTableColumns(ConnectorSession session, SchemaTablePrefix prefix)
{
    ImmutableMap.Builder<SchemaTableName, List<ColumnMetadata>> columns = ImmutableMap.builder();
    for (SchemaTableName tableName : listTables(session, prefix.getSchemaName())) {
        try {
            JdbcTableHandle tableHandle = jdbcClient.getTableHandle(tableName);
            if (tableHandle == null) {
                continue;
            }
            columns.put(tableName, getTableMetadata(session, tableHandle).getColumns());
        }
        catch (TableNotFoundException e) {
            // table disappeared during listing operation
        }
    }
    return columns.build();
}

@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 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 InternalTable buildColumns(Session session, String catalogName, Map<String, NullableValue> filters)
{
    InternalTable.Builder table = InternalTable.builder(informationSchemaTableColumns(TABLE_COLUMNS));
    for (Entry<QualifiedObjectName, List<ColumnMetadata>> entry : getColumnsList(session, catalogName, filters).entrySet()) {
        QualifiedObjectName tableName = entry.getKey();
        int ordinalPosition = 1;
        for (ColumnMetadata column : entry.getValue()) {
            if (column.isHidden()) {
                continue;
            }
            table.add(
                    tableName.getCatalogName(),
                    tableName.getSchemaName(),
                    tableName.getObjectName(),
                    column.getName(),
                    ordinalPosition,
                    null,
                    "YES",
                    column.getType().getDisplayName(),
                    column.getComment());
            ordinalPosition++;
        }
    }
    return table.build();
}

@Override
public RecordCursor cursor(ConnectorTransactionHandle transactionHandle, ConnectorSession connectorSession, TupleDomain<Integer> constraint)
{
    GlobalSystemTransactionHandle transaction = checkType(transactionHandle, GlobalSystemTransactionHandle.class, "transaction");
    Session session = toSession(transaction.getTransactionId(), connectorSession);
    Optional<String> catalogFilter = stringFilter(constraint, 0);
    Optional<String> schemaFilter = stringFilter(constraint, 1);
    Optional<String> tableFilter = stringFilter(constraint, 2);

    Builder table = InMemoryRecordSet.builder(METADATA);
    for (String catalog : filter(metadata.getCatalogNames().keySet(), catalogFilter)) {
        QualifiedTablePrefix prefix = FilterUtil.tablePrefix(catalog, schemaFilter, tableFilter);
        for (Entry<QualifiedObjectName, List<ColumnMetadata>> entry : metadata.listTableColumns(session, prefix).entrySet()) {
            addColumnRows(table, entry.getKey(), entry.getValue());
        }
    }
    return table.build().cursor();
}

private static RecordSet toRecordSet(ConnectorTransactionHandle sourceTransaction, SystemTable table, ConnectorSession session, TupleDomain<Integer> constraint)
{
    return new RecordSet()
    {
        private final List<Type> types = table.getTableMetadata().getColumns().stream()
                .map(ColumnMetadata::getType)
                .collect(toImmutableList());

        @Override
        public List<Type> getColumnTypes()
        {
            return types;
        }

        @Override
        public RecordCursor cursor()
        {
            return table.cursor(sourceTransaction, session, constraint);
        }
    };
}

@Override
public Map<SchemaTableName, List<ColumnMetadata>> listTableColumns(ConnectorSession session, SchemaTablePrefix prefix)
{
    checkNotNull(prefix, "prefix is null");
    log.debug("Called listTableColumns on %s.%s", prefix.getSchemaName(), prefix.getTableName());

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

    // NOTE: prefix.getTableName or prefix.getSchemaName can be null
    List<SchemaTableName> tableNames;
    if (prefix.getSchemaName() != null && prefix.getTableName() != null) {
        tableNames = ImmutableList.of(new SchemaTableName(prefix.getSchemaName(), prefix.getTableName()));
    }
    else {
        tableNames = listTables(session, null);
    }

    for (SchemaTableName tableName : tableNames) {
        ConnectorTableMetadata tableMetadata = getTableMetadata(tableName);
        if (tableMetadata != null) {
            columns.put(tableName, tableMetadata.getColumns());
        }
    }
    return columns.build();
}

@JsonCreator
public HDFSTable(
        @JsonProperty("table") HDFSTableHandle table,
        @JsonProperty("tableLayout") HDFSTableLayoutHandle tableLayout,
        @JsonProperty("columns") List<HDFSColumnHandle> columns,
        @JsonProperty("columnMetadatas") List<ColumnMetadata> columnMetadatas)
{
    this.table = requireNonNull(table, "table is null");
    this.tableLayout = requireNonNull(tableLayout, "tableLayout is null");
    this.columns = requireNonNull(columns, "columns is null");
    this.columnMetadatas = requireNonNull(columnMetadatas, "columnMetadas is null");
}

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

/**
 * Gets the metadata for all columns that match the specified table prefix.
 *
 * @param session session
 * @param prefix prefix
 */
@Override
public Map<SchemaTableName, List<ColumnMetadata>> listTableColumns(ConnectorSession session, SchemaTablePrefix prefix)
{
    Map<SchemaTableName, List<ColumnMetadata>> tableColumns = new HashMap<>();
    List<SchemaTableName> tableNames = metaDataQuery.listTables(prefix);
    for (SchemaTableName table : tableNames) {
        List<ColumnMetadata> columnMetadatas = metaDataQuery.getTableColMetadata(connectorId, table.getSchemaName(),
                table.getTableName()).orElse(new ArrayList<>());
        tableColumns.putIfAbsent(table, columnMetadatas);
    }
    return tableColumns;
}

@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 ColumnMetadata getColumnMetadata(
        ConnectorSession session,
        ConnectorTableHandle tableHandle,
        ColumnHandle columnHandle
) {
    convertTableHandle(tableHandle);
    return convertColumnHandle(columnHandle).getColumnMetadata();
}

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

private static void assertPrimitiveField(Map<String, ColumnMetadata> map, String name, Type type, boolean partitionKey)
{
    assertTrue(map.containsKey(name));
    ColumnMetadata column = map.get(name);
    assertEquals(column.getType(), type, name);
    assertEquals(column.isPartitionKey(), partitionKey, name);
}

@Test
public void testGetTableSchemaOffline()
        throws Exception
{
    ConnectorTableHandle tableHandle = getTableHandle(tableOffline);
    ConnectorTableMetadata tableMetadata = metadata.getTableMetadata(newSession(), tableHandle);
    Map<String, ColumnMetadata> map = uniqueIndex(tableMetadata.getColumns(), ColumnMetadata::getName);

    assertPrimitiveField(map, "t_string", VARCHAR, false);
}

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

@Test
public void testColumnMetadata()
{
    assertEquals(exampleTable.getColumnsMetadata(), ImmutableList.of(
            new ColumnMetadata("a", VARCHAR, false),
            new ColumnMetadata("b", BIGINT, false)));
}

@Test
public void getColumnMetadata()
{
    assertEquals(metadata.getColumnMetadata(SESSION, NUMBERS_TABLE_HANDLE, new ExampleColumnHandle(CONNECTOR_ID, "text", VARCHAR, 0)),
            new ColumnMetadata("text", VARCHAR, false));

    // example connector assumes that the table handle and column handle are
    // properly formed, so it will return a metadata object for any
    // ExampleTableHandle and ExampleColumnHandle passed in.  This is on because
    // it is not possible for the Presto Metadata system to create the handles
    // directly.
}

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

private static void assertValueTypes(MaterializedRow row, List<ColumnMetadata> schema)
{
    for (int columnIndex = 0; columnIndex < schema.size(); columnIndex++) {
        ColumnMetadata column = schema.get(columnIndex);
        Object value = row.getField(columnIndex);
        if (value != null) {
            if (BOOLEAN.equals(column.getType())) {
                assertInstanceOf(value, Boolean.class);
            }
            else if (BIGINT.equals(column.getType())) {
                assertInstanceOf(value, Long.class);
            }
            else if (DOUBLE.equals(column.getType())) {
                assertInstanceOf(value, Double.class);
            }
            else if (VARCHAR.equals(column.getType())) {
                assertInstanceOf(value, String.class);
            }
            else if (VARBINARY.equals(column.getType())) {
                assertInstanceOf(value, SqlVarbinary.class);
            }
            else if (TIMESTAMP.equals(column.getType())) {
                assertInstanceOf(value, SqlTimestamp.class);
            }
            else if (DATE.equals(column.getType())) {
                assertInstanceOf(value, SqlDate.class);
            }
            else if (column.getType() instanceof ArrayType) {
                assertInstanceOf(value, List.class);
            }
            else if (column.getType() instanceof MapType) {
                assertInstanceOf(value, Map.class);
            }
            else {
                fail("Unknown primitive type " + columnIndex);
            }
        }
    }
}

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

    ImmutableListMultimap.Builder<SchemaTableName, ColumnMetadata> columns = ImmutableListMultimap.builder();
    for (TableColumn tableColumn : dao.listTableColumns(prefix.getSchemaName(), prefix.getTableName())) {
        if (tableColumn.getColumnName().equals(SAMPLE_WEIGHT_COLUMN_NAME)) {
            continue;
        }
        ColumnMetadata columnMetadata = new ColumnMetadata(tableColumn.getColumnName(), tableColumn.getDataType(), false);
        columns.put(tableColumn.getTable(), columnMetadata);
    }
    return Multimaps.asMap(columns.build());
}

@Test
public void testGetTableSchemaUnpartitioned()
        throws Exception
{
    ConnectorTableHandle tableHandle = getTableHandle(tableUnpartitioned);
    ConnectorTableMetadata tableMetadata = metadata.getTableMetadata(newSession(), tableHandle);
    Map<String, ColumnMetadata> map = uniqueIndex(tableMetadata.getColumns(), ColumnMetadata::getName);

    assertPrimitiveField(map, "t_string", VARCHAR, false);
    assertPrimitiveField(map, "t_tinyint", BIGINT, false);
}

@Test
public void testListTableColumns()
{
    metadata.createTable(SESSION, getOrdersTable());
    Map<SchemaTableName, List<ColumnMetadata>> columns = metadata.listTableColumns(SESSION, new SchemaTablePrefix());
    assertEquals(columns, ImmutableMap.of(DEFAULT_TEST_ORDERS, getOrdersTable().getColumns()));
}

@Test
public void testGetTableSchemaOfflinePartition()
        throws Exception
{
    ConnectorTableHandle tableHandle = getTableHandle(tableOfflinePartition);
    ConnectorTableMetadata tableMetadata = metadata.getTableMetadata(newSession(), tableHandle);
    Map<String, ColumnMetadata> map = uniqueIndex(tableMetadata.getColumns(), ColumnMetadata::getName);

    assertPrimitiveField(map, "t_string", VARCHAR, false);
}

private static void appendFields(ImmutableList.Builder<ColumnMetadata> builder, RedisTableFieldGroup group)
{
    if (group != null) {
        List<RedisTableFieldDescription> fields = group.getFields();
        if (fields != null) {
            for (RedisTableFieldDescription fieldDescription : fields) {
                builder.add(fieldDescription.getColumnMetadata());
            }
        }
    }
}

@Override
public Map<SchemaTableName, List<ColumnMetadata>> listTableColumns(ConnectorSession session, SchemaTablePrefix prefix)
{
    return tables.values().stream()
            .filter(table -> prefix.matches(table.toSchemaTableName()))
            .collect(toMap(BlackHoleTableHandle::toSchemaTableName, handle -> handle.toTableMetadata().getColumns()));
}