@Override
protected Object visitArrayConstructor(ArrayConstructor node,Object context)
{
    Type elementType = ((ArrayType) expressionTypes.get(node)).getElementType();
    BlockBuilder arrayBlockBuilder = elementType.createBlockBuilder(new BlockBuilderStatus(),node.getValues().size());

    for (Expression expression : node.getValues()) {
        Object value = process(expression,context);
        if (value instanceof Expression) {
            return visitFunctionCall(new FunctionCall(Qualifiedname.of(ArrayConstructor.ARRAY_CONSTRUCTOR),node.getValues()),context);
        }
        writeNativeValue(elementType,arrayBlockBuilder,value);
    }

    return arrayBlockBuilder.build();
}

@Test
public void testArraySubscript()
        throws Exception
{
    assertExpression("ARRAY [1,2][1]",new SubscriptExpression(
                    new ArrayConstructor(ImmutableList.<Expression>of(new LongLiteral("1"),new LongLiteral("2"))),new LongLiteral("1"))
    );
    try {
        assertExpression("CASE WHEN TRUE THEN ARRAY[1,2] END[1]",null);
        fail();
    }
    catch (RuntimeException e) {
        // Expected
    }
}

@Override
protected String visitArrayConstructor(ArrayConstructor node,Void context)
{
    ImmutableList.Builder<String> valueStrings = ImmutableList.builder();
    for (Expression value : node.getValues()) {
        valueStrings.add(sqlFormatter.formatsql(value));
    }
    return "ARRAY[" + Joiner.on(",").join(valueStrings.build()) + "]";
}

@Override
protected String visitArrayConstructor(ArrayConstructor node,StackableAstVisitorContext<Integer> indent)
{
    ImmutableList.Builder<String> valueStrings = ImmutableList.builder();
    for (Expression value : node.getValues()) {
        valueStrings.add(formatExpression(value,parameters,indent.getContext() + 1));
    }
    return "ARRAY[" + Joiner.on(",").join(valueStrings.build()) + "]";
}

@Override
protected RowExpression visitArrayConstructor(ArrayConstructor node,Void context)
{
    List<RowExpression> arguments = node.getValues().stream()
            .map(value -> process(value,context))
            .collect(toImmutableList());
    List<Type> argumentTypes = arguments.stream()
            .map(RowExpression::getType)
            .collect(toImmutableList());
    return call(arrayConstructorSignature(types.get(node),argumentTypes),types.get(node),arguments);
}

@Override
protected Type visitArrayConstructor(ArrayConstructor node,StackableAstVisitorContext<AnalysisContext> context)
{
    Type type = coercetoSingleType(context,"All ARRAY elements must be the same type: %s",node.getValues());
    Type arrayType = typeManager.getParameterizedType(ARRAY.getName(),ImmutableList.of(type.getTypeSignature()),ImmutableList.of());
    expressionTypes.put(node,arrayType);
    return arrayType;
}

@Override
protected String visitArrayConstructor(ArrayConstructor node,Boolean unmangleNames)
{
    ImmutableList.Builder<String> valueStrings = ImmutableList.builder();
    for (Expression value : node.getValues()) {
        valueStrings.add(formatsql(value,unmangleNames));
    }
    return "ARRAY[" + Joiner.on(",").join(valueStrings.build()) + "]";
}

@Test
public void testArrayConstructor()
        throws Exception
{
    assertExpression("ARRAY []",new ArrayConstructor(ImmutableList.<Expression>of()));
    assertExpression("ARRAY [1,2]",new ArrayConstructor(ImmutableList.<Expression>of(new LongLiteral("1"),new LongLiteral("2"))));
    assertExpression("ARRAY [1.0,2.5]",new ArrayConstructor(ImmutableList.<Expression>of(new DoubleLiteral("1.0"),new DoubleLiteral("2.5"))));
    assertExpression("ARRAY ['hi']",new ArrayConstructor(ImmutableList.<Expression>of(new StringLiteral("hi"))));
    assertExpression("ARRAY ['hi','hello']",new ArrayConstructor(ImmutableList.<Expression>of(new StringLiteral("hi"),new StringLiteral("hello"))));
}

@Override
protected String visitArrayConstructor(ArrayConstructor node,").join(valueStrings.build()) + "]";
}

@Override
protected Boolean visitArrayConstructor(ArrayConstructor node,Void context)
{
    return node.getValues().stream().allMatch(expression -> process(expression,context));
}

@Override
public Node visitArrayConstructor(sqlbaseParser.ArrayConstructorContext context)
{
    return new ArrayConstructor(getLocation(context),visit(context.expression(),Expression.class));
}

@Override
public Connector create(final String connectorId,Map<String,String> requiredConfig)
{
    requireNonNull(requiredConfig,"requiredConfig is null");
    requireNonNull(optionalConfig,"optionalConfig is null");

    try {
        // A plugin is not required to use Guice; it is just very convenient
        Bootstrap app = new Bootstrap(
                new JsonModule(),new ExampleModule(connectorId,typeManager));

    Injector injector = app
                .strictConfig()
                .donotinitializeLogging()
                .setrequiredConfigurationProperties(requiredConfig)
                .setoptionalConfigurationProperties(optionalConfig)
                .initialize();

        return injector.getInstance(ExampleConnector.class);
    }
    catch (Exception e) {
        throw Throwables.propagate(e);
    }
}

@Override
public Connector create(String connectorId,"optionalConfig is null");

    try (threadcontextClassLoader ignored = new threadcontextClassLoader(classLoader)) {
        Bootstrap app = new Bootstrap(new JdbcModule(connectorId),module);

        Injector injector = app
                .strictConfig()
                .donotinitializeLogging()
                .setrequiredConfigurationProperties(requiredConfig)
                .setoptionalConfigurationProperties(optionalConfig)
                .initialize();

        return injector.getInstance(JdbcConnector.class);
    }
    catch (Exception e) {
        throw Throwables.propagate(e);
    }
}

private static void assertCreateConnector(String metastoreUri)
{
    HiveConnectorFactory connectorFactory = new HiveConnectorFactory(
            "hive-test",ImmutableMap.<String,String>builder()
                    .put("node.environment","test")
                    .put("hive.metastore.uri",metastoreUri)
                    .build(),HiveConnector.class.getClassLoader(),null,new TypeRegistry(),new GroupByHashPageIndexerFactory());

    Connector connector = connectorFactory.create("hive-test",String>of());
    assertInstanceOf(connector.getMetadata(),ClassLoaderSafeConnectorMetadata.class);
    assertInstanceOf(connector.getSplitManager(),ClassLoaderSafeConnectorSplitManager.class);
    assertInstanceOf(connector.getPageSourceProvider(),ConnectorPageSourceProvider.class);
}

@Override
public Connector create(final String connectorId,String> requiredConfig) {
    checkNotNull(requiredConfig,"requiredConfig is null");
    checkNotNull(optionalConfig,new RiakModule(connectorId,typeManager));

        Injector injector = app
                .strictConfig()
                .donotinitializeLogging()
                .setrequiredConfigurationProperties(requiredConfig)
                .setoptionalConfigurationProperties(optionalConfig)
                .initialize();

        return injector.getInstance(RiakConnector.class);
    } catch (Exception e) {
        throw Throwables.propagate(e);
    }
}

@Override
public Connector create(String connectorId,String> config)
{
    requireNonNull(connectorId,"connectorId is null");
    requireNonNull(config,"config is null");

    try {
        Bootstrap app = new Bootstrap(
                new JsonModule(),new RedisConnectorModule(),binder -> {
                    binder.bind(RedisConnectorId.class).toInstance(new RedisConnectorId(connectorId));
                    binder.bind(TypeManager.class).toInstance(typeManager);
                    binder.bind(NodeManager.class).toInstance(nodeManager);

                    if (tableDescriptionsupplier.isPresent()) {
                        binder.bind(new TypeLiteral<supplier<Map<SchemaTableName,RedisTableDescription>>>() {}).toInstance(tableDescriptionsupplier.get());
                    }
                    else {
                        binder.bind(new TypeLiteral<supplier<Map<SchemaTableName,RedisTableDescription>>>() {})
                                .to(RedisTableDescriptionsupplier.class)
                                .in(Scopes.SINGLetoN);
                    }
                }
        );

        Injector injector = app.strictConfig()
                .donotinitializeLogging()
                .setrequiredConfigurationProperties(config)
                .setoptionalConfigurationProperties(optionalConfig)
                .initialize();

        return injector.getInstance(RedisConnector.class);
    }
    catch (Exception e) {
        throw Throwables.propagate(e);
    }
}

@Test
public void testStartup()
{
    ConnectorFactory factory = testConnectorExists();
    Connector c = factory.create("test-connector",String>builder()
            .put("redis.table-names","test")
            .put("redis.nodes","localhost:6379")
            .build());
    assertNotNull(c);
}

public sqlEngine(StructuredStore store,ExecutorService executor) {
    this.store = store;
    this.executor = executor;
    MetadataManager MetadataManager = new MetadataManager();

    SplitManager splitManager = new SplitManager(Sets.<ConnectorSplitManager> newHashSet());

    this.dataStreamManager = new DataStreamManager();
    HandleResolver handleResolver = new HandleResolver();
    Map<String,ConnectorFactory> connectorFactories = Maps.newHashMap();
    Map<String,Connector> globalConnectors = Maps.newHashMap();

    RecordSinkManager recordSinkManager = new RecordSinkManager();
    Map<String,ConnectorOutputHandleResolver> handleIdResolvers = Maps.newHashMap();
    OutputTableHandleResolver outputTableHandleResolver = new OutputTableHandleResolver(handleIdResolvers);

    this.connectorManager = new ConnectorManager(MetadataManager,splitManager,dataStreamManager,recordSinkManager,handleResolver,outputTableHandleResolver,connectorFactories,globalConnectors);

    // NodeManager nodeManager = new InMemoryNodeManager();
    PlanoptimizersFactory planoptimizersFactory = new PlanoptimizersFactory(MetadataManager,splitManager);
    List<Planoptimizer> planoptimizers = planoptimizersFactory.get();

    this.MetadataManager = MetadataManager;
    this.planoptimizers = planoptimizers;
    this.periodicImportManager = new StubPeriodicImportManager();
    this.storageManager = new StubStorageManager();

    NodeManager nodeManager = new InMemoryNodeManager();
    CloudataConnectorFactory cloudataConnectorFactory = new CloudataConnectorFactory(nodeManager,Maps.<String,String> newHashMap(),store);

    connectorManager.addConnectorFactory(cloudataConnectorFactory);

    connectorManager.createConnection(catalogName,CloudataConnectorFactory.PROVIDER_ID,String> newHashMap());

    this.cloudataConnector = cloudataConnectorFactory.get(catalogName);
}

@Override
public Connector create(final String connectorId,"optionalConfig is null");

    try {
        // // A plugin is not required to use Guice; it is just very convenient
        // Bootstrap app = new Bootstrap(new JsonModule(),new ExampleModule(connectorId));
        //
        // Injector injector = app.strictConfig().donotinitializeLogging()
        // .setrequiredConfigurationProperties(requiredConfig)
        // .setoptionalConfigurationProperties(optionalConfig).initialize();

        ClasstoInstanceMap<Object> services = ImmutableClasstoInstanceMap.builder()

        .put(ConnectorMetadata.class,new CloudataConnectorMetadata(connectorId,store))

        .put(ConnectorSplitManager.class,new CloudataSplitManager(nodeManager,connectorId))

        .put(ConnectorRecordSetProvider.class,new CloudataConnectorRecordSetProvider())

        .put(ConnectorHandleResolver.class,new CloudataConnectorHandleResolver()).build();

        CloudataConnector connector = new CloudataConnector(store,services);
        connectors.put(connectorId,connector);
        return connector;
    } catch (Exception e) {
        throw Throwables.propagate(e);
    }
}

@Override
public Connector create(String connectorId,String> config)
{
    requireNonNull(config,"config is null");

    try (threadcontextClassLoader ignored = new threadcontextClassLoader(classLoader)) {
        Bootstrap app = new Bootstrap(
                new NodeModule(),new MBeanModule(),new JsonModule(),new HiveClientModule(connectorId,metastore,typeManager,pageIndexerFactory),installModuleIf(
                        SecurityConfig.class,security -> ALLOW_ALL_ACCESS_CONTROL.equalsIgnoreCase(security.getSecuritySystem()),new NoSecurityModule()),security -> "read-only".equalsIgnoreCase(security.getSecuritySystem()),new ReadOnlySecurityModule()),security -> "sql-standard".equalsIgnoreCase(security.getSecuritySystem()),new sqlStandardSecurityModule()),binder -> {
                    MBeanServer platformMBeanServer = ManagementFactory.getPlatformMBeanServer();
                    binder.bind(MBeanServer.class).toInstance(new RebindSafeMBeanServer(platformMBeanServer));
                }
        );

        Injector injector = app
                .strictConfig()
                .donotinitializeLogging()
                .setrequiredConfigurationProperties(config)
                .setoptionalConfigurationProperties(optionalConfig)
                .initialize();

        LifeCycleManager lifeCycleManager = injector.getInstance(LifeCycleManager.class);
        ConnectorMetadata Metadata = injector.getInstance(ConnectorMetadata.class);
        ConnectorSplitManager splitManager = injector.getInstance(ConnectorSplitManager.class);
        ConnectorPageSourceProvider connectorPageSource = injector.getInstance(ConnectorPageSourceProvider.class);
        ConnectorPageSinkProvider pageSinkProvider = injector.getInstance(ConnectorPageSinkProvider.class);
        HiveSessionProperties hiveSessionProperties = injector.getInstance(HiveSessionProperties.class);
        HiveTableProperties hiveTableProperties = injector.getInstance(HiveTableProperties.class);
        ConnectorAccessControl accessControl = injector.getInstance(ConnectorAccessControl.class);

        return new HiveConnector(
                lifeCycleManager,new ClassLoaderSafeConnectorMetadata(Metadata,classLoader),new ClassLoaderSafeConnectorSplitManager(splitManager,new ClassLoaderSafeConnectorPageSourceProvider(connectorPageSource,new ClassLoaderSafeConnectorPageSinkProvider(pageSinkProvider,ImmutableSet.of(),hiveSessionProperties.getSessionProperties(),hiveTableProperties.getTableProperties(),accessControl);
    }
    catch (Exception e) {
        throw Throwables.propagate(e);
    }
}

@Override
    public Connector create(String connectorId,String> config)
    {
        checkNotNull(config,"config is null");

        try {
            KafkaClientModule kafkaClientModule = new KafkaClientModule(connectorId);

            Bootstrap app = new Bootstrap(
                    new NodeModule(),kafkaClientModule
                );

            Injector injector = app.strictConfig().donotinitializeLogging()
                    .setrequiredConfigurationProperties(config)
                    .quiet()
                    .requireExplicitBindings(false)
                    .setoptionalConfigurationProperties(optionalConfig).initialize();

            KafkaClientConfig clientConfig = KafkaClientConfig.INSTANCE;
            KafkaPluginConfig pluginConfig = KafkaPluginConfig.INSTANCE;
            KafkaConnectorId kafkaConnectorId = KafkaConnectorId.INSTANCE;
            KafkaHiveClient hiveClient = new KafkaHiveClient(kafkaConnectorId,clientConfig,pluginConfig);
            KafkaMetadata kafkaMetadata = new KafkaMetadata(hiveClient,kafkaConnectorId);
            KafkaSplitManager kafkaSplitManager = new KafkaSplitManager(hiveClient,kafkaConnectorId,clientConfig);
            KafkaRecordSetProvider kafkaRecordSetProvider = new KafkaRecordSetProvider(kafkaConnectorId);
            KafkaHandleResolver kafkaHandleResolver = new KafkaHandleResolver(kafkaConnectorId);

            ConnectorMetadata connMetadata = new ClassLoaderSafeConnectorMetadata(kafkaMetadata,classLoader);
            ConnectorSplitManager connSplitManager = new ClassLoaderSafeConnectorSplitManager(kafkaSplitManager,classLoader);
            ConnectorRecordSetProvider connRecordSetProvider = new ClassLoaderSafeConnectorRecordSetProvider(kafkaRecordSetProvider,classLoader);
            ConnectorHandleResolver connHandleResolver = new ClassLoaderSafeConnectorHandleResolver(kafkaHandleResolver,classLoader);

            return new KafkaConnector(connMetadata,connSplitManager,connRecordSetProvider,connHandleResolver);

//            return injector.getInstance(KafkaConnector.class);

//            KafkaMetadata kafkaMetadata = injector.getInstance(KafkaMetadata.class);
//            KafkaSplitManager kafkaSplitManager = injector.getInstance(KafkaSplitManager.class);
//            KafkaRecordSetProvider kafkaRecordSetProvider = injector.getInstance(KafkaRecordSetProvider.class);
//            KafkaHandleResolver kafkaHandleResolver = injector.getInstance(KafkaHandleResolver.class);
//            return new KafkaConnector(kafkaMetadata,kafkaSplitManager,//                kafkaRecordSetProvider,kafkaHandleResolver);
//            return new KafkaConnector(
//                    new ClassLoaderSafeConnectorMetadata(kafkaMetadata,//                    new ClassLoaderSafeConnectorSplitManager(kafkaSplitManager,//                    new ClassLoaderSafeConnectorRecordSetProvider(kafkaRecordSetProvider,//                    new ClassLoaderSafeConnectorHandleResolver(kafkaHandleResolver,classLoader));
        } catch (Exception e) {
            e.printstacktrace();
            throw Throwables.propagate(e);
        }
    }

@Test
public void testSanity()
        throws InterruptedException
{
    List<ConnectorTableLayoutResult> layouts = Metadata.getTableLayouts(SESSION,tableHandle,Constraint.alwaysTrue(),Optional.empty());
    assertEquals(layouts.size(),1);
    ConnectorTableLayoutResult layout = getonlyElement(layouts);
    assertInstanceOf(layout.getTableLayout().getHandle(),RaptorTableLayoutHandle.class);

    ConnectorSplitSource splitSource = getSplits(raptorSplitManager,layout);
    int splitCount = 0;
    while (!splitSource.isFinished()) {
        splitCount += getFutureValue(splitSource.getNextBatch(1000)).size();
    }
    assertEquals(splitCount,4);
}

@Test
public void testAssignRandomNodeWhenBackupAvailable()
        throws InterruptedException,URISyntaxException
{
    InMemoryNodeManager nodeManager = new InMemoryNodeManager();
    RaptorConnectorId connectorId = new RaptorConnectorId("raptor");
    Nodesupplier nodesupplier = new RaptorNodesupplier(nodeManager,connectorId);
    PrestoNode node = new PrestoNode(UUID.randomUUID().toString(),new URI("http://127.0.0.1/"),NodeVersion.UNKNowN);
    nodeManager.addNode(connectorId.toString(),node);
    RaptorSplitManager raptorSplitManagerWithBackup = new RaptorSplitManager(connectorId,nodesupplier,shardManager,true);

    deleteShardNodes();

    ConnectorTableLayoutResult layout = getonlyElement(Metadata.getTableLayouts(SESSION,Optional.empty()));
    ConnectorSplitSource partitionSplit = getSplits(raptorSplitManagerWithBackup,layout);
    List<ConnectorSplit> batch = getFutureValue(partitionSplit.getNextBatch(1),PrestoException.class);
    assertEquals(getonlyElement(getonlyElement(batch).getAddresses()),node.getHostAndPort());
}

@Override
public List<ConnectorTableLayoutResult> getTableLayouts(
        ConnectorSession session,ConnectorTableHandle handle,Constraint<ColumnHandle> constraint,Optional<Set<ColumnHandle>> desiredColumns)
{
    requireNonNull(handle,"handle is null");
    checkArgument(handle instanceof BlackHoleTableHandle);
    BlackHoleTableHandle blackHoleHandle = (BlackHoleTableHandle) handle;

    BlackHoleTableLayoutHandle layoutHandle = new BlackHoleTableLayoutHandle(
            blackHoleHandle.getSplitCount(),blackHoleHandle.getPagesPerSplit(),blackHoleHandle.getRowsPerPage(),blackHoleHandle.getFieldsLength());
    return ImmutableList.of(new ConnectorTableLayoutResult(getTableLayout(session,layoutHandle),TupleDomain.all()));
}

@Override
public List<TableLayoutResult> getLayouts(Session session,TableHandle table,Optional<Set<ColumnHandle>> desiredColumns)
{
    if (constraint.getSummary().isNone()) {
        return ImmutableList.of();
    }

    TupleDomain<ColumnHandle> summary = constraint.getSummary();
    String connectorId = table.getConnectorId();
    ConnectorTableHandle connectorTable = table.getConnectorHandle();
    Predicate<Map<ColumnHandle,NullableValue>> predicate = constraint.predicate();

    ConnectorEntry entry = getConnectorMetadata(connectorId);
    ConnectorMetadata Metadata = entry.getMetadata(session);
    ConnectorTransactionHandle transaction = entry.getTransactionHandle(session);
    ConnectorSession connectorSession = session.toConnectorSession(entry.getCatalog());
    List<ConnectorTableLayoutResult> layouts = Metadata.getTableLayouts(connectorSession,connectorTable,new Constraint<>(summary,predicate::test),desiredColumns);

    return layouts.stream()
            .map(layout -> new TableLayoutResult(fromConnectorLayout(connectorId,transaction,layout.getTableLayout()),layout.getUnenforcedConstraint()))
            .collect(toImmutableList());
}

@Test
public void testGetRecordsS3()
        throws Exception
{
    ConnectorTableHandle table = getTableHandle(tableS3);
    List<ColumnHandle> columnHandles = ImmutableList.copyOf(Metadata.getColumnHandles(SESSION,table).values());
    Map<String,Integer> columnIndex = indexColumns(columnHandles);

    List<ConnectorTableLayoutResult> tableLayoutResults = Metadata.getTableLayouts(SESSION,table,new Constraint<>(TupleDomain.all(),bindings -> true),Optional.empty());
    HiveTableLayoutHandle layoutHandle = (HiveTableLayoutHandle) getonlyElement(tableLayoutResults).getTableLayout().getHandle();
    assertEquals(layoutHandle.getPartitions().get().size(),1);
    ConnectorSplitSource splitSource = splitManager.getSplits(SESSION,layoutHandle);

    long sum = 0;

    for (ConnectorSplit split : getAllSplits(splitSource)) {
        try (ConnectorPageSource pageSource = pageSourceProvider.createPageSource(SESSION,split,columnHandles)) {
            MaterializedResult result = materializeSourceDataStream(SESSION,pageSource,getTypes(columnHandles));

            for (MaterializedRow row : result) {
                sum += (Long) row.getField(columnIndex.get("t_bigint"));
            }
        }
    }
    assertEquals(sum,78300);
}

@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,new Constraint<>(tupleDomain,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");
    }
}

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

    ConnectorTableHandle table = getTableHandle(tablePartitionSchemaChangeNonCanonical);
    ColumnHandle column = Metadata.getColumnHandles(session,table).get("t_boolean");
    assertNotNull(column);
    List<ConnectorTableLayoutResult> tableLayoutResults = Metadata.getTableLayouts(session,new Constraint<>(TupleDomain.fromFixedValues(ImmutableMap.of(column,NullableValue.of(BOOLEAN,false))),Optional.empty());
    ConnectorTableLayoutHandle layoutHandle = getonlyElement(tableLayoutResults).getTableLayout().getHandle();
    assertEquals(getAllPartitions(layoutHandle).size(),1);
    assertEquals(getPartitionId(getAllPartitions(layoutHandle).get(0)),"t_boolean=0");

    ConnectorSplitSource splitSource = splitManager.getSplits(session,layoutHandle);
    ConnectorSplit split = getonlyElement(getAllSplits(splitSource));

    ImmutableList<ColumnHandle> columnHandles = ImmutableList.of(column);
    try (ConnectorPageSource ignored = pageSourceProvider.createPageSource(session,columnHandles)) {
        // Todo coercion of non-canonical values should be supported
        fail("expected exception");
    }
    catch (PrestoException e) {
        assertEquals(e.getErrorCode(),HIVE_INVALID_PARTITION_VALUE.toErrorCode());
    }
}

private List<HivePartition> getPartitions(String tableName)
{
    Session session = getSession();
    Metadata Metadata = ((distributedQueryRunner) queryRunner).getCoordinator().getMetadata();

    return transaction(queryRunner.getTransactionManager())
            .readOnly()
            .execute(session,transactionSession -> {
                Optional<TableHandle> tableHandle = Metadata.getTableHandle(transactionSession,new QualifiedobjectName(HIVE_CATALOG,TPCH_SCHEMA,tableName));
                assertTrue(tableHandle.isPresent());

                List<TableLayoutResult> layouts = Metadata.getLayouts(transactionSession,tableHandle.get(),Optional.empty());
                TableLayout layout = Iterables.getonlyElement(layouts).getLayout();
                return ((HiveTableLayoutHandle) layout.getHandle().getConnectorHandle()).getPartitions().get();
            });
}

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

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

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

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

@Override
public List<ConnectorTableLayoutResult> getTableLayouts(ConnectorSession session,Optional<Set<ColumnHandle>> desiredColumns)
{
    RaptorTableHandle handle = checkType(table,RaptorTableHandle.class,"table");
    ConnectorTableLayout layout = new ConnectorTableLayout(new RaptorTableLayoutHandle(handle,constraint.getSummary()));
}

@Test(expectedExceptions = PrestoException.class,expectedExceptionsMessageRegExp = "No host for shard .* found: \\[\\]")
public void testNoHostForShard()
        throws InterruptedException
{
    deleteShardNodes();

    ConnectorTableLayoutResult layout = getonlyElement(Metadata.getTableLayouts(SESSION,Optional.empty()));
    ConnectorSplitSource splitSource = getSplits(raptorSplitManager,layout);
    getFutureValue(splitSource.getNextBatch(1000));
}

@Test(expectedExceptions = PrestoException.class,expectedExceptionsMessageRegExp = "No nodes available to run query")
public void testNoNodes()
        throws InterruptedException,URISyntaxException
{
    deleteShardNodes();

    RaptorSplitManager raptorSplitManagerWithBackup = new RaptorSplitManager(new RaptorConnectorId("fbraptor"),ImmutableSet::of,true);
    ConnectorTableLayoutResult layout = getonlyElement(Metadata.getTableLayouts(SESSION,Optional.empty()));
    ConnectorSplitSource splitSource = getSplits(raptorSplitManagerWithBackup,layout);
    getFutureValue(splitSource.getNextBatch(1000),PrestoException.class);
}

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

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

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

@Override
public 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)
{
    JmxTableHandle handle = checkType(table,JmxTableHandle.class,"table");
    ConnectorTableLayout layout = new ConnectorTableLayout(new JmxTableLayoutHandle(handle,constraint.getSummary()));
}

@Override
public List<ConnectorTableLayoutResult> getTableLayouts(ConnectorSession session,Optional<Set<ColumnHandle>> desiredColumns)
{
    JdbcTableHandle tableHandle = checkType(table,JdbcTableHandle.class,"table");
    ConnectorTableLayout layout = new ConnectorTableLayout(new JdbcTableLayoutHandle(tableHandle,constraint.getSummary()));
}

@Override
public List<ConnectorTableLayoutResult> getTableLayouts(ConnectorSession session,Optional<Set<ColumnHandle>> desiredColumns)
{
    informationSchemaTableHandle handle = checkType(table,informationSchemaTableHandle.class,"table");
    ConnectorTableLayout layout = new ConnectorTableLayout(new informationSchemaTableLayoutHandle(handle,constraint.getSummary()));
}

@Override
public List<ConnectorTableLayoutResult> getTableLayouts(ConnectorSession session,Optional<Set<ColumnHandle>> desiredColumns)
{
    SystemTableHandle tableHandle = checkType(table,SystemTableHandle.class,"table");
    ConnectorTableLayout layout = new ConnectorTableLayout(new SystemTableLayoutHandle(tableHandle.getConnectorId(),constraint.getSummary()));
}

@Override
public List<ConnectorTableLayoutResult> getTableLayouts(ConnectorSession session,Optional<Set<ColumnHandle>> desiredColumns)
{
    TpchTableHandle tableHandle = checkType(table,TpchTableHandle.class,"table");

    Optional<Set<ColumnHandle>> partitioningColumns = Optional.empty();
    List<LocalProperty<ColumnHandle>> localProperties = ImmutableList.of();

    Map<String,ColumnHandle> columns = getColumnHandles(session,tableHandle);
    if (tableHandle.getTableName().equals(TpchTable.ORDERS.getTableName())) {
        partitioningColumns = Optional.of(ImmutableSet.of(columns.get(OrderColumn.ORDER_KEY.getColumnName())));
        localProperties = ImmutableList.of(new SortingProperty<>(columns.get(OrderColumn.ORDER_KEY.getColumnName()),SortOrder.ASC_NULLS_FirsT));
    }
    else if (tableHandle.getTableName().equals(TpchTable.LINE_ITEM.getTableName())) {
        partitioningColumns = Optional.of(ImmutableSet.of(columns.get(LineItemColumn.ORDER_KEY.getColumnName())));
        localProperties = ImmutableList.of(
                new SortingProperty<>(columns.get(LineItemColumn.ORDER_KEY.getColumnName()),SortOrder.ASC_NULLS_FirsT),new SortingProperty<>(columns.get(LineItemColumn.LINE_NUMBER.getColumnName()),SortOrder.ASC_NULLS_FirsT));
    }

    ConnectorTableLayout layout = new ConnectorTableLayout(
            new TpchTableLayoutHandle(tableHandle),Optional.<List<ColumnHandle>>empty(),TupleDomain.<ColumnHandle>all(),// Todo: return well-kNown properties (e.g.,orderkey > 0,etc)
            partitioningColumns,Optional.empty(),localProperties);

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

@Override
public ConnectorTableLayout getTableLayout(ConnectorSession session,ConnectorTableLayoutHandle handle)
{
    TpchTableLayoutHandle layout = checkType(handle,TpchTableLayoutHandle.class,"layout");

    // tables in this connector have a single layout
    return getTableLayouts(session,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,constraint,desiredColumns);
    }
}

@Override
public List<ConnectorTableLayoutResult> getTableLayouts(ConnectorSession session,Optional<Set<ColumnHandle>> desiredColumns)
{
    CassandratableHandle handle = checkType(table,CassandratableHandle.class,"table");
    CassandraPartitionResult result = partitionManager.getPartitions(handle,constraint.getSummary());
    ConnectorTableLayout layout = getTableLayout(session,new CassandratableLayoutHandle(handle,result.getPartitions()));
    return ImmutableList.of(new ConnectorTableLayoutResult(layout,result.getUnenforcedConstraint()));
}

private List<HivePartition> getorComputePartitions(HiveTableLayoutHandle layoutHandle,ConnectorSession session,ConnectorTableHandle tableHandle)
{
    if (layoutHandle.getPartitions().isPresent()) {
        return layoutHandle.getPartitions().get();
    }
    else {
        TupleDomain<ColumnHandle> promisedPredicate = layoutHandle.getPromisedPredicate();
        Predicate<Map<ColumnHandle,NullableValue>> predicate = convertToPredicate(promisedPredicate);
        List<ConnectorTableLayoutResult> tableLayoutResults = getTableLayouts(session,new Constraint<>(promisedPredicate,predicate),Optional.empty());
        return checkType(Iterables.getonlyElement(tableLayoutResults).getTableLayout().getHandle(),HiveTableLayoutHandle.class,"tableLayoutHandle").getPartitions().get();
    }
}

@Override
public List<ConnectorTableLayoutResult> getTableLayouts(ConnectorSession session,ConnectorTableHandle tableHandle,Optional<Set<ColumnHandle>> desiredColumns)
{
    HiveTableHandle handle = checkType(tableHandle,HiveTableHandle.class,"tableHandle");

    HivePartitionResult hivePartitionResult = partitionManager.getPartitions(session,metastore,constraint.getSummary());

    return ImmutableList.of(new ConnectorTableLayoutResult(
            getTableLayout(session,new HiveTableLayoutHandle(handle.getClientId(),hivePartitionResult.getPartitions(),hivePartitionResult.getEnforcedConstraint())),hivePartitionResult.getUnenforcedConstraint()));
}

@Test
public void testGetPartitions()
        throws Exception
{
    ConnectorTableHandle tableHandle = getTableHandle(tablePartitionFormat);
    List<ConnectorTableLayoutResult> tableLayoutResults = Metadata.getTableLayouts(newSession(),Optional.empty());
    assertExpectedTableLayout(getonlyElement(tableLayoutResults).getTableLayout(),tableLayout);
}

@Test
public void testGetPartitionsWithBindings()
        throws Exception
{
    ConnectorTableHandle tableHandle = getTableHandle(tablePartitionFormat);
    List<ConnectorTableLayoutResult> tableLayoutResults = Metadata.getTableLayouts(newSession(),new Constraint<>(TupleDomain.withColumnDomains(ImmutableMap.of(intColumn,Domain.singleValue(BIGINT,5L))),tableLayout);
}

@Test
public void testGetPartitionNames()
        throws Exception
{
    ConnectorTableHandle tableHandle = getTableHandle(tablePartitionFormat);
    List<ConnectorTableLayoutResult> tableLayoutResults = Metadata.getTableLayouts(newSession(),tableLayout);
}

@Test
public void testGetPartitionNamesUnpartitioned()
        throws Exception
{
    ConnectorTableHandle tableHandle = getTableHandle(tableunpartitioned);
    List<ConnectorTableLayoutResult> tableLayoutResults = Metadata.getTableLayouts(newSession(),Optional.empty());
    assertEquals(getAllPartitions(getonlyElement(tableLayoutResults).getTableLayout().getHandle()).size(),1);
    assertExpectedTableLayout(getonlyElement(tableLayoutResults).getTableLayout(),unpartitionedTableLayout);
}

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

    ConnectorTableHandle tableHandle = getTableHandle(tablePartitionFormat);
    List<ConnectorTableLayoutResult> tableLayoutResults = Metadata.getTableLayouts(session,Optional.empty());
    ConnectorSplitSource splitSource = splitManager.getSplits(session,getonlyElement(tableLayoutResults).getTableLayout().getHandle());

    assertEquals(getSplitCount(splitSource),partitionCount);
}

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

    ConnectorTableHandle tableHandle = getTableHandle(tableunpartitioned);
    List<ConnectorTableLayoutResult> tableLayoutResults = Metadata.getTableLayouts(session,1);
}

@Test
public void testGetPartitionTableOffline()
        throws Exception
{
    ConnectorTableHandle tableHandle = getTableHandle(tableOffline);
    try {
        Metadata.getTableLayouts(newSession(),Optional.empty());
        fail("expected TableOfflineException");
    }
    catch (TableOfflineException e) {
        assertEquals(e.getTableName(),tableOffline);
    }
}

private MaterializedResult readTable(
        ConnectorTableHandle tableHandle,List<ColumnHandle> columnHandles,TupleDomain<ColumnHandle> tupleDomain,OptionalInt expectedSplitCount,Optional<HiveStorageFormat> expectedStorageFormat)
        throws Exception
{
    List<ConnectorTableLayoutResult> tableLayoutResults = Metadata.getTableLayouts(session,Optional.empty());
    ConnectorTableLayoutHandle layoutHandle = getonlyElement(tableLayoutResults).getTableLayout().getHandle();
    List<ConnectorSplit> splits = getAllSplits(splitManager.getSplits(session,layoutHandle));
    if (expectedSplitCount.isPresent()) {
        assertEquals(splits.size(),expectedSplitCount.getAsInt());
    }

    ImmutableList.Builder<MaterializedRow> allRows = ImmutableList.builder();
    for (ConnectorSplit split : splits) {
        try (ConnectorPageSource pageSource = pageSourceProvider.createPageSource(session,columnHandles)) {
            if (expectedStorageFormat.isPresent()) {
                assertPageSourceType(pageSource,expectedStorageFormat.get());
            }
            MaterializedResult result = materializeSourceDataStream(session,getTypes(columnHandles));
            allRows.addAll(result.getMaterializedRows());
        }
    }
    return new MaterializedResult(allRows.build(),getTypes(columnHandles));
}

private List<ConnectorSplit> getAllSplits(ConnectorTableHandle tableHandle,TupleDomain<ColumnHandle> tupleDomain)
        throws InterruptedException
{
    ConnectorSession session = newSession();
    List<ConnectorTableLayoutResult> tableLayoutResults = Metadata.getTableLayouts(session,Optional.empty());
    ConnectorTableLayoutHandle layoutHandle = getonlyElement(tableLayoutResults).getTableLayout().getHandle();
    return getAllSplits(splitManager.getSplits(session,layoutHandle));
}

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

@Override
public ConnectorTableLayout getTableLayout(ConnectorSession session,ConnectorTableLayoutHandle handle) {
  SpreadsheetTableLayoutHandle layout = (SpreadsheetTableLayoutHandle) handle;
  List<ConnectorTableLayoutResult> tableLayouts = getTableLayouts(session,Optional.empty());
  ConnectorTableLayoutResult connectorTableLayoutResult = tableLayouts.get(0);
  return connectorTableLayoutResult.getTableLayout();
}

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

@Override
protected Void visitAddColumn(AddColumn node,Integer indent)
{
    builder.append("ALTER TABLE ")
            .append(node.getName())
            .append(" ADD COLUMN ")
            .append(node.getColumn().getName())
            .append(" ")
            .append(node.getColumn().getType());

    return null;
}

@Override
public CompletableFuture<?> execute(AddColumn statement,TransactionManager transactionManager,Metadata Metadata,AccessControl accessControl,QueryStateMachine stateMachine)
{
    Session session = stateMachine.getSession();
    QualifiedobjectName tableName = createQualifiedobjectName(session,statement,statement.getName());
    Optional<TableHandle> tableHandle = Metadata.getTableHandle(session,tableName);
    if (!tableHandle.isPresent()) {
        throw new SemanticException(MISSING_TABLE,"Table '%s' does not exist",tableName);
    }

    accessControl.checkCanAddColumns(session.getrequiredTransactionId(),session.getIdentity(),tableName);

    Map<String,ColumnHandle> columnHandles = Metadata.getColumnHandles(session,tableHandle.get());

    TableElement element = statement.getColumn();
    Type type = Metadata.getType(parseTypeSignature(element.getType()));
    if ((type == null) || type.equals(UNKNowN)) {
        throw new SemanticException(TYPE_MISMATCH,element,"UnkNown type for column '%s' ",element.getName());
    }
    if (columnHandles.containsKey(element.getName())) {
        throw new SemanticException(COLUMN_ALREADY_EXISTS,"Column '%s' already exists",element.getName());
    }

    Metadata.addColumn(session,tableHandle.get(),new ColumnMetadata(element.getName(),type,false));

    return completedFuture(null);
}

@Override
protected Void visitAddColumn(AddColumn node,Integer indent)
{
    builder.append("ALTER TABLE ")
            .append(node.getName())
            .append(" ADD COLUMN ")
            .append(node.getColumn().getName())
            .append(" ")
            .append(node.getColumn().getType());

    return null;
}

@Override
protected Void visitAddColumn(AddColumn node,Integer indent)
{
    builder.append("ALTER TABLE ")
            .append(node.getName())
            .append(" ADD COLUMN ")
            .append(node.getColumn().getName())
            .append(" ")
            .append(node.getColumn().getType());

    return null;
}

@Override
public Node visitAddColumn(sqlbaseParser.AddColumnContext context)
{
    return new AddColumn(getLocation(context),getQualifiedname(context.qualifiedname()),(TableElement) visit(context.tableElement()));
}

@Test
public void testAddColumn()
        throws Exception
{
    assertStatement("ALTER TABLE foo.t ADD COLUMN c bigint",new AddColumn(Qualifiedname.of("foo","t"),new TableElement("c","bigint")));
}

@Test
public void testLimitAll()
{
    Query valuesQuery = query(values(
            row(new LongLiteral("1"),new StringLiteral("1")),row(new LongLiteral("2"),new StringLiteral("2"))));

    assertStatement("SELECT * FROM (VALUES (1,'1'),(2,'2')) LIMIT ALL",simpleQuery(selectList(new AllColumns()),subquery(valuesQuery),Optional.empty(),ImmutableList.of(),Optional.of("ALL")));
}

@Test
public void testWith()
        throws Exception
{
    assertStatement("WITH a (t,u) AS (SELECT * FROM x),b AS (SELECT * FROM y) TABLE z",new Query(Optional.of(new With(false,ImmutableList.of(
                    new WithQuery("a",table(Qualifiedname.of("x"))),ImmutableList.of("t","u")),new WithQuery("b",table(Qualifiedname.of("y"))),null)))),new Table(Qualifiedname.of("z")),Optional.<String>empty(),Optional.<Approximate>empty()));

    assertStatement("WITH RECURSIVE a AS (SELECT * FROM x) TABLE y",new Query(Optional.of(new With(true,new Table(Qualifiedname.of("y")),Optional.<Approximate>empty()));
}

@Test
public void testExplain()
        throws Exception
{
    assertStatement("EXPLAIN SELECT * FROM t",new Explain(simpleQuery(selectList(new AllColumns()),table(Qualifiedname.of("t"))),ImmutableList.of()));
    assertStatement("EXPLAIN (TYPE LOGICAL) SELECT * FROM t",new Explain(
                    simpleQuery(selectList(new AllColumns()),ImmutableList.of(new ExplainType(ExplainType.Type.LOGICAL))));
    assertStatement("EXPLAIN (TYPE LOGICAL,FORMAT TEXT) SELECT * FROM t",ImmutableList.of(
                            new ExplainType(ExplainType.Type.LOGICAL),new Explainformat(Explainformat.Type.TEXT))));
}

@Test
public void testUnnest()
        throws Exception
{
    assertStatement("SELECT * FROM t CROSS JOIN UNnesT(a)",simpleQuery(
                    selectList(new AllColumns()),new Join(
                            Join.Type.CROSS,new Table(Qualifiedname.of("t")),new Unnest(ImmutableList.of(new QualifiednameReference(Qualifiedname.of("a"))),false),Optional.empty())));
    assertStatement("SELECT * FROM t CROSS JOIN UNnesT(a) WITH ORDINALITY",true),Optional.empty())));
}

@Override
protected String visitAllColumns(AllColumns node,Void context)
{
    if (node.getPrefix().isPresent()) {
        return node.getPrefix().get() + ".*";
    }

    return "*";
}

@Override
protected String visitAllColumns(AllColumns node,StackableAstVisitorContext<Integer> indent)
{
    if (node.getPrefix().isPresent()) {
        return node.getPrefix().get() + ".*";
    }

    return "*";
}

@Override
protected Void visitAllColumns(AllColumns node,Integer context)
{
    builder.append(node.toString());

    return null;
}

@Override
protected RelationType visitShowCatalogs(ShowCatalogs node,AnalysisContext context)
{
    List<Expression> rows = Metadata.getCatalogNames().keySet().stream()
            .map(name -> row(new StringLiteral(name)))
            .collect(toList());

    Query query = simpleQuery(
            selectList(new AllColumns()),aliased(new Values(rows),"catalogs",ImmutableList.of("Catalog")));

    return process(query,context);
}

private RelationType computeOutputDescriptor(QuerySpecification node,RelationType inputTupleDescriptor)
{
    ImmutableList.Builder<Field> outputFields = ImmutableList.builder();

    for (SelectItem item : node.getSelect().getSelectItems()) {
        if (item instanceof AllColumns) {
            // expand * and T.*
            Optional<Qualifiedname> starPrefix = ((AllColumns) item).getPrefix();

            for (Field field : inputTupleDescriptor.resolveFieldsWithPrefix(starPrefix)) {
                outputFields.add(Field.newUnqualified(field.getName(),field.getType()));
            }
        }
        else if (item instanceof SingleColumn) {
            SingleColumn column = (SingleColumn) item;
            Expression expression = column.getExpression();

            Optional<String> alias = column.getAlias();
            if (!alias.isPresent()) {
                Qualifiedname name = null;
                if (expression instanceof QualifiednameReference) {
                    name = ((QualifiednameReference) expression).getName();
                }
                else if (expression instanceof DereferenceExpression) {
                    name = DereferenceExpression.getQualifiedname((DereferenceExpression) expression);
                }
                if (name != null) {
                    alias = Optional.of(getLast(name.getoriginalParts()));
                }
            }

            outputFields.add(Field.newUnqualified(alias,analysis.getType(expression))); // Todo don't use analysis as a side-channel. Use outputExpressions to look up the type
        }
        else {
            throw new IllegalArgumentException("Unsupported SelectItem type: " + item.getClass().getName());
        }
    }

    return new RelationType(outputFields.build());
}

@Override
protected Void visitAllColumns(AllColumns node,Integer context)
{
    builder.append(node.toString());

    return null;
}

@Override
public Node visitSelectAll(sqlbaseParser.SelectAllContext context)
{
    if (context.qualifiedname() != null) {
        return new AllColumns(getLocation(context),getQualifiedname(context.qualifiedname()));
    }

    return new AllColumns(getLocation(context));
}

@Override
protected String visitAllColumns(AllColumns node,Boolean unmangleNames)
{
    if (node.getPrefix().isPresent()) {
        return node.getPrefix().get() + ".*";
    }

    return "*";
}

@Test
public void testValues()
{
    Query valuesQuery = query(values(
            row(new StringLiteral("a"),new LongLiteral("1"),new DoubleLiteral("2.2")),row(new StringLiteral("b"),new LongLiteral("2"),new DoubleLiteral("3.3"))));

    assertStatement("VALUES ('a',1,2.2),('b',2,3.3)",valuesQuery);

    assertStatement("SELECT * FROM (VALUES ('a',3.3))",subquery(valuesQuery)));
}

@Test
public void testInsertInto()
        throws Exception
{
    Qualifiedname table = Qualifiedname.of("a");
    Query query = simpleQuery(selectList(new AllColumns()),table(Qualifiedname.of("t")));

    assertStatement("INSERT INTO a SELECT * FROM t",new Insert(table,query));

    assertStatement("INSERT INTO a (c1,c2) SELECT * FROM t",Optional.of(ImmutableList.of("c1","c2")),query));
}

@Test
public void testCreateView()
        throws Exception
{
    assertStatement("CREATE VIEW a AS SELECT * FROM t",new CreateView(
            Qualifiedname.of("a"),false));

    assertStatement("CREATE OR REPLACE VIEW a AS SELECT * FROM t",true));
}

@Test
public void testImplicitJoin()
        throws Exception
{
    assertStatement("SELECT * FROM a,b",new Join(Join.Type.IMPLICIT,new Table(Qualifiedname.of("a")),new Table(Qualifiedname.of("b")),Optional.<JoinCriteria>empty())));
}

@Test
public void testJoinPrecedence()
{
    assertStatement("SELECT * FROM a CROSS JOIN b LEFT JOIN c ON true",new Join(
                            Join.Type.LEFT,new Join(
                                    Join.Type.CROSS,Optional.empty()
                            ),new Table(Qualifiedname.of("c")),Optional.of(new JoinOn(BooleanLiteral.TRUE_LIteraL)))));
    assertStatement("SELECT * FROM a CROSS JOIN b NATURAL JOIN c CROSS JOIN d NATURAL JOIN e",new Join(
                            Join.Type.INNER,new Join(
                                            Join.Type.INNER,new Join(
                                                    Join.Type.CROSS,Optional.empty()
                                            ),Optional.of(new NaturalJoin())),new Table(Qualifiedname.of("d")),new Table(Qualifiedname.of("e")),Optional.of(new NaturalJoin()))));
}

@Override
protected Void visitAllColumns(AllColumns node,Integer context)
{
    builder.append(node.toString());

    return null;
}

@Override
protected String visitAllColumns(AllColumns node,Boolean unmangleNames)
{
    if (node.getPrefix().isPresent()) {
        return node.getPrefix().get() + ".*";
    }

    return "*";
}

private List<FieldOrExpression> analyzeSelect(QuerySpecification node,RelationType tupleDescriptor,AnalysisContext context)
{
    ImmutableList.Builder<FieldOrExpression> outputExpressionBuilder = ImmutableList.builder();

    for (SelectItem item : node.getSelect().getSelectItems()) {
        if (item instanceof AllColumns) {
            // expand * and T.*
            Optional<Qualifiedname> starPrefix = ((AllColumns) item).getPrefix();

            List<Field> fields = tupleDescriptor.resolveFieldsWithPrefix(starPrefix);
            if (fields.isEmpty()) {
                if (starPrefix.isPresent()) {
                    throw new SemanticException(MISSING_TABLE,item,"Table '%s' not found",starPrefix.get());
                }
                else {
                    throw new SemanticException(WILDCARD_WITHOUT_FROM,"SELECT * not allowed in queries without FROM clause");
                }
            }

            for (Field field : fields) {
                int fieldindex = tupleDescriptor.indexOf(field);
                outputExpressionBuilder.add(new FieldOrExpression(fieldindex));

                if (node.getSelect().isdistinct() && !field.getType().isComparable()) {
                    throw new SemanticException(TYPE_MISMATCH,node.getSelect(),"disTINCT can only be applied to comparable types (actual: %s)",field.getType());
                }
            }
        }
        else if (item instanceof SingleColumn) {
            SingleColumn column = (SingleColumn) item;
            ExpressionAnalysis expressionAnalysis = analyzeExpression(column.getExpression(),tupleDescriptor,context);
            analysis.recordSubqueries(node,expressionAnalysis);
            outputExpressionBuilder.add(new FieldOrExpression(column.getExpression()));

            Type type = expressionAnalysis.getType(column.getExpression());
            if (node.getSelect().isdistinct() && !type.isComparable()) {
                throw new SemanticException(TYPE_MISMATCH,"disTINCT can only be applied to comparable types (actual: %s): %s",type,column.getExpression());
            }
        }
        else {
            throw new IllegalArgumentException("Unsupported SelectItem type: " + item.getClass().getName());
        }
    }

    ImmutableList<FieldOrExpression> result = outputExpressionBuilder.build();
    analysis.setoutputExpressions(node,result);

    return result;
}