@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,"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<SchemaTableName,tableMetadata.getColumns());
        }
    }
    return columns.build();
}

@Override
public Map<SchemaTableName,List<ColumnMetadata>> columns = ImmutableMap.builder();

    List<SchemaTableName> tableNames;
    if (prefix.getSchemaName() == null) {
        tableNames = listTables(session,null);
    }
    else {
        tableNames = 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,"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 Map<SchemaTableName,SchemaTablePrefix prefix)
{
    ImmutableMap.Builder<SchemaTableName,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();
}

@Override
public Map<QualifiedobjectName,ViewDeFinition> getViews(Session session,QualifiedTablePrefix prefix)
{
    requireNonNull(prefix,"prefix is null");
    SchemaTablePrefix tablePrefix = prefix.asSchemaTablePrefix();

    Map<QualifiedobjectName,ViewDeFinition> views = new LinkedHashMap<>();
    for (ConnectorEntry connectorEntry : allConnectorsFor(prefix.getCatalogName())) {
        ConnectorMetadata Metadata = connectorEntry.getMetadata(session);
        ConnectorSession connectorSession = session.toConnectorSession(connectorEntry.getCatalog());
        for (Entry<SchemaTableName,ConnectorViewDeFinition> entry : Metadata.getViews(connectorSession,tablePrefix).entrySet()) {
            QualifiedobjectName viewName = new QualifiedobjectName(
                    prefix.getCatalogName(),entry.getKey().getSchemaName(),entry.getKey().getTableName());
            views.put(viewName,deserializeView(entry.getValue().getViewData()));
        }
    }
    return ImmutableMap.copyOf(views);
}

@Override
public Map<SchemaTableName,ConnectorViewDeFinition> getViews(ConnectorSession session,ConnectorViewDeFinition> views = ImmutableMap.builder();
    List<SchemaTableName> tableNames;
    if (prefix.getTableName() != null) {
        tableNames = ImmutableList.of(new SchemaTableName(prefix.getSchemaName(),prefix.getTableName()));
    }
    else {
        tableNames = listViews(session,prefix.getSchemaName());
    }

    for (SchemaTableName schemaTableName : tableNames) {
        Optional<Table> table = metastore.getTable(schemaTableName.getSchemaName(),schemaTableName.getTableName());
        if (table.isPresent() && HiveUtil.isPrestoView(table.get())) {
            views.put(schemaTableName,new ConnectorViewDeFinition(
                    schemaTableName,Optional.ofNullable(table.get().getowner()),decodeViewData(table.get().getViewOriginalText())));
        }
    }

    return views.build();
}

@Override
public Map<SchemaTableName,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();
}

public List<SchemaTableName> listTables(SchemaTablePrefix prefix)
{
    log.info("List all tables with prefix " + prefix.toString());
    List<SchemaTableName> tables = new ArrayList<>();
    String dbPrefix = prefix.getSchemaName();
    log.debug("listTables dbPrefix: " + dbPrefix);
    String tblPrefix = prefix.getTableName();
    log.debug("listTables tblPrefix: " + tblPrefix);

    // if dbPrefix not mean to match all
    String tblName;
    String dbname;
    if (dbPrefix != null) {
        if (tblPrefix != null) {
            tblName = tblPrefix;
            dbname = dbPrefix;
        }
        else {
            MetaProto.StringListType stringListType = MetaClient.listTables(dbPrefix);
            log.info("record size: " + stringListType.getStrCount());
            if (stringListType.getStrCount() == 0) {
                return tables;
            }
            for (int i = 0; i < stringListType.getStrCount(); i++) {
                tblName = stringListType.getStr(0);
                dbname = dbPrefix;
                log.debug("listTables tableName: " + formName(dbname,tblName));
                tables.add(new SchemaTableName(dbname,tblName));
            }
        }
    }
    return tables;
}

/**
 * List table names,possibly filtered by schema. An empty list is returned if none match.
 *
 * @param session session
 * @param schemaNameOrNull schema name
 */
@Override
public List<SchemaTableName> listTables(ConnectorSession session,String schemaNameOrNull)
{
    if (schemaNameOrNull == null) {
        return new ArrayList<>();
    }
    return MetaDataQuery.listTables(new SchemaTablePrefix(schemaNameOrNull));
}

/**
 * Gets the Metadata for all columns that match the specified table prefix.
 *
 * @param session session
 * @param prefix prefix
 */
@Override
public Map<SchemaTableName,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;
}

private List<SchemaTableName> listTables(ConnectorSession session,SchemaTablePrefix prefix)
{
    if (prefix.getSchemaName() == null) {
        return listTables(session,prefix.getSchemaName());
    }
    return ImmutableList.of(new SchemaTableName(prefix.getSchemaName(),prefix.getTableName()));
}

@Override
public Map<SchemaTableName,"prefix is null");
    ImmutableMap.Builder<SchemaTableName,prefix)) {
        ConnectorTableMetadata tableMetadata = getTableMetadata(tableName);
        // table can disappear during listing operation
        if (tableMetadata != null) {
            columns.put(tableName,tableMetadata.getColumns());
        }
    }
    return columns.build();
}

private List<SchemaTableName> listTables(ConnectorSession session,prefix.getTableName()));
}

@Override
public Map<SchemaTableName,"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());
}

@Override
public Map<SchemaTableName,ConnectorViewDeFinition> map = ImmutableMap.builder();
    for (ViewResult view : dao.getViews(prefix.getSchemaName(),prefix.getTableName())) {
        map.put(view.getName(),new ConnectorViewDeFinition(view.getName(),Optional.empty(),view.getData()));
    }
    return map.build();
}

@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 testListTableColumnsFiltering()
{
    Metadata.createTable(SESSION,List<ColumnMetadata>> filterCatalog = Metadata.listTableColumns(SESSION,new SchemaTablePrefix());
    Map<SchemaTableName,List<ColumnMetadata>> filterSchema = Metadata.listTableColumns(SESSION,new SchemaTablePrefix("test"));
    Map<SchemaTableName,List<ColumnMetadata>> filterTable = Metadata.listTableColumns(SESSION,new SchemaTablePrefix("test","orders"));
    assertEquals(filterCatalog,filterSchema);
    assertEquals(filterCatalog,filterTable);
}

@Test
public void testViews()
{
    SchemaTableName test1 = new SchemaTableName("test","test_view1");
    SchemaTableName test2 = new SchemaTableName("test","test_view2");

    // create views
    Metadata.createView(SESSION,test1,"test1",false);
    Metadata.createView(SESSION,test2,"test2",false);

    // verify listing
    List<SchemaTableName> list = Metadata.listViews(SESSION,"test");
    assertEqualsIgnoreOrder(list,ImmutableList.of(test1,test2));

    // verify getting data
    Map<SchemaTableName,ConnectorViewDeFinition> views = Metadata.getViews(SESSION,new SchemaTablePrefix("test"));
    assertEquals(views.keySet(),ImmutableSet.of(test1,test2));
    assertEquals(views.get(test1).getViewData(),"test1");
    assertEquals(views.get(test2).getViewData(),"test2");

    // drop first view
    Metadata.dropView(SESSION,test1);

    views = Metadata.getViews(SESSION,ImmutableSet.of(test2));

    // drop second view
    Metadata.dropView(SESSION,test2);

    views = Metadata.getViews(SESSION,new SchemaTablePrefix("test"));
    assertTrue(views.isEmpty());

    // verify listing everything
    views = Metadata.getViews(SESSION,new SchemaTablePrefix());
    assertTrue(views.isEmpty());
}

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

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

@Override
public Map<SchemaTableName,ConnectorTableMetadata> entry : tables.entrySet()) {
        if (prefix.matches(entry.getKey())) {
            builder.put(entry.getKey(),entry.getValue().getColumns());
        }
    }
    return builder.build();
}

public SchemaTablePrefix asSchemaTablePrefix()
{
    if (!schemaName.isPresent()) {
        return new SchemaTablePrefix();
    }
    else if (!tableName.isPresent()) {
        return new SchemaTablePrefix(schemaName.get());
    }
    else {
        return new SchemaTablePrefix(schemaName.get(),tableName.get());
    }
}

@Override
public Map<SchemaTableName,List<ColumnMetadata>> tableColumns = ImmutableMap.builder();
    for (SchemaTableName tableName : listTables(session,prefix.getSchemaName())) {
        ImmutableList.Builder<ColumnMetadata> columns = ImmutableList.builder();
        for (ColumnMetadata column : tables.get(tableName).getColumns()) {
            columns.add(new ColumnMetadata(column.getName(),column.getType(),false));
        }
        tableColumns.put(tableName,columns.build());
    }
    return tableColumns.build();
}

@Override
public Map<SchemaTableName,ConnectorViewDeFinition> map = ImmutableMap.builder();
    for (Map.Entry<SchemaTableName,String> entry : views.entrySet()) {
        if (prefix.matches(entry.getKey())) {
            map.put(entry.getKey(),new ConnectorViewDeFinition(entry.getKey(),entry.getValue()));
        }
    }
    return map.build();
}

@Override
public Map<SchemaTableName,List<ColumnMetadata>> tableColumns = ImmutableMap.builder();
    for (String schemaName : getSchemaNames(session,prefix.getSchemaName())) {
        for (TpchTable<?> tpchTable : TpchTable.getTables()) {
            if (prefix.getTableName() == null || tpchTable.getTableName().equals(prefix.getTableName())) {
                ConnectorTableMetadata tableMetadata = getTableMetadata(schemaName,tpchTable);
                tableColumns.put(new SchemaTableName(schemaName,tpchTable.getTableName()),tableMetadata.getColumns());
            }
        }
    }
    return tableColumns.build();
}

@Override
public Map<SchemaTableName,SchemaTablePrefix prefix)
{
    try (threadcontextClassLoader ignored = new threadcontextClassLoader(classLoader)) {
        return delegate.listTableColumns(session,prefix);
    }
}

@Override
public Map<SchemaTableName,SchemaTablePrefix prefix)
{
    try (threadcontextClassLoader ignored = new threadcontextClassLoader(classLoader)) {
        return delegate.getViews(session,prefix);
    }
}

@Override
public Map<SchemaTableName,prefix)) {
        try {
            columns.put(tableName,tableName).getColumns());
        }
        catch (NotFoundException e) {
            // table disappeared during listing operation
        }
    }
    return columns.build();
}

private List<SchemaTableName> listTables(ConnectorSession session,prefix.getTableName()));
}

@Test
public void testListUnkNownSchema()
{
    assertNull(Metadata.getTableHandle(SESSION,new SchemaTableName("totally_invalid_database_name","dual")));
    assertEquals(Metadata.listTables(SESSION,"totally_invalid_database_name"),ImmutableList.of());
    assertEquals(Metadata.listTableColumns(SESSION,new SchemaTablePrefix("totally_invalid_database_name","dual")),ImmutableMap.of());
}

private List<SchemaTableName> listTables(ConnectorSession session,SchemaTablePrefix prefix)
{
    if (prefix.getSchemaName() == null || prefix.getTableName() == null) {
        return listTables(session,prefix.getTableName()));
}

@Test
public void testGetAllTableColumns()
{
    Map<SchemaTableName,List<ColumnMetadata>> allColumns = Metadata.listTableColumns(newSession(),new SchemaTablePrefix());
    assertTrue(allColumns.containsKey(tablePartitionFormat));
    assertTrue(allColumns.containsKey(tableunpartitioned));
}

@Test
public void testGetAllTableColumnsInSchema()
{
    Map<SchemaTableName,new SchemaTablePrefix(database));
    assertTrue(allColumns.containsKey(tablePartitionFormat));
    assertTrue(allColumns.containsKey(tableunpartitioned));
}

@Test
public void testListUnkNownSchema()
{
    ConnectorSession session = newSession();
    assertNull(Metadata.getTableHandle(session,new SchemaTableName(INVALID_DATABASE,INVALID_TABLE)));
    assertEquals(Metadata.listTables(session,INVALID_DATABASE),ImmutableList.of());
    assertEquals(Metadata.listTableColumns(session,new SchemaTablePrefix(INVALID_DATABASE,INVALID_TABLE)),ImmutableMap.of());
    assertEquals(Metadata.listViews(session,ImmutableList.of());
    assertEquals(Metadata.getViews(session,ImmutableMap.of());
}

@Override
public Map<SchemaTableName,List<ColumnMetadata>> listTableColumns(ConnectorSession connectorSession,SchemaTablePrefix schemaTablePrefix)
{
    return null;
}

@Override
public Map<SchemaTableName,SchemaTablePrefix prefix)
{
    return emptyMap();
}

@Override
public Map<SchemaTableName,SchemaTablePrefix prefix)
{
    return Metadata.listTableColumns(session,prefix);
}

@Override
public Map<SchemaTableName,SchemaTablePrefix prefix)
{
    return Metadata.getViews(session,prefix);
}

@Test
public void testCompactDictionaryBlocks()
        throws Exception
{
    Slice[] expectedValues = createExpectedValues(10);
    BlockBuilder blockBuilder = BIGINT.createBlockBuilder(new BlockBuilderStatus(),expectedValues.length);
    for (Slice expectedValue : expectedValues) {
        BIGINT.writeLong(blockBuilder,expectedValue.length());
    }
    Block lengthsDictionary = blockBuilder.build();

    // Create 2 dictionary blocks with the same source id
    DictionaryId commonSourceId = randomDictionaryId();
    DictionaryBlock commonSourceIdBlock1 = createDictionaryBlock(expectedValues,100,commonSourceId);
    DictionaryBlock commonSourceIdBlock2 = new DictionaryBlock(commonSourceIdBlock1.getPositionCount(),lengthsDictionary,commonSourceIdBlock1.getIds(),commonSourceId);

    // Create block with a different source id
    DictionaryBlock randomSourceIdBlock = createDictionaryBlock(expectedValues,randomDictionaryId());

    Page page = new Page(commonSourceIdBlock1,randomSourceIdBlock,commonSourceIdBlock2);
    page.compact();

    // Blocks that had the same source id before compacting page should have the same source id after compacting page
    assertNotEquals(((DictionaryBlock) page.getBlock(0)).getDictionarySourceId(),((DictionaryBlock) page.getBlock(1)).getDictionarySourceId());
    assertEquals(((DictionaryBlock) page.getBlock(0)).getDictionarySourceId(),((DictionaryBlock) page.getBlock(2)).getDictionarySourceId());
}

public static Block createTestBlock()
{
    BlockBuilder blockBuilder = INTERVAL_DAY_TIME.createBlockBuilder(new BlockBuilderStatus(),15);
    INTERVAL_DAY_TIME.writeLong(blockBuilder,1111);
    INTERVAL_DAY_TIME.writeLong(blockBuilder,2222);
    INTERVAL_DAY_TIME.writeLong(blockBuilder,3333);
    INTERVAL_DAY_TIME.writeLong(blockBuilder,4444);
    return blockBuilder.build();
}

@Test
public void testPrimitiveLongSerialization()
{
    StateCompiler compiler = new StateCompiler();

    AccumulatorStateFactory<LongState> factory = compiler.generateStateFactory(LongState.class);
    AccumulatorStateSerializer<LongState> serializer = compiler.generateStateSerializer(LongState.class);
    LongState state = factory.createSingleState();
    LongState deserializedState = factory.createSingleState();

    state.setLong(2);

    BlockBuilder builder = BIGINT.createBlockBuilder(new BlockBuilderStatus(),1);
    serializer.serialize(state,builder);

    Block block = builder.build();

    assertEquals(BIGINT.getLong(block,0),state.getLong());
    serializer.deserialize(block,deserializedState);
    assertEquals(deserializedState.getLong(),state.getLong());
}

@Test
public void testvarianceStateSerialization()
{
    StateCompiler compiler = new StateCompiler();

    AccumulatorStateFactory<VarianceState> factory = compiler.generateStateFactory(VarianceState.class);
    AccumulatorStateSerializer<VarianceState> serializer = compiler.generateStateSerializer(VarianceState.class);
    VarianceState singleState = factory.createSingleState();
    VarianceState deserializedState = factory.createSingleState();

    singleState.setMean(1);
    singleState.setCount(2);
    singleState.setM2(3);

    BlockBuilder builder = VarcharType.VARCHAR.createBlockBuilder(new BlockBuilderStatus(),1);
    serializer.serialize(singleState,builder);

    Block block = builder.build();
    serializer.deserialize(block,deserializedState);

    assertEquals(deserializedState.getCount(),singleState.getCount());
    assertEquals(deserializedState.getMean(),singleState.getMean());
    assertEquals(deserializedState.getM2(),singleState.getM2());
}

private static Block createChannel(int positionCount,int arraySize,ArrayType arrayType)
{
    BlockBuilder blockBuilder = arrayType.createBlockBuilder(new BlockBuilderStatus(),positionCount);
    for (int position = 0; position < positionCount; position++) {
        BlockBuilder entryBuilder = blockBuilder.beginBlockEntry();
        for (int i = 0; i < arraySize; i++) {
            if (arrayType.getElementType().getJavaType() == long.class) {
                arrayType.getElementType().writeLong(entryBuilder,ThreadLocalRandom.current().nextLong());
            }
            else if (arrayType.getElementType().equals(VARCHAR)) {
                arrayType.getElementType().writeSlice(entryBuilder,Slices.utf8Slice("test_string"));
            }
            else {
                throw new UnsupportedOperationException();
            }
        }
        blockBuilder.closeEntry();
    }
    return blockBuilder.build();
}

@Test
public void testPrimitiveBooleanSerialization()
{
    StateCompiler compiler = new StateCompiler();

    AccumulatorStateFactory<BooleanState> factory = compiler.generateStateFactory(BooleanState.class);
    AccumulatorStateSerializer<BooleanState> serializer = compiler.generateStateSerializer(BooleanState.class);
    BooleanState state = factory.createSingleState();
    BooleanState deserializedState = factory.createSingleState();

    state.setBoolean(true);

    BlockBuilder builder = BooleanType.BOOLEAN.createBlockBuilder(new BlockBuilderStatus(),deserializedState);
    assertEquals(deserializedState.getBoolean(),state.getBoolean());
}

public static Block createStringArraysBlock(Iterable<? extends Iterable<String>> values)
{
    ArrayType arrayType = new ArrayType(VARCHAR);
    BlockBuilder builder = arrayType.createBlockBuilder(new BlockBuilderStatus(),100);

    for (Iterable<String> value : values) {
        if (value == null) {
            builder.appendNull();
        }
        else {
            arrayType.writeObject(builder,createStringsBlock(value));
        }
    }

    return builder.build();
}

public static void output(ArrayType outputType,MinMaxByNState state,BlockBuilder out)
{
    TypedkeyvalueHeap heap = state.getTypedkeyvalueHeap();
    if (heap == null || heap.isEmpty()) {
        out.appendNull();
        return;
    }

    Type elementType = outputType.getElementType();

    BlockBuilder arrayBlockBuilder = out.beginBlockEntry();
    BlockBuilder reversedBlockBuilder = elementType.createBlockBuilder(new BlockBuilderStatus(),heap.getCapacity());
    long startSize = heap.getEstimatedSize();
    heap.popAll(reversedBlockBuilder);
    state.addMemoryUsage(heap.getEstimatedSize() - startSize);

    for (int i = reversedBlockBuilder.getPositionCount() - 1; i >= 0; i--) {
        elementType.appendTo(reversedBlockBuilder,i,arrayBlockBuilder);
    }
    out.closeEntry();
}

public TypedSet(Type elementType,int expectedSize)
{
    checkArgument(expectedSize > 0,"expectedSize must be > 0");
    this.elementType = requireNonNull(elementType,"elementType must not be null");
    this.elementBlock = elementType.createBlockBuilder(new BlockBuilderStatus(),expectedSize);

    int hashSize = arraySize(expectedSize,FILL_RATIO);
    this.maxFill = calculateMaxFill(hashSize);
    this.hashMask = hashSize - 1;

    blockPositionByHash.ensureCapacity(hashSize);
    for (int i = 0; i < hashSize; i++) {
        blockPositionByHash.set(i,EMPTY_SLOT);
    }

    this.containsNullElement = false;
}

public static Block createTestBlock()
{
    BlockBuilder blockBuilder = INTERVAL_YEAR_MONTH.createBlockBuilder(new BlockBuilderStatus(),15);
    INTERVAL_YEAR_MONTH.writeLong(blockBuilder,1111);
    INTERVAL_YEAR_MONTH.writeLong(blockBuilder,2222);
    INTERVAL_YEAR_MONTH.writeLong(blockBuilder,3333);
    INTERVAL_YEAR_MONTH.writeLong(blockBuilder,4444);
    return blockBuilder.build();
}

public Block markdistinctRows(Page page)
{
    GroupByIdBlock ids = groupByHash.getGroupIds(page);
    BlockBuilder blockBuilder = BOOLEAN.createBlockBuilder(new BlockBuilderStatus(),ids.getPositionCount());
    for (int i = 0; i < ids.getPositionCount(); i++) {
        if (ids.getGroupId(i) == nextdistinctId) {
            BOOLEAN.writeBoolean(blockBuilder,true);
            nextdistinctId++;
        }
        else {
            BOOLEAN.writeBoolean(blockBuilder,false);
        }
    }

    return blockBuilder.build();
}

@Test
public void testEvaluateClassifierPredictions()
        throws Exception
{
    TypeRegistry typeRegistry = new TypeRegistry();
    typeRegistry.addParametricType(new ClassifierParametricType());
    typeRegistry.addType(RegressorType.REGRESSOR);
    typeRegistry.addType(ModelType.MODEL);
    Metadata.addFunctions(new MLFunctionFactory(typeRegistry).listFunctions());
    InternalAggregationFunction aggregation = Metadata.getFunctionRegistry().getAggregateFunctionImplementation(new Signature("evaluate_classifier_predictions",AGGREGATE,StandardTypes.VARCHAR,StandardTypes.BIGINT,StandardTypes.BIGINT));
    Accumulator accumulator = aggregation.bind(ImmutableList.of(0,1),Optional.empty(),1.0).createAccumulator();
    accumulator.addInput(getPage());
    BlockBuilder finalOut = accumulator.getFinalType().createBlockBuilder(new BlockBuilderStatus(),1);
    accumulator.evaluateFinal(finalOut);
    Block block = finalOut.build();

    String output = VARCHAR.getSlice(block,0).toStringUtf8();
    List<String> parts = ImmutableList.copyOf(Splitter.on('\n').omitEmptyStrings().split(output));
    assertEquals(parts.size(),7,output);
    assertEquals(parts.get(0),"Accuracy: 1/2 (50.00%)");
}

public static Block mapConcat(Type keyType,Type valueType,Block leftMap,Block rightMap)
{
    TypedSet typedSet = new TypedSet(keyType,rightMap.getPositionCount());
    BlockBuilder blockBuilder = new InterleavedBlockBuilder(ImmutableList.of(keyType,valueType),new BlockBuilderStatus(),leftMap.getPositionCount() + rightMap.getPositionCount());
    for (int i = 0; i < rightMap.getPositionCount(); i += 2) {
        typedSet.add(rightMap,i);
        keyType.appendTo(rightMap,blockBuilder);
        valueType.appendTo(rightMap,i + 1,blockBuilder);
    }
    for (int i = 0; i < leftMap.getPositionCount(); i += 2) {
        if (!typedSet.contains(leftMap,i)) {
            keyType.appendTo(leftMap,blockBuilder);
            valueType.appendTo(leftMap,blockBuilder);
        }
    }
    return blockBuilder.build();
}

public static Block createTestBlock()
{
    BlockBuilder blockBuilder = BOOLEAN.createBlockBuilder(new BlockBuilderStatus(),15);
    BOOLEAN.writeBoolean(blockBuilder,true);
    BOOLEAN.writeBoolean(blockBuilder,false);
    BOOLEAN.writeBoolean(blockBuilder,false);
    return blockBuilder.build();
}

@Override
public void start()
{
    if (builder == null) {
        if (nullBuilder == null || (nullBuilder.getPositionCount() >= NULL_BUILDER_POSITIONS_THRESHOLD && nullBuilder.getSizeInBytes() >= NULL_BUILDER_SIZE_IN_BYTES_THRESHOLD)) {
            nullBuilder = rowType.createBlockBuilder(new BlockBuilderStatus(),NULL_BUILDER_POSITIONS_THRESHOLD);
        }
        currentEntryBuilder = nullBuilder.beginBlockEntry();
    }
    else {
        while (builder.getPositionCount() < fieldindex) {
            builder.appendNull();
        }
        currentEntryBuilder = builder.beginBlockEntry();
    }
    for (BlockConverter converter : converters) {
        converter.beforeValue(currentEntryBuilder);
    }
}

public static Block createTestBlock()
{
    BlockBuilder blockBuilder = TIMESTAMP_WITH_TIME_ZONE.createBlockBuilder(new BlockBuilderStatus(),15);
    TIMESTAMP_WITH_TIME_ZONE.writeLong(blockBuilder,packDateTimeWithZone(1111,getTimeZoneKeyForOffset(0)));
    TIMESTAMP_WITH_TIME_ZONE.writeLong(blockBuilder,getTimeZoneKeyForOffset(1)));
    TIMESTAMP_WITH_TIME_ZONE.writeLong(blockBuilder,getTimeZoneKeyForOffset(2)));
    TIMESTAMP_WITH_TIME_ZONE.writeLong(blockBuilder,packDateTimeWithZone(2222,getTimeZoneKeyForOffset(3)));
    TIMESTAMP_WITH_TIME_ZONE.writeLong(blockBuilder,getTimeZoneKeyForOffset(4)));
    TIMESTAMP_WITH_TIME_ZONE.writeLong(blockBuilder,getTimeZoneKeyForOffset(5)));
    TIMESTAMP_WITH_TIME_ZONE.writeLong(blockBuilder,getTimeZoneKeyForOffset(6)));
    TIMESTAMP_WITH_TIME_ZONE.writeLong(blockBuilder,getTimeZoneKeyForOffset(7)));
    TIMESTAMP_WITH_TIME_ZONE.writeLong(blockBuilder,packDateTimeWithZone(3333,getTimeZoneKeyForOffset(8)));
    TIMESTAMP_WITH_TIME_ZONE.writeLong(blockBuilder,getTimeZoneKeyForOffset(9)));
    TIMESTAMP_WITH_TIME_ZONE.writeLong(blockBuilder,packDateTimeWithZone(4444,getTimeZoneKeyForOffset(10)));
    return blockBuilder.build();
}

public static Block createArrayBigintBlock(Iterable<? extends Iterable<Long>> values)
{
    ArrayType arrayType = new ArrayType(BIGINT);
    BlockBuilder builder = arrayType.createBlockBuilder(new BlockBuilderStatus(),100);

    for (Iterable<Long> value : values) {
        if (value == null) {
            builder.appendNull();
        }
        else {
            arrayType.writeObject(builder,createLongsBlock(value));
        }
    }

    return builder.build();
}

public static Block createTestBlock()
{
    BlockBuilder blockBuilder = DOUBLE.createBlockBuilder(new BlockBuilderStatus(),15);
    DOUBLE.writeDouble(blockBuilder,11.11);
    DOUBLE.writeDouble(blockBuilder,22.22);
    DOUBLE.writeDouble(blockBuilder,33.33);
    DOUBLE.writeDouble(blockBuilder,44.44);
    return blockBuilder.build();
}

public static Block createTestBlock()
{
    BlockBuilder blockBuilder = COLOR.createBlockBuilder(new BlockBuilderStatus(),15);
    COLOR.writeLong(blockBuilder,rgb(1,1,1));
    COLOR.writeLong(blockBuilder,rgb(2,2,2));
    COLOR.writeLong(blockBuilder,rgb(3,3,3));
    COLOR.writeLong(blockBuilder,rgb(4,4,4));
    return blockBuilder.build();
}

@Test
public void testBigintSerializedSize()
{
    BlockBuilder builder = BIGINT.createBlockBuilder(new BlockBuilderStatus(),5);

    // empty page
    Page page = new Page(builder.build());
    int pageSize = serializedSize(ImmutableList.of(BIGINT),page);
    assertEquals(pageSize,35); // page overhead

    // page with one value
    BIGINT.writeLong(builder,123);
    page = new Page(builder.build());
    int firstValueSize = serializedSize(ImmutableList.of(BIGINT),page) - pageSize;
    assertEquals(firstValueSize,9); // value size + value overhead

    // page with two values
    BIGINT.writeLong(builder,456);
    page = new Page(builder.build());
    int secondValueSize = serializedSize(ImmutableList.of(BIGINT),page) - (pageSize + firstValueSize);
    assertEquals(secondValueSize,8); // value size (value overhead is shared with prevIoUs value)
}

public static Block createTestBlock()
{
    BlockBuilder blockBuilder = TIME_WITH_TIME_ZONE.createBlockBuilder(new BlockBuilderStatus(),15);
    TIME_WITH_TIME_ZONE.writeLong(blockBuilder,getTimeZoneKeyForOffset(0)));
    TIME_WITH_TIME_ZONE.writeLong(blockBuilder,getTimeZoneKeyForOffset(1)));
    TIME_WITH_TIME_ZONE.writeLong(blockBuilder,getTimeZoneKeyForOffset(2)));
    TIME_WITH_TIME_ZONE.writeLong(blockBuilder,getTimeZoneKeyForOffset(3)));
    TIME_WITH_TIME_ZONE.writeLong(blockBuilder,getTimeZoneKeyForOffset(4)));
    TIME_WITH_TIME_ZONE.writeLong(blockBuilder,getTimeZoneKeyForOffset(5)));
    TIME_WITH_TIME_ZONE.writeLong(blockBuilder,getTimeZoneKeyForOffset(6)));
    TIME_WITH_TIME_ZONE.writeLong(blockBuilder,getTimeZoneKeyForOffset(7)));
    TIME_WITH_TIME_ZONE.writeLong(blockBuilder,getTimeZoneKeyForOffset(8)));
    TIME_WITH_TIME_ZONE.writeLong(blockBuilder,getTimeZoneKeyForOffset(9)));
    TIME_WITH_TIME_ZONE.writeLong(blockBuilder,getTimeZoneKeyForOffset(10)));
    return blockBuilder.build();
}

private static Object mapKeyToObject(ConnectorSession session,String jsonKey,Type type)
{
    BlockBuilder blockBuilder;
    if (type instanceof FixedWidthType) {
        blockBuilder = type.createBlockBuilder(new BlockBuilderStatus(),1);
    }
    else {
        blockBuilder = type.createBlockBuilder(new BlockBuilderStatus(),jsonKey.length());
    }
    if (type.getJavaType() == boolean.class) {
        type.writeBoolean(blockBuilder,Boolean.parseBoolean(jsonKey));
    }
    else if (type.getJavaType() == long.class) {
        type.writeLong(blockBuilder,Long.parseLong(jsonKey));
    }
    else if (type.getJavaType() == double.class) {
        type.writeDouble(blockBuilder,Double.parseDouble(jsonKey));
    }
    else if (type.getJavaType() == Slice.class) {
        type.writeSlice(blockBuilder,Slices.utf8Slice(jsonKey));
    }
    return type.getobjectValue(session,blockBuilder.build(),0);
}

private Block toBlock(Object value)
{
    BlockBuilder blockBuilder = type.createBlockBuilder(new BlockBuilderStatus(),1);
    Class<?> javaType = type.getJavaType();
    if (value == null) {
        blockBuilder.appendNull();
    }
    else if (javaType == boolean.class) {
        type.writeBoolean(blockBuilder,(Boolean) value);
    }
    else if (javaType == long.class) {
        type.writeLong(blockBuilder,(Long) value);
    }
    else if (javaType == double.class) {
        type.writeDouble(blockBuilder,(Double) value);
    }
    else if (javaType == Slice.class) {
        Slice slice = (Slice) value;
        type.writeSlice(blockBuilder,slice,slice.length());
    }
    else {
        type.writeObject(blockBuilder,value);
    }
    return blockBuilder.build();
}

@Test
public void testRowHistograms()
        throws Exception
{
    RowType innerRowType = new RowType(ImmutableList.of(BIGINT,DOUBLE),Optional.of(ImmutableList.of("f1","f2")));
    MapType mapType = new MapType(innerRowType,BIGINT);
    InternalAggregationFunction aggregationFunction = Metadata.getFunctionRegistry().getAggregateFunctionImplementation(new Signature(NAME,mapType.getTypeSignature().toString(),innerRowType.getTypeSignature().toString()));

    BlockBuilder builder = innerRowType.createBlockBuilder(new BlockBuilderStatus(),3);
    innerRowType.writeObject(builder,testRowBigintBigint(1L,1.0));
    innerRowType.writeObject(builder,testRowBigintBigint(2L,2.0));
    innerRowType.writeObject(builder,testRowBigintBigint(3L,3.0));

    assertAggregation(
            aggregationFunction,1.0,ImmutableMap.of(ImmutableList.of(1L,1.0),1L,ImmutableList.of(2L,2.0),ImmutableList.of(3L,3.0),1L),builder.build());
}

private void test(IntStream keyInputStream,Stream<String> valueInputStream,BlockComparator comparator,Iterator<String> outputIterator)
{
    BlockBuilder keysBlockBuilder = BIGINT.createBlockBuilder(new BlockBuilderStatus(),INPUT_SIZE);
    BlockBuilder valuesBlockBuilder = VARCHAR.createBlockBuilder(new BlockBuilderStatus(),INPUT_SIZE);
    keyInputStream.forEach(x -> BIGINT.writeLong(keysBlockBuilder,x));
    valueInputStream.forEach(x -> VARCHAR.writeString(valuesBlockBuilder,x));

    TypedkeyvalueHeap heap = new TypedkeyvalueHeap(comparator,BIGINT,VARCHAR,OUTPUT_SIZE);
    heap.addAll(keysBlockBuilder,valuesBlockBuilder);

    BlockBuilder resultBlockBuilder = VARCHAR.createBlockBuilder(new BlockBuilderStatus(),OUTPUT_SIZE);
    heap.popAll(resultBlockBuilder);

    Block resultBlock = resultBlockBuilder.build();
    assertEquals(resultBlock.getPositionCount(),OUTPUT_SIZE);
    for (int i = 0; i < OUTPUT_SIZE; i++) {
        assertEquals(VARCHAR.getSlice(resultBlock,i).toStringUtf8(),outputIterator.next());
    }
}

private static Block serializeList(Type type,BlockBuilder builder,Object object) {
    List<?> list = (List) object;
    if (list == null) {
        requireNonNull(builder,"parent builder is null").appendNull();
        return null;
    }

    List<Type> typeParameters = type.getTypeParameters();
    checkArgument(typeParameters.size() == 1,"list must have exactly 1 type parameter");
    Type elementType = typeParameters.get(0);

    BlockBuilder currentBuilder;
    if (builder != null) {
        currentBuilder = builder.beginBlockEntry();
    } else {
        currentBuilder = elementType.createBlockBuilder(new BlockBuilderStatus(),list.size());
    }

    for (Object element : list) {
        serializeObject(elementType,currentBuilder,element);
    }

    if (builder != null) {
        builder.closeEntry();
        return null;
    } else {
        Block resultBlock = currentBuilder.build();
        return resultBlock;
    }
}

private static Block serializeMap(Type type,Object object) {
    Map<?,?> map = (Map) object;
    if (map == null) {
        requireNonNull(builder,"parent builder is null").appendNull();
        return null;
    }

    List<Type> typeParameters = type.getTypeParameters();
    checkArgument(typeParameters.size() == 2,"map must have exactly 2 type parameter");
    Type keyType = typeParameters.get(0);
    Type valueType = typeParameters.get(1);

    BlockBuilder currentBuilder;
    if (builder != null) {
        currentBuilder = builder.beginBlockEntry();
    } else {
        currentBuilder = new InterleavedBlockBuilder(typeParameters,map.size());
    }

    for (Map.Entry<?,?> entry : map.entrySet()) {
        // Hive skips map entries with null keys
        if (entry.getKey() != null) {
            serializeObject(keyType,entry.getKey());
            serializeObject(valueType,entry.getValue());
        }
    }

    if (builder != null) {
        builder.closeEntry();
        return null;
    } else {
        Block resultBlock = currentBuilder.build();
        return resultBlock;
    }
}

@sqlType("array(double)")
public static Block geohash_decode(@sqlType(StandardTypes.VARCHAR) Slice geohash) {

    BlockBuilder blockBuilder = DOUBLE.createBlockBuilder(new BlockBuilderStatus(),2);
    LatLong coordinates = GeoHash.decodeHash(geohash.toStringUtf8());
    DOUBLE.writeDouble(blockBuilder,coordinates.getLat());
    DOUBLE.writeDouble(blockBuilder,coordinates.getLon());
    return blockBuilder.build();
}

private static Block toBlock(double[] poly) {
    FixedWidthBlockBuilder blockBuilder = new FixedWidthBlockBuilder(8,poly.length);
    for (double d : poly) {
        blockBuilder.writeLong(Double.doubletoLongBits(d));
        blockBuilder.closeEntry();
    }
    return blockBuilder.build();
}

private Block createZeroBlock(Type type,int rowsCount,Slice constantSlice)
{
    checkArgument(SUPPORTED_TYPES.contains(type),"Unsupported type [%s]",type);
    BlockBuilder builder;

    if (type instanceof FixedWidthType) {
        builder = type.createBlockBuilder(new BlockBuilderStatus(),rowsCount);
    }
    else {
        builder = type.createBlockBuilder(new BlockBuilderStatus(),rowsCount,constantSlice.length());
    }

    for (int i = 0; i < rowsCount; i++) {
        Class<?> javaType = type.getJavaType();
        if (javaType == boolean.class) {
            type.writeBoolean(builder,false);
        }
        else if (javaType == long.class) {
            type.writeLong(builder,0);
        }
        else if (javaType == double.class) {
            type.writeDouble(builder,0.0);
        }
        else if (javaType == Slice.class) {
            type.writeSlice(builder,constantSlice,constantSlice.length());
        }
        else {
            throw new UnsupportedOperationException("UnkNown javaType: " + javaType.getName());
        }
    }
    return builder.build();
}

public static Block arrayBlockOf(Type elementType,Object... values)
{
    BlockBuilder blockBuilder = elementType.createBlockBuilder(new BlockBuilderStatus(),1024);
    for (Object value : values) {
        appendToBlockBuilder(elementType,value,blockBuilder);
    }
    return blockBuilder.build();
}

public static Block mapBlockOf(Type keyType,Object key,Object value)
{
    BlockBuilder blockBuilder = new InterleavedBlockBuilder(ImmutableList.of(keyType,1024);
    appendToBlockBuilder(keyType,key,blockBuilder);
    appendToBlockBuilder(valueType,blockBuilder);
    return blockBuilder.build();
}

public static Block rowBlockOf(List<Type> parameterTypes,Object... values)
{
    InterleavedBlockBuilder blockBuilder = new InterleavedBlockBuilder(parameterTypes,1024);
    for (int i = 0; i < values.length; i++) {
        appendToBlockBuilder(parameterTypes.get(i),values[i],blockBuilder);
    }
    return blockBuilder.build();
}

private static Block serializeList(Type type,Object object,ListObjectInspector inspector)
{
    List<?> list = inspector.getList(object);
    if (list == null) {
        requireNonNull(builder,"list must have exactly 1 type parameter");
    Type elementType = typeParameters.get(0);
    ObjectInspector elementInspector = inspector.getListElementObjectInspector();
    BlockBuilder currentBuilder;
    if (builder != null) {
        currentBuilder = builder.beginBlockEntry();
    }
    else {
        currentBuilder = elementType.createBlockBuilder(new BlockBuilderStatus(),element,elementInspector);
    }

    if (builder != null) {
        builder.closeEntry();
        return null;
    }
    else {
        Block resultBlock = currentBuilder.build();
        return resultBlock;
    }
}

@Test
public void testStackRepresentation()
        throws Exception
{
    Block array = arrayBlockOf(BIGINT,2L);
    Block actualBlock = mapBlockOf(DOUBLE,new ArrayType(BIGINT),ImmutableMap.of(1.0,array));
    DynamicsliceOutput actualSliceOutput = new DynamicsliceOutput(100);
    writeBlock(actualSliceOutput,actualBlock);

    Block expectedBlock = new InterleavedBlockBuilder(ImmutableList.<Type>of(DOUBLE,new ArrayType(BIGINT)),3)
            .writeDouble(1.0)
            .closeEntry()
            .writeObject(
                    BIGINT
                    .createBlockBuilder(new BlockBuilderStatus(),1)
                    .writeLong(1L)
                    .closeEntry()
                    .writeLong(2L)
                    .closeEntry()
                    .build()
            )
            .closeEntry()
            .build();
    DynamicsliceOutput expectedSliceOutput = new DynamicsliceOutput(100);
    writeBlock(expectedSliceOutput,expectedBlock);

    assertEquals(actualSliceOutput.slice(),expectedSliceOutput.slice());
}

public static Block createTestBlock(Type arrayType)
{
    BlockBuilder blockBuilder = arrayType.createBlockBuilder(new BlockBuilderStatus(),4);
    arrayType.writeObject(blockBuilder,arrayBlockOf(BIGINT,2));
    arrayType.writeObject(blockBuilder,3));
    arrayType.writeObject(blockBuilder,200,300));
    return blockBuilder.build();
}

@Override
public Block[] getSequenceBlocks(int start,int length)
{
    BlockBuilder blockBuilder = VARBINARY.createBlockBuilder(new BlockBuilderStatus(),length);
    for (int i = 0; i < length; i++) {
        VARBINARY.writeSlice(blockBuilder,Slices.wrappedBuffer(Ints.toByteArray(i)));
    }
    return new Block[] {blockBuilder.build()};
}

public static Block mapBlockOf(Type keyType,Map<?,?> value)
{
    BlockBuilder blockBuilder = new InterleavedBlockBuilder(ImmutableList.of(keyType,value.size() * 2);
    for (Map.Entry<?,?> entry : value.entrySet()) {
        appendToBlockBuilder(keyType,entry.getKey(),blockBuilder);
        appendToBlockBuilder(valueType,entry.getValue(),blockBuilder);
    }
    return blockBuilder.build();
}

public static void input(Type type,ArrayAggregationState state,Block value,int position)
{
    BlockBuilder blockBuilder = state.getBlockBuilder();
    if (blockBuilder == null) {
        blockBuilder = type.createBlockBuilder(new BlockBuilderStatus(),100);
        state.setBlockBuilder(blockBuilder);
    }
    long startSize = blockBuilder.getRetainedSizeInBytes();
    type.appendTo(value,position,blockBuilder);
    state.addMemoryUsage(blockBuilder.getRetainedSizeInBytes() - startSize);
}

public keyvaluePairs(Type keyType,boolean isMultiValue)
{
    this.keyType = requireNonNull(keyType,"keyType is null");
    this.valueType = requireNonNull(valueType,"valueType is null");
    this.keySet = new TypedSet(keyType,EXPECTED_ENTRIES);
    keyBlockBuilder = this.keyType.createBlockBuilder(new BlockBuilderStatus(),EXPECTED_ENTRIES,expectedValueSize(keyType,EXPECTED_ENTRY_SIZE));
    valueBlockBuilder = this.valueType.createBlockBuilder(new BlockBuilderStatus(),expectedValueSize(valueType,EXPECTED_ENTRY_SIZE));

    this.isMultiValue = isMultiValue;
}

@Override
public final void load(LazyBlock lazyBlock)
{
    if (loaded) {
        return;
    }

    checkState(batchId == expectedBatchId);

    try {
        Block block = parquetReader.readPrimitive(columnDescriptor,type);
        lazyBlock.setBlock(block);
    }
    catch (IOException e) {
        throw new HdfsCursorException();
    }
    loaded = true;
}

@ScalarFunction("parse_agent")
@Description("Returns Map,which has keys such as 'category','name','os','version','vendor' and 'os_version'")
@sqlType("map<varchar,varchar>")
public Block parseAgent(@TypeParameter("map<varchar,varchar>") Type mapType,@sqlType(StandardTypes.VARCHAR) Slice slice) {
    String argument = slice.toStringUtf8();
    Map<String,String> stringMap = Classifier.parse(argument);

    if (pageBuilder.isFull()) {
        pageBuilder.reset();
    }

    BlockBuilder blockBuilder = pageBuilder.getBlockBuilder(0);
    BlockBuilder singleMapBlockBuilder = blockBuilder.beginBlockEntry();
    for (Map.Entry<String,String> entry : stringMap.entrySet()) {
        VARCHAR.writeSlice(singleMapBlockBuilder,Slices.utf8Slice(entry.getKey()));
        VARCHAR.writeSlice(singleMapBlockBuilder,Slices.utf8Slice(entry.getValue()));
    }
    blockBuilder.closeEntry();
    pageBuilder.declarePosition();

    return (Block) mapType.getobject(blockBuilder,blockBuilder.getPositionCount() - 1);
}

@Test
public void testpoly_contains() throws Exception {


    double[] poly = new double[]{
            45,9.5,45.5,9,46,10,45,10
    };

    Block blockpoly = toBlock(poly);
    assertFalse(polyContains.contains(DoubleType.DOUBLE,blockpoly,6,3));
    assertFalse(polyContains.contains(DoubleType.DOUBLE,9));
    assertTrue(polyContains.contains(DoubleType.DOUBLE,45.7,9.7));

}

@Override
public void addInput(Page page)
{
    requireNonNull(page,"page is null");
    checkState(state == State.RUNNING,"Operator is %s",state);

    Block[] blocks = new Block[inputChannels.size()];
    for (int outputChannel = 0; outputChannel < inputChannels.size(); outputChanneL++) {
        blocks[outputChannel] = page.getBlock(inputChannels.get(outputChannel));
    }
    Block sampleWeightBlock = null;
    if (sampleWeightChannel.isPresent()) {
        sampleWeightBlock = page.getBlock(sampleWeightChannel.get());
    }
    pageSink.appendPage(new Page(blocks),sampleWeightBlock);
    rowCount += page.getPositionCount();
}

@Test
public void testroundtrip()
{
    BlockBuilder expectedBlockBuilder = VARCHAR.createBlockBuilder(new BlockBuilderStatus(),5);
    VARCHAR.writeString(expectedBlockBuilder,"alice");
    VARCHAR.writeString(expectedBlockBuilder,"bob");
    VARCHAR.writeString(expectedBlockBuilder,"charlie");
    VARCHAR.writeString(expectedBlockBuilder,"dave");
    Block expectedBlock = expectedBlockBuilder.build();

    Page expectedPage = new Page(expectedBlock,expectedBlock,expectedBlock);

    DynamicsliceOutput sliceOutput = new DynamicsliceOutput(1024);
    writePages(blockEncodingManager,sliceOutput,expectedPage,expectedPage);

    List<Type> types = ImmutableList.<Type>of(VARCHAR,VARCHAR,VARCHAR);
    Iterator<Page> pageIterator = readPages(blockEncodingManager,sliceOutput.slice().getinput());
    assertPageEquals(types,pageIterator.next(),expectedPage);
    assertPageEquals(types,expectedPage);
    assertFalse(pageIterator.hasNext());
}

public static Block toArray(Type arrayType,ConnectorSession connectorSession,Slice json)
{
    try {
        List<?> array = (List<?>) stackRepresentationToObject(connectorSession,json,arrayType);
        if (array == null) {
            return null;
        }
        Type elementType = ((ArrayType) arrayType).getElementType();
        BlockBuilder blockBuilder = elementType.createBlockBuilder(new BlockBuilderStatus(),array.size());
        for (Object element : array) {
            appendToBlockBuilder(elementType,element,blockBuilder);
        }
        return blockBuilder.build();
    }
    catch (RuntimeException e) {
        throw new PrestoException(INVALID_CAST_ARGUMENT,"Value cannot be cast to " + arrayType,e);
    }
}

private static IntComparator IntBlockCompare(Type type,Block block)
{
    return new AbstractIntComparator()
    {
        @Override
        public int compare(int left,int right)
        {
            if (block.isNull(left) && block.isNull(right)) {
                return 0;
            }
            if (block.isNull(left)) {
                return -1;
            }
            if (block.isNull(right)) {
                return 1;
            }
            return type.compareto(block,left,block,right);
        }
    };
}

private static Page[] offsetColumns(Page[] pages,int offset)
{
    Page[] newPages = new Page[pages.length];
    for (int i = 0; i < pages.length; i++) {
        Page page = pages[i];
        Block[] newBlocks = new Block[page.getChannelCount() + offset];
        for (int channel = 0; channel < offset; channeL++) {
            newBlocks[channel] = createNullRLEBlock(page.getPositionCount());
        }
        for (int channel = 0; channel < page.getBlocks().length; channeL++) {
            newBlocks[channel + offset] = page.getBlocks()[channel];
        }
        newPages[i] = new Page(page.getPositionCount(),newBlocks);
    }
    return newPages;
}

public static Block createTestBlock()
{
    BlockBuilder blockBuilder = TIMESTAMP.createBlockBuilder(new BlockBuilderStatus(),15);
    TIMESTAMP.writeLong(blockBuilder,1111);
    TIMESTAMP.writeLong(blockBuilder,2222);
    TIMESTAMP.writeLong(blockBuilder,3333);
    TIMESTAMP.writeLong(blockBuilder,4444);
    return blockBuilder.build();
}

@UsedByGeneratedCode
public static long arrayPosition(Type type,MethodHandle equalMethodHandle,Block array,boolean element)
{
    int size = array.getPositionCount();
    for (int i = 0; i < size; i++) {
        if (!array.isNull(i)) {
            boolean arrayValue = type.getBoolean(array,i);
            try {
                if ((boolean) equalMethodHandle.invokeExact(arrayValue,element)) {
                    return i + 1; // result is 1-based (instead of 0)
                }
            }
            catch (Throwable t) {
                Throwables.propagateIfInstanceOf(t,Error.class);
                Throwables.propagateIfInstanceOf(t,PrestoException.class);
                throw new PrestoException(INTERNAL_ERROR,t);
            }
        }
    }
    return 0;
}

@Override
public final void load(LazyBlock lazyBlock)
{
    if (loaded) {
        return;
    }

    checkState(batchId == expectedBatchId);

    try {
        Block block = parquetReader.readBlock(columnDescriptor,batchSize,type);
        lazyBlock.setBlock(block);
    }
    catch (IOException e) {
        throw new PrestoException(HIVE_CURSOR_ERROR,e);
    }
    loaded = true;
}

@UsedByGeneratedCode
public static Object subscript(MethodHandle keyEqualsMethod,Type keyType,Type valueType,Block map,double key)
{
    for (int position = 0; position < map.getPositionCount(); position += 2) {
        try {
            if ((boolean) keyEqualsMethod.invokeExact(keyType.getDouble(map,position),key)) {
                return readNativeValue(valueType,map,position + 1); // position + 1: value position
            }
        }
        catch (Throwable t) {
            Throwables.propagateIfInstanceOf(t,Error.class);
            Throwables.propagateIfInstanceOf(t,PrestoException.class);
            throw new PrestoException(INTERNAL_ERROR,t);
        }
    }
    return null;
}

public static void combine(keyvaluePaiRSState state,keyvaluePaiRSState otherState)
{
    if (state.get() != null && otherState.get() != null) {
        Block keys = otherState.get().getKeys();
        Block values = otherState.get().getValues();
        keyvaluePairs pairs = state.get();
        long startSize = pairs.estimatedInMemorySize();
        for (int i = 0; i < keys.getPositionCount(); i++) {
            try {
                pairs.add(keys,values,i,i);
            }
            catch (ExceededMemoryLimitException e) {
                throw new PrestoException(INVALID_FUNCTION_ARGUMENT,format("The result of map_agg may not exceed %s",e.getMaxMemory()));
            }
        }
        state.addMemoryUsage(pairs.estimatedInMemorySize() - startSize);
    }
    else if (state.get() == null) {
        state.set(otherState.get());
    }
}

@Override
public void addInput(Page page)
{
    requireNonNull(page,"page is null");
    checkState(!finishing,"Operator is finishing");
    checkState(outputPage == null,"Operator still has pending output");
    operatorContext.setMemoryReservation(markdistinctHash.getEstimatedSize());

    Block markerBlock = markdistinctHash.markdistinctRows(page);

    // add the new boolean column to the page
    Block[] sourceBlocks = page.getBlocks();
    Block[] outputBlocks = new Block[sourceBlocks.length + 1]; // +1 for the single boolean output channel

    System.arraycopy(sourceBlocks,outputBlocks,sourceBlocks.length);
    outputBlocks[sourceBlocks.length] = markerBlock;

    outputPage = new Page(outputBlocks);
}

public static void input(BlockComparator comparator,MinMaxByNState state,Block value,Block key,int blockIndex,long n)
{
    TypedkeyvalueHeap heap = state.getTypedkeyvalueHeap();
    if (heap == null) {
        if (n <= 0) {
            throw new PrestoException(INVALID_FUNCTION_ARGUMENT,"third argument of max_by/min_by must be a positive integer");
        }
        heap = new TypedkeyvalueHeap(comparator,keyType,valueType,Ints.checkedCast(n));
        state.setTypedkeyvalueHeap(heap);
    }

    long startSize = heap.getEstimatedSize();
    if (!key.isNull(blockIndex)) {
        heap.add(key,value,blockIndex);
    }
    state.addMemoryUsage(heap.getEstimatedSize() - startSize);
}

public void addPage(Page page)
{
    // ignore empty pages
    if (page.getPositionCount() == 0) {
        return;
    }

    positionCount += page.getPositionCount();

    int pageIndex = (channels.length > 0) ? channels[0].size() : 0;
    for (int i = 0; i < channels.length; i++) {
        Block block = page.getBlock(i);
        channels[i].add(block);
        pagesMemorySize += block.getRetainedSizeInBytes();
    }

    for (int position = 0; position < page.getPositionCount(); position++) {
        long sliceAddress = encodeSyntheticAddress(pageIndex,position);
        valueAddresses.add(sliceAddress);
    }

    estimatedSize = calculateEstimatedSize();
}

@Override
protected List<Object> computeNext()
{
    position++;
    if (position >= page.getPositionCount()) {
        return endOfData();
    }

    List<Object> values = new ArrayList<>(page.getChannelCount());
    for (int channel = 0; channel < page.getChannelCount(); channeL++) {
        Type type = types.get(channel);
        Block block = page.getBlock(channel);
        values.add(type.getobjectValue(session,position));
    }
    return Collections.unmodifiableList(values);
}

@Test
public void testComplexSerialization()
{
    StateCompiler compiler = new StateCompiler();

    AccumulatorStateFactory<TestComplexState> factory = compiler.generateStateFactory(TestComplexState.class);
    AccumulatorStateSerializer<TestComplexState> serializer = compiler.generateStateSerializer(TestComplexState.class);
    TestComplexState singleState = factory.createSingleState();
    TestComplexState deserializedState = factory.createSingleState();

    singleState.setBoolean(true);
    singleState.setLong(1);
    singleState.setDouble(2.0);
    singleState.setByte((byte) 3);

    BlockBuilder builder = VarcharType.VARCHAR.createBlockBuilder(new BlockBuilderStatus(),1);
    serializer.serialize(singleState,builder);

    Block block = builder.build();
    serializer.deserialize(block,deserializedState);

    assertEquals(deserializedState.getBoolean(),singleState.getBoolean());
    assertEquals(deserializedState.getLong(),singleState.getLong());
    assertEquals(deserializedState.getDouble(),singleState.getDouble());
    assertEquals(deserializedState.getByte(),singleState.getByte());
}

public static Block getHashBlock(List<? extends Type> hashTypes,Block... hashBlocks)
{
    checkArgument(hashTypes.size() == hashBlocks.length);
    int[] hashChannels = new int[hashBlocks.length];
    for (int i = 0; i < hashBlocks.length; i++) {
        hashChannels[i] = i;
    }
    HashGenerator hashGenerator = new InterpretedHashGenerator(ImmutableList.copyOf(hashTypes),hashChannels);
    int positionCount = hashBlocks[0].getPositionCount();
    BlockBuilder builder = BIGINT.createFixedSizeBlockBuilder(positionCount);
    Page page = new Page(hashBlocks);
    for (int i = 0; i < positionCount; i++) {
        BIGINT.writeLong(builder,hashGenerator.hashPosition(i,page));
    }
    return builder.build();
}

static List<Page> dropChannel(List<Page> pages,List<Integer> channels)
{
    List<Page> actualPages = new ArrayList<>();
    for (Page page : pages) {
        int channel = 0;
        Block[] blocks = new Block[page.getChannelCount() - channels.size()];
        for (int i = 0; i < page.getChannelCount(); i++) {
            if (channels.contains(i)) {
                continue;
            }
            blocks[channeL++] = page.getBlock(i);
        }
        actualPages.add(new Page(blocks));
    }
    return actualPages;
}

public static Block createTestBlock()
{
    BlockBuilder blockBuilder = BIGINT.createBlockBuilder(new BlockBuilderStatus(),15);
    BIGINT.writeLong(blockBuilder,1111);
    BIGINT.writeLong(blockBuilder,2222);
    BIGINT.writeLong(blockBuilder,3333);
    BIGINT.writeLong(blockBuilder,4444);
    return blockBuilder.build();
}

private Page getNonLazyPage(Page page)
{
    ImmutableSet.Builder<Integer> builder = ImmutableSet.builder();
    for (ProjectionFunction projection : projections) {
        builder.addAll(projection.getInputChannels());
    }
    Set<Integer> inputChannels = builder.build();

    if (inputChannels.isEmpty()) {
        return page;
    }

    Block[] blocks = page.getBlocks();
    for (int inputChannel : inputChannels) {
        Block block = page.getBlock(inputChannel);
        if (block instanceof LazyBlock) {
            blocks[inputChannel] = ((LazyBlock) block).getBlock();
        }
    }
    return new Page(blocks);
}

public static Boolean contains(Type elementType,MethodHandle equals,Block arrayBlock,double value)
{
    boolean foundNull = false;
    for (int i = 0; i < arrayBlock.getPositionCount(); i++) {
        if (arrayBlock.isNull(i)) {
            foundNull = true;
            continue;
        }
        try {
            if ((boolean) equals.invokeExact(elementType.getDouble(arrayBlock,i),value)) {
                return true;
            }
        }
        catch (Throwable t) {
            Throwables.propagateIfInstanceOf(t,PrestoException.class);

            throw new PrestoException(INTERNAL_ERROR,t);
        }
    }
    if (foundNull) {
        return null;
    }
    return false;
}

@Test
public void testReuse()
        throws Exception
{
    BytesWritable value = new BytesWritable();

    byte[] first = "hello world".getBytes(UTF_8);
    value.set(first,first.length);

    byte[] second = "bye".getBytes(UTF_8);
    value.set(second,second.length);

    Type type = new Typetoken<Map<BytesWritable,Integer>>() {}.getType();
    ObjectInspector inspector = getInspector(type);

    Block actual = getBlockObject(new MapType(VARCHAR,BIGINT),ImmutableMap.of(value,0),inspector);
    Block expected = mapBlockOf(VARCHAR,BIGINT,"bye",0);

    assertBlockEquals(actual,expected);
}

@Override
public final void load(LazyBlock lazyBlock)
{
    if (loaded) {
        return;
    }

    checkState(batchId == expectedBatchId);

    try {
        Block block = recordReader.readBlock(type,columnIndex);
        lazyBlock.setBlock(block);
    }
    catch (IOException e) {
        if (e instanceof OrcCorruptionException) {
            throw new PrestoException(HIVE_BAD_DATA,e);
        }
        throw new PrestoException(HIVE_CURSOR_ERROR,e);
    }

    loaded = true;
}

public HDFSPageSource(
        ParquetReader parquetReader,ParquetDataSource dataSource,MessageType fileSchema,MessageType requestedSchema,long totalBytes,List<HDFSColumnHandle> columns,TypeManager typeManager)
{
    checkArgument(totalBytes >= 0,"totalBytes is negative");

    this.parquetReader = requireNonNull(parquetReader,"parquetReader is null");
    this.dataSource = requireNonNull(dataSource,"dataSource is null");
    this.fileSchema = requireNonNull(fileSchema,"fileSchema is null");
    this.requestedSchema = requireNonNull(requestedSchema,"requestedSchema is null");
    this.totalBytes = totalBytes;

    this.columnSize = columns.size();
    this.constantBlocks = new Block[columnSize];
    ImmutableList.Builder<String> namesBuilder = ImmutableList.builder();
    ImmutableList.Builder<Type> typesBuilder = ImmutableList.builder();
    for (int columnIndex = 0; columnIndex < columnSize; columnIndex++) {
        HDFSColumnHandle column = columns.get(columnIndex);
        String name = column.getName();
        Type type = typeManager.getType(column.getType().getTypeSignature());

        namesBuilder.add(name);
        typesBuilder.add(type);

        if (getParquetType(column,fileSchema) == null) {
            constantBlocks[columnIndex] = RunLengthEncodedBlock.create(type,null,MAX_VECTOR_LENGTH);
        }
    }
    columnNames = namesBuilder.build();
    types = typesBuilder.build();
}

private static Block serializeObject(Type type,BlockBuilder builder,Object object) {
    if (!isstructuralType(type)) {
        serializePrimitive(type,builder,object);
        return null;
    } else if (isArrayType(type)) {
        return serializeList(type,object);
    } else if (isMapType(type)) {
        return serializeMap(type,object);
    } else if (isRowType(type)) {
        return serializeStruct(type,object);
    }
    throw new RuntimeException("UnkNown object type: " + type);
}

private static Block serializeList(Type type,Object object) {
    List<?> list = (List) object;
    if (list == null) {
        requireNonNull(builder,"parent builder is null").appendNull();
        return null;
    }

    List<Type> typeParameters = type.getTypeParameters();
    checkArgument(typeParameters.size() == 1,"list must have exactly 1 type parameter");
    Type elementType = typeParameters.get(0);

    BlockBuilder currentBuilder;
    if (builder != null) {
        currentBuilder = builder.beginBlockEntry();
    } else {
        currentBuilder = elementType.createBlockBuilder(new BlockBuilderStatus(),list.size());
    }

    for (Object element : list) {
        serializeObject(elementType,currentBuilder,element);
    }

    if (builder != null) {
        builder.closeEntry();
        return null;
    } else {
        Block resultBlock = currentBuilder.build();
        return resultBlock;
    }
}

private static Block serializeMap(Type type,Object object) {
    Map<?,?> map = (Map) object;
    if (map == null) {
        requireNonNull(builder,"parent builder is null").appendNull();
        return null;
    }

    List<Type> typeParameters = type.getTypeParameters();
    checkArgument(typeParameters.size() == 2,"map must have exactly 2 type parameter");
    Type keyType = typeParameters.get(0);
    Type valueType = typeParameters.get(1);

    BlockBuilder currentBuilder;
    if (builder != null) {
        currentBuilder = builder.beginBlockEntry();
    } else {
        currentBuilder = new InterleavedBlockBuilder(typeParameters,new BlockBuilderStatus(),map.size());
    }

    for (Map.Entry<?,?> entry : map.entrySet()) {
        // Hive skips map entries with null keys
        if (entry.getKey() != null) {
            serializeObject(keyType,entry.getKey());
            serializeObject(valueType,entry.getValue());
        }
    }

    if (builder != null) {
        builder.closeEntry();
        return null;
    } else {
        Block resultBlock = currentBuilder.build();
        return resultBlock;
    }
}

@Override
public void appendTo(Block block,int position,BlockBuilder blockBuilder)
{
    if (block.isNull(position)) {
        blockBuilder.appendNull();
    }
    else {
        block.writeBytesTo(position,block.getLength(position),blockBuilder);
        blockBuilder.closeEntry();
    }
}

@Override
public Object getobjectValue(ConnectorSession session,Block block,int position)
{
    if (block.isNull(position)) {
        return null;
    }

    return new sqlVarbinary(block.getSlice(position,block.getLength(position)).getBytes());
}

@Override
public void deserialize(Block block,int index,HyperLogLogState state)
{
    if (!block.isNull(index)) {
        Slice slice = HyperLogLogType.HYPER_LOG_LOG.getSlice(block,index);
        state.setHyperLogLog(HyperLogLog.fromBytes(slice.getBytes()).toDenseHLL());
    }
}

@TypeParameter(StandardTypes.DOUBLE)
@sqlType(StandardTypes.BOOLEAN)
@Nullable
public static Boolean contains(
        @TypeParameter(StandardTypes.DOUBLE) Type elementType,@sqlType("array(double)") Block arrayBlock,@sqlType(StandardTypes.DOUBLE) double lng,@sqlType(StandardTypes.DOUBLE) double lat)
{
    double[] array= new double[arrayBlock.getPositionCount()] ;
    polygon poly = new polygon();

    for (int i = 0; i < arrayBlock.getPositionCount(); i++) {

        if (arrayBlock.isNull(i)) {
            continue;
        }
        array[i]=elementType.getDouble(arrayBlock,i);

    }

    poly.startPath(array[0],array[1]);
    for (int i = 2; i < array.length; i += 2) {
        poly.lineto(array[i],array[i + 1]);
    }
    return OperatorContains.local().execute(poly,new Point(lng,lat),null);
}

@sqlType("array(double)")
public static Block geohash_decode(@sqlType(StandardTypes.VARCHAR) Slice geohash) {

    BlockBuilder blockBuilder = DOUBLE.createBlockBuilder(new BlockBuilderStatus(),2);
    LatLong coordinates = GeoHash.decodeHash(geohash.toStringUtf8());
    DOUBLE.writeDouble(blockBuilder,coordinates.getLat());
    DOUBLE.writeDouble(blockBuilder,coordinates.getLon());
    return blockBuilder.build();
}

private static Block toBlock(double[] poly) {
    FixedWidthBlockBuilder blockBuilder = new FixedWidthBlockBuilder(8,poly.length);
    for (double d : poly) {
        blockBuilder.writeLong(Double.doubletoLongBits(d));
        blockBuilder.closeEntry();
    }
    return blockBuilder.build();
}

@Test
public void testDouble()
        throws Exception
{
    InternalAggregationFunction doubleAgg = Metadata.getFunctionRegistry().getAggregateFunctionImplementation(new Signature("checksum",AGGREGATE,VARBINARY,DOUBLE));
    Block block = createDoublesBlock(null,2.0,3.0,Double.NEGATIVE_INFINITY,Double.POSITIVE_INFINITY,Double.NaN);
    assertAggregation(doubleAgg,1.0,expectedChecksum(DoubleType.DOUBLE,block),block);
}

public static void input(Type type,SliceState state,int position)
{
    if (state.getSlice() != null) {
        return;
    }
    state.setSlice(type.getSlice(block,position));
}

private static boolean filter(int position,Block discountBlock,Block shipDateBlock,Block quantityBlock)
{
    return !shipDateBlock.isNull(position) && DATE.getLong(shipDateBlock,position) >= MIN_SHIP_DATE &&
            !shipDateBlock.isNull(position) && DATE.getLong(shipDateBlock,position) < MAX_SHIP_DATE &&
            !discountBlock.isNull(position) && DOUBLE.getDouble(discountBlock,position) >= 0.05 &&
            !discountBlock.isNull(position) && DOUBLE.getDouble(discountBlock,position) <= 0.07 &&
            !quantityBlock.isNull(position) && BIGINT.getLong(quantityBlock,position) < 24;
}

@Override
public int hashPosition(int position,Page page)
{
    Block[] blocks = page.getBlocks();
    int result = HashGenerationoptimizer.INITIAL_HASH_VALUE;
    for (int i = 0; i < hashChannels.length; i++) {
        Type type = hashChannelTypes.get(i);
        result = (int) CombineHashFunction.getHash(result,TypeUtils.hashPosition(type,blocks[hashChannels[i]],position));
    }
    return result;
}

public static Block createStringsBlock(Iterable<String> values)
{
    BlockBuilder builder = VARCHAR.createBlockBuilder(new BlockBuilderStatus(),100);

    for (String value : values) {
        if (value == null) {
            builder.appendNull();
        }
        else {
            VARCHAR.writeString(builder,value);
        }
    }

    return builder.build();
}

@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,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,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,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,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,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,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,"presto_test_types_rctext")) == null) {
        return;
    }

    ConnectorTableHandle tableHandle = getTableHandle(new SchemaTableName(database,"presto_test_types_rctext"));
    ConnectorTableMetadata tableMetadata = Metadata.getTableMetadata(session,ImmutableSet.<HiveRecordCursorProvider>of(new ColumnarTextHiveRecordCursorProvider()),columnHandles);
    assertGetRecords(RCTEXT,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,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,in(requiredTableScanOutputs));

    return new TableScanNode(
            node.getId(),node.getTable(),node.getLayout(),node.getCurrentConstraint(),node.getoriginalConstraint());
}

@Override
public ConnectorPageSource createPageSource(ConnectorTransactionHandle transactionHandle,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,ConnectorTableHandle tableHandle)
{
    HDFSTableHandle table = checkType(tableHandle,"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,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,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,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,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(),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,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,"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,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,clientManager,handles,messageDecoder,messageFieldDecoders,kinesisConnectorConfig);
}

@Override
public Map<String,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(),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.singleValue(VARCHAR,utf8Slice("value")))
                    .put(B,Domain.singleValue(DOUBLE,0.0))
                    .put(C,Domain.create(ValueSet.ofRanges(Range.lessthan(DOUBLE,10.0)),false))
                    .build());

    TupleDomain<ColumnHandle> expectedTupleDomain = TupleDomain.withColumnDomains(
            ImmutableMap.<ColumnHandle,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,Optional<Set<ColumnHandle>> desiredColumns)
{
    try (threadcontextClassLoader ignored = new threadcontextClassLoader(classLoader)) {
        return delegate.getTableLayouts(session,table,constraint,desiredColumns);
    }
}

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,"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,"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,ordinalPosition,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,"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,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,SchemaTablePrefix prefix)
{
    ImmutableMap.Builder<SchemaTableName,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(),"num_ers"));
    assertEquals(specialTableMetadata.getColumns(),ImmutableList.of(
            new ColumnMetadata("te_t",new ColumnMetadata("va%ue",false)));

    // unkNown tables should produce null
    unkNownTableMetadata(new JdbcTableHandle(CONNECTOR_ID,new SchemaTableName("u","numbers"),null,"unkNown","unkNown"));
    unkNownTableMetadata(new JdbcTableHandle(CONNECTOR_ID,"example","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(),"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,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,fiberColIndex,function,timstampColIndex,dataType);
    }

/**
 * Gets the Metadata for all columns that match the specified table prefix.
 *
 * @param session session
 * @param prefix prefix
 */
@Override
public Map<SchemaTableName,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",new ColumnMetadata("text",createUnboundedVarcharType())));
}

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

@Override
public RecordSet getRecordSet(
        ConnectorTransactionHandle connectorTransactionHandle,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",new ColumnMetadata("retweet_count",new ColumnMetadata("user_name",new ColumnMetadata("user_screen_name",createUnboundedVarcharType())));
}

@Override
public ColumnMetadata getColumnMetadata(
        ConnectorSession session,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(),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(),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",false);
}

@Override
public Map<SchemaTableName,"prefix is null");
    ImmutableMap.Builder<SchemaTableName,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",new ColumnMetadata("b",false)));
}

@Test
public void getColumnMetadata()
{
    assertEquals(Metadata.getColumnMetadata(SESSION,NUMBERS_TABLE_HANDLE,new ExampleColumnHandle(CONNECTOR_ID,"text",0)),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",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,"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(),false);
    assertPrimitiveField(map,"t_tinyint",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(),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,SchemaTablePrefix prefix)
{
    return tables.values().stream()
            .filter(table -> prefix.matches(table.toSchemaTableName()))
            .collect(toMap(BlackHoleTableHandle::toSchemaTableName,handle -> handle.toTableMetadata().getColumns()));
}