presto-kinesis/src/main/java/io/prestosql/plugin/kinesis/KinesisMetadata.java/KinesisMetadata/listTableColumns
class KinesisMetadata:    @Override
    public Map<SchemaTableName, List<ColumnMetadata>> listTableColumns(ConnectorSession session, SchemaTablePrefix prefix)
    {
        requireNonNull(prefix, "prefix is null");
        ImmutableMap.Builder<SchemaTableName, List<ColumnMetadata>> columns = ImmutableMap.builder();

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

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