private Object getobject(Literal literal){
    Object value = null;
    if(literal instanceof LongLiteral) value = ((LongLiteral)literal).getValue();
    else if(literal instanceof BooleanLiteral) value = ((BooleanLiteral)literal).getValue();
    else if(literal instanceof DoubleLiteral) value = ((DoubleLiteral)literal).getValue();
    else if(literal instanceof StringLiteral) value = ((StringLiteral)literal).getValue();
    else if(literal instanceof TimeLiteral) value = ((TimeLiteral)literal).getValue();
    else if(literal instanceof TimestampLiteral) value = ((TimestampLiteral)literal).getValue();
    return value;
}

@Override
protected RowExpression visitTimeLiteral(TimeLiteral node,Void context)
{
    long value;
    if (types.get(node).equals(TIME_WITH_TIME_ZONE)) {
        value = parseTimeWithTimeZone(node.getValue());
    }
    else {
        // parse in time zone of client
        value = parseTimeWithoutTimeZone(timeZoneKey,node.getValue());
    }
    return constant(value,types.get(node));
}

@Override
protected Type visitTimeLiteral(TimeLiteral node,StackableAstVisitorContext<AnalysisContext> context)
{
    Type type;
    if (timeHasTimeZone(node.getValue())) {
        type = TIME_WITH_TIME_ZONE;
    }
    else {
        type = TIME;
    }
    expressionTypes.put(node,type);
    return type;
}

@Override
public Node visitTypeConstructor(sqlbaseParser.TypeConstructorContext context)
{
    String type = context.identifier().getText();
    String value = unquote(context.STRING().getText());

    if (type.equalsIgnoreCase("time")) {
        return new TimeLiteral(getLocation(context),value);
    }
    if (type.equalsIgnoreCase("timestamp")) {
        return new TimestampLiteral(getLocation(context),value);
    }

    return new GenericLiteral(getLocation(context),type,value);
}

@Test
public void testLiterals()
        throws Exception
{
    assertExpression("TIME" + " 'abc'",new TimeLiteral("abc"));
    assertExpression("TIMESTAMP" + " 'abc'",new TimestampLiteral("abc"));
    assertExpression("INTERVAL '33' day",new IntervalLiteral("33",Sign.POSITIVE,IntervalField.DAY,Optional.empty()));
    assertExpression("INTERVAL '33' day to second",Optional.of(IntervalField.SECOND)));
}

@Override
protected String visitTimeLiteral(TimeLiteral node,Void context)
{
    return "TIME '" + node.getValue() + "'";
}

@Override
protected String visitTimeLiteral(TimeLiteral node,StackableAstVisitorContext<Integer> indent)
{
    return "TIME '" + node.getValue() + "'";
}

@Override
protected Long visitTimeLiteral(TimeLiteral node,ConnectorSession session)
{
    return parseTime(session.getTimeZoneKey(),node.getValue());
}

@Override
protected String visitTimeLiteral(TimeLiteral node,Boolean unmangleNames)
{
    return "TIME '" + node.getValue() + "'";
}

@Test
public void testTime()
        throws Exception
{
    assertExpression("TIME '03:04:05'",new TimeLiteral("03:04:05"));
}

@Override
protected String visitTimeLiteral(TimeLiteral node,Boolean unmangleNames)
{
    return "TIME '" + node.getValue() + "'";
}

@Override
protected List<QuerySource> visitRelation(Relation node,QueryState state){
    if(node instanceof Join){
        return node.accept(this,state);
    }else if( node instanceof SampledRelation){
        state.addException("Sampled relations are not supported");
        return null;
    }else if( node instanceof AliasedRelation){
        AliasedRelation ar = (AliasedRelation)node;
        state.setkeyvalue("table_alias",ar.getAlias());
        List<QuerySource> relations = ar.getRelation().accept(this,state);
        for(QuerySource rr : relations) rr.setAlias(ar.getAlias());
        return relations;
    }else if( node instanceof QueryBody){
        return node.accept(this,state);
    }else{
        state.addException("Unable to parse node because it has an unkNown type :"+node.getClass());
        return null;
    }
}

@Override
protected RelationType visitAliasedRelation(AliasedRelation relation,AnalysisContext context)
{
    RelationType child = process(relation.getRelation(),context);

    // todo this check should be inside of TupleDescriptor.withAlias,but the exception needs the node object
    if (relation.getColumnNames() != null) {
        int totalColumns = child.getVisibleFieldCount();
        if (totalColumns != relation.getColumnNames().size()) {
            throw new SemanticException(MISMATCHED_COLUMN_ALIASES,relation,"Column alias list has %s entries but '%s' has %s columns available",relation.getColumnNames().size(),relation.getAlias(),totalColumns);
        }
    }

    RelationType descriptor = child.withAlias(relation.getAlias(),relation.getColumnNames());

    analysis.setoutputDescriptor(relation,descriptor);
    return descriptor;
}

@Override
protected Void visitAliasedRelation(AliasedRelation node,Integer indent)
{
    process(node.getRelation(),indent);

    builder.append(' ')
            .append(formatName(node.getAlias()));
    appendaliasColumns(builder,node.getColumnNames());

    return null;
}

@Override
protected RelationPlan visitAliasedRelation(AliasedRelation node,Void context)
{
    RelationPlan subPlan = process(node.getRelation(),context);

    RelationType outputDescriptor = analysis.getoutputDescriptor(node);

    return new RelationPlan(subPlan.getRoot(),outputDescriptor,subPlan.getoutputSymbols(),subPlan.getSampleWeight());
}

@Override
protected Void visitAliasedRelation(AliasedRelation node,indent);

    builder.append(' ')
            .append(node.getAlias());

    appendaliasColumns(builder,node.getColumnNames());

    return null;
}

@Override
public Node visitAliasedRelation(sqlbaseParser.AliasedRelationContext context)
{
    Relation child = (Relation) visit(context.relationPrimary());

    if (context.identifier() == null) {
        return child;
    }

    return new AliasedRelation(getLocation(context),child,context.identifier().getText(),getColumnAliases(context.columnAliases()));
}

@Override
protected Void visitAliasedRelation(AliasedRelation node,node.getColumnNames());

    return null;
}

public static Relation aliased(Relation relation,String alias,List<String> columnAliases)
{
    return new AliasedRelation(relation,alias,columnAliases);
}

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

@Override
public PhysicalOperation visitProject(ProjectNode node,LocalExecutionPlanContext context)
{
    PlanNode sourceNode;
    Expression filterExpression;
    if (node.getSource() instanceof FilterNode) {
        FilterNode filterNode = (FilterNode) node.getSource();
        sourceNode = filterNode.getSource();
        filterExpression = filterNode.getPredicate();
    }
    else {
        sourceNode = node.getSource();
        filterExpression = BooleanLiteral.TRUE_LIteraL;
    }

    List<Symbol> outputSymbols = node.getoutputSymbols();

    return visitScanFilterandProject(context,node.getId(),sourceNode,filterExpression,node.getAssignments(),outputSymbols);
}

private PlanNode rewriteFilterSource(FilterNode filterNode,PlanNode source,Symbol rowNumberSymbol,int upperBound)
{
    ExtractionResult extractionResult = fromPredicate(Metadata,session,filterNode.getPredicate(),types);
    TupleDomain<Symbol> tupleDomain = extractionResult.getTupleDomain();

    if (!isEqualRange(tupleDomain,rowNumberSymbol,upperBound)) {
        return new FilterNode(filterNode.getId(),source,filterNode.getPredicate());
    }

    // Remove the row number domain because it is absorbed into the node
    Map<Symbol,Domain> newDomains = tupleDomain.getDomains().get().entrySet().stream()
            .filter(entry -> !entry.getKey().equals(rowNumberSymbol))
            .collect(toMap(Map.Entry::getKey,Map.Entry::getValue));

    // Construct a new predicate
    TupleDomain<Symbol> newTupleDomain = TupleDomain.withColumnDomains(newDomains);
    Expression newPredicate = ExpressionUtils.combineConjuncts(
            extractionResult.getRemainingExpression(),toPredicate(newTupleDomain));

    if (newPredicate.equals(BooleanLiteral.TRUE_LIteraL)) {
        return source;
    }
    return new FilterNode(filterNode.getId(),newPredicate);
}

/**
 * Extracts the literal value from an expression (if expression is supported)
 * @param expression
 * @param state
 * @return a Long,Boolean,Double or String object
 */
private Object getLiteralValue(Expression expression,QueryState state){
    if(expression instanceof LongLiteral) return ((LongLiteral)expression).getValue();
    else if(expression instanceof BooleanLiteral) return ((BooleanLiteral)expression).getValue();
    else if(expression instanceof DoubleLiteral) return ((DoubleLiteral)expression).getValue();
    else if(expression instanceof StringLiteral) return ((StringLiteral)expression).getValue();
    else if(expression instanceof ArithmeticUnaryExpression){
        ArithmeticUnaryExpression unaryExp = (ArithmeticUnaryExpression)expression;
        Sign sign = unaryExp.getSign();
        Number num = (Number)getLiteralValue(unaryExp.getValue(),state);
        if(sign == Sign.MINUS){
            if(num instanceof Long) return -1*num.longValue();
            else if(num instanceof Double) return -1*num.doubleValue();
            else {
                state.addException("Unsupported numeric literal expression encountered : "+num.getClass());
                return null;
            }
        }
        return num;
    } else if(expression instanceof FunctionCall){
        FunctionCall fc = (FunctionCall)expression;
        if(fc.getName().toString().equals("Now")) return new Date();
        else state.addException("Function '"+fc.getName()+"' is not supported");
    }else if(expression instanceof CurrentTime){
        CurrentTime ct = (CurrentTime)expression;
        if(ct.getType() == CurrentTime.Type.DATE) return new LocalDate().toDate();
        else if(ct.getType() == CurrentTime.Type.TIME) return new Date(new LocalTime(DateTimeZone.UTC).getMillisOfDay());
        else if(ct.getType() == CurrentTime.Type.TIMESTAMP) return new Date();
        else if(ct.getType() == CurrentTime.Type.LOCALTIME) return new Date(new LocalTime(DateTimeZone.UTC).getMillisOfDay());
        else if(ct.getType() == CurrentTime.Type.LOCALTIMESTAMP) return new Date();
        else state.addException("CurrentTime function '"+ct.getType()+"' is not supported");

    }else state.addException("Literal type "+expression.getClass().getSimpleName()+" is not supported");
    return null;
}

private Object getobject(Literal literal){
    Object value = null;
    if(literal instanceof LongLiteral) value = ((LongLiteral)literal).getValue();
    else if(literal instanceof BooleanLiteral) value = ((BooleanLiteral)literal).getValue();
    else if(literal instanceof DoubleLiteral) value = ((DoubleLiteral)literal).getValue();
    else if(literal instanceof StringLiteral) value = ((StringLiteral)literal).getValue();
    else if(literal instanceof TimeLiteral) value = ((TimeLiteral)literal).getValue();
    else if(literal instanceof TimestampLiteral) value = ((TimestampLiteral)literal).getValue();
    return value;
}

private Object getLiteralValue(Expression expression) throws sqlException{
    if(expression instanceof LongLiteral) return ((LongLiteral)expression).getValue();
    else if(expression instanceof BooleanLiteral) return ((BooleanLiteral)expression).getValue();
    else if(expression instanceof DoubleLiteral) return ((DoubleLiteral)expression).getValue();
    else if(expression instanceof StringLiteral) return ((StringLiteral)expression).getValue();
    throw new sqlException("Unsupported literal type: "+expression);
}

@Override
public PlanNode visitFilter(FilterNode node,RewriteContext<Void> context)
{
    PlanNode source = context.rewrite(node.getSource());
    Expression canonicalized = canonicalizeExpression(node.getPredicate());
    if (canonicalized.equals(BooleanLiteral.TRUE_LIteraL)) {
        return source;
    }
    return new FilterNode(node.getId(),canonicalized);
}

@Override
public PlanNode optimize(PlanNode plan,Session session,Map<Symbol,Type> types,SymbolAllocator symbolAllocator,PlanNodeIdAllocator idAllocator)
{
    requireNonNull(plan,"plan is null");
    requireNonNull(session,"session is null");
    requireNonNull(types,"types is null");
    requireNonNull(idAllocator,"idAllocator is null");

    return SimplePlanRewriter.rewriteWith(new Rewriter(symbolAllocator,idAllocator,Metadata,sqlParser,session),plan,BooleanLiteral.TRUE_LIteraL);
}

@Override
public PlanNode visitPlan(PlanNode node,RewriteContext<Expression> context)
{
    PlanNode rewrittenNode = context.defaultRewrite(node,BooleanLiteral.TRUE_LIteraL);
    if (!context.get().equals(BooleanLiteral.TRUE_LIteraL)) {
        // Drop in a FilterNode b/c we cannot push our predicate down any further
        rewrittenNode = new FilterNode(idAllocator.getNextId(),rewrittenNode,context.get());
    }
    return rewrittenNode;
}

@Override
public PlanNode visitTableScan(TableScanNode node,RewriteContext<Expression> context)
{
    Expression predicate = simplifyExpression(context.get());

    if (!BooleanLiteral.TRUE_LIteraL.equals(predicate)) {
        return new FilterNode(idAllocator.getNextId(),node,predicate);
    }

    return node;
}

@Override
public PlanNode visitTableScan(TableScanNode node,RewriteContext<Void> context)
{
    if (node.getLayout().isPresent()) {
        return node;
    }

    return planTableScan(node,BooleanLiteral.TRUE_LIteraL);
}

@Test
public void testJoinPrecedence()
{
    assertStatement("SELECT * FROM a CROSS JOIN b LEFT JOIN c ON true",simpleQuery(
                    selectList(new AllColumns()),new Join(
                            Join.Type.LEFT,new Join(
                                    Join.Type.CROSS,new Table(Qualifiedname.of("a")),new Table(Qualifiedname.of("b")),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 String visitBooleanLiteral(BooleanLiteral node,Void context)
{
    return String.valueOf(node.getValue());
}

@Override
protected String visitBooleanLiteral(BooleanLiteral node,StackableAstVisitorContext<Integer> indent)
{
    return String.valueOf(node.getValue());
}

@Override
protected IComparison visitExpression(Expression node,QueryState state) {
    if( node instanceof LogicalBinaryExpression){
        LogicalBinaryExpression boolExp = (LogicalBinaryExpression)node;
        IComparison left = boolExp.getLeft().accept(this,state);
        IComparison right = boolExp.getRight().accept(this,state);
        return new BooleanComparison(left,right,boolExp.getType() == Type.AND);
    }else if( node instanceof ComparisonExpression){
        ComparisonExpression compareExp = (ComparisonExpression)node;
        Column column = new SelectParser().visitExpression(compareExp.getLeft(),state);
        Column leftCol = state.getheading().getColumnByLabel(column.getLabel());
        if(leftCol == null){
            state.addException("Having reference "+column+" not found in SELECT clause");
            return null;
        }
        // right hand side is a concrete literal to compare with 
        if(compareExp.getRight() instanceof Literal){
            Object value;
            if(compareExp.getRight() instanceof LongLiteral) value = ((LongLiteral)compareExp.getRight()).getValue();
            else if(compareExp.getRight() instanceof BooleanLiteral) value = ((BooleanLiteral)compareExp.getRight()).getValue();
            else if(compareExp.getRight() instanceof DoubleLiteral) value = ((DoubleLiteral)compareExp.getRight()).getValue();
            else if(compareExp.getRight() instanceof StringLiteral) value = ((StringLiteral)compareExp.getRight()).getValue();
            else {
                state.addException("Unable to get value from "+compareExp.getRight());
                return null;
            }
            return new SimpleComparison(leftCol,compareExp.getType(),(Number)value);

            // right hand side refers to another column     
        } else if(compareExp.getRight() instanceof DereferenceExpression || compareExp.getRight() instanceof QualifiednameReference){
            String col2;
            if(compareExp.getLeft() instanceof DereferenceExpression){
                // parse columns like 'reference.field'
                col2 = SelectParser.visitDereferenceExpression((DereferenceExpression)compareExp.getRight());
            }else{
                col2 = ((QualifiednameReference)compareExp.getRight()).getName().toString();
            }
            col2 = heading.findOriginal(state.originalsql(),col2,"having.+","\\W");
            Column rightCol = state.getheading().getColumnByLabel(col2);
            if(rightCol == null){
                state.addException("column "+col2+" not found in SELECT clause");
                return null;
            }
            return new SimpleComparison(leftCol,rightCol);
        }else { // unkNown right hand side so
            state.addException("Unable to get value from "+compareExp.getRight());
            return null;
        }

    }else if( node instanceof NotExpression){
        state.addException("NOT is currently not supported,use '<>' instead");
    }else{
        state.addException("Unable to parse "+node+" ("+node.getClass().getName()+") is not a supported expression");
    }
    return null;
}

@Override
protected RowExpression visitBooleanLiteral(BooleanLiteral node,Void context)
{
    return constant(node.getValue(),BOOLEAN);
}

@Override
protected Object visitBooleanLiteral(BooleanLiteral node,ConnectorSession session)
{
    return node.getValue();
}

@Override
protected Object visitBooleanLiteral(BooleanLiteral node,Object context)
{
    return node.equals(BooleanLiteral.TRUE_LIteraL);
}

@Override
public PlanWithProperties visitTableScan(TableScanNode node,Context context)
{
    return planTableScan(node,BooleanLiteral.TRUE_LIteraL,context);
}

private PlanWithProperties planTableScan(TableScanNode node,Expression predicate,Context context)
{
    // don't include non-deterministic predicates
    Expression deterministicPredicate = stripNonDeterministicConjuncts(predicate);

    DomainTranslator.ExtractionResult decomposedPredicate = DomainTranslator.fromPredicate(
            Metadata,deterministicPredicate,symbolAllocator.getTypes());

    TupleDomain<ColumnHandle> simplifiedConstraint = decomposedPredicate.getTupleDomain()
            .transform(node.getAssignments()::get)
            .intersect(node.getCurrentConstraint());

    Map<ColumnHandle,Symbol> assignments = ImmutableBiMap.copyOf(node.getAssignments()).inverse();

    Expression constraint = combineConjuncts(
            deterministicPredicate,DomainTranslator.toPredicate(node.getCurrentConstraint().transform(assignments::get)));

    // Layouts will be returned in order of the connector's preference
    List<TableLayoutResult> layouts = Metadata.getLayouts(
            session,node.getTable(),new Constraint<>(simplifiedConstraint,bindings -> !shouldPrune(constraint,bindings)),Optional.of(node.getoutputSymbols().stream()
                    .map(node.getAssignments()::get)
                    .collect(toImmutableSet())));

    if (layouts.isEmpty()) {
        return new PlanWithProperties(
                new ValuesNode(idAllocator.getNextId(),node.getoutputSymbols(),ImmutableList.of()),ActualProperties.undistributed());
    }

    // Filter out layouts that cannot supply all the required columns
    layouts = layouts.stream()
            .filter(layoutHasAllNeededOutputs(node))
            .collect(toList());
    checkState(!layouts.isEmpty(),"No usable layouts for %s",node);

    List<PlanWithProperties> possiblePlans = layouts.stream()
            .map(layout -> {
                TableScanNode tableScan = new TableScanNode(
                        node.getId(),Optional.of(layout.getLayout().getHandle()),simplifiedConstraint.intersect(layout.getLayout().getPredicate()),Optional.ofNullable(node.getoriginalConstraint()).orElse(predicate));

                PlanWithProperties result = new PlanWithProperties(tableScan,deriveProperties(tableScan,ImmutableList.of()));

                Expression resultingPredicate = combineConjuncts(
                        DomainTranslator.toPredicate(layout.getUnenforcedConstraint().transform(assignments::get)),stripDeterministicConjuncts(predicate),decomposedPredicate.getRemainingExpression());

                if (!BooleanLiteral.TRUE_LIteraL.equals(resultingPredicate)) {
                    return withDerivedProperties(
                            new FilterNode(idAllocator.getNextId(),result.getNode(),resultingPredicate),ImmutableList.of()));
                }

                return result;
            })
            .collect(toList());

    return pickPlan(possiblePlans,context);
}

private PlanNode planTableScan(TableScanNode node,Expression predicate)
{
    DomainTranslator.ExtractionResult decomposedPredicate = DomainTranslator.fromPredicate(
            Metadata,predicate,symbolAllocator.getTypes());

    TupleDomain<ColumnHandle> simplifiedConstraint = decomposedPredicate.getTupleDomain()
            .transform(node.getAssignments()::get)
            .intersect(node.getCurrentConstraint());

    List<TableLayoutResult> layouts = Metadata.getLayouts(
            session,bindings -> true),Optional.of(ImmutableSet.copyOf(node.getAssignments().values())));

    if (layouts.isEmpty()) {
        return new ValuesNode(idAllocator.getNextId(),ImmutableList.of());
    }

    TableLayoutResult layout = layouts.get(0);

    TableScanNode result = new TableScanNode(
            node.getId(),Optional.ofNullable(node.getoriginalConstraint()).orElse(predicate));

    Map<ColumnHandle,Symbol> assignments = ImmutableBiMap.copyOf(node.getAssignments()).inverse();
    Expression resultingPredicate = combineConjuncts(
            decomposedPredicate.getRemainingExpression(),DomainTranslator.toPredicate(layout.getUnenforcedConstraint().transform(assignments::get)));

    if (!BooleanLiteral.TRUE_LIteraL.equals(resultingPredicate)) {
        return new FilterNode(idAllocator.getNextId(),result,resultingPredicate);
    }

    return result;
}

@Override
public PlanNode visitTableScan(TableScanNode node,RewriteContext<Context> context)
{
    return planTableScan(node,context.get());
}

@Override
protected ExtractionResult visitBooleanLiteral(BooleanLiteral node,Boolean complement)
{
    boolean value = complement ? !node.getValue() : node.getValue();
    return new ExtractionResult(value ? TupleDomain.all() : TupleDomain.none(),TRUE_LIteraL);
}

@Override
protected Type visitBooleanLiteral(BooleanLiteral node,StackableAstVisitorContext<AnalysisContext> context)
{
    expressionTypes.put(node,BOOLEAN);
    return BOOLEAN;
}

@Test
public void testTableScan()
        throws Exception
{
    // Effective predicate is True if there is no effective predicate
    Map<Symbol,ColumnHandle> assignments = Maps.filterKeys(scanAssignments,Predicates.in(ImmutableList.of(A,B,C,D)));
    PlanNode node = new TableScanNode(
            newId(),DUAL_TABLE_HANDLE,ImmutableList.copyOf(assignments.keySet()),assignments,Optional.empty(),TupleDomain.all(),null);
    Expression effectivePredicate = EffectivePredicateExtractor.extract(node,TYPES);
    Assert.assertEquals(effectivePredicate,BooleanLiteral.TRUE_LIteraL);

    node = new TableScanNode(
            newId(),TupleDomain.none(),null);
    effectivePredicate = EffectivePredicateExtractor.extract(node,FALSE_LIteraL);

    node = new TableScanNode(
            newId(),TupleDomain.withColumnDomains(ImmutableMap.of(scanAssignments.get(A),Domain.singleValue(BIGINT,1L))),TYPES);
    Assert.assertEquals(normalizeConjuncts(effectivePredicate),normalizeConjuncts(equals(number(1L),AE)));

    node = new TableScanNode(
            newId(),TupleDomain.withColumnDomains(ImmutableMap.of(
                    scanAssignments.get(A),1L),scanAssignments.get(B),2L))),normalizeConjuncts(equals(number(2L),BE),equals(number(1L),BooleanLiteral.TRUE_LIteraL);
}

@Override
public Node visitBooleanValue(sqlbaseParser.BooleanValueContext context)
{
    return new BooleanLiteral(getLocation(context),context.getText());
}

@Override
protected String visitBooleanLiteral(BooleanLiteral node,Boolean unmangleNames)
{
    return String.valueOf(node.getValue());
}

@Override
protected String visitBooleanLiteral(BooleanLiteral node,Boolean unmangleNames)
{
    return String.valueOf(node.getValue());
}

@Override
public Type visitCast(Cast node,StackableAstVisitorContext<AnalysisContext> context)
{
    Type type = typeManager.getType(parseTypeSignature(node.getType()));
    if (type == null) {
        throw new SemanticException(TYPE_MISMATCH,node,"UnkNown type: " + node.getType());
    }

    if (type.equals(UNKNowN)) {
        throw new SemanticException(TYPE_MISMATCH,"UNKNowN is not a valid type");
    }

    Type value = process(node.getExpression(),context);
    if (!value.equals(UNKNowN)) {
        try {
            functionRegistry.getCoercion(value,type);
        }
        catch (OperatorNotFoundException e) {
            throw new SemanticException(TYPE_MISMATCH,"Cannot cast %s to %s",value,type);
        }
    }

    expressionTypes.put(node,type);
    return type;
}

@Override
protected RowExpression visitCast(Cast node,Void context)
{
    RowExpression value = process(node.getExpression(),context);

    if (node.isSafe()) {
        return call(tryCastSignature(types.get(node),value.getType()),types.get(node),value);
    }

    return call(castSignature(types.get(node),value);
}

@Override
public Object visitCast(Cast node,Object context)
{
    Object value = process(node.getExpression(),context);

    if (value instanceof Expression) {
        return new Cast((Expression) value,node.getType(),node.isSafe());
    }

    // hack!!! don't optimize CASTs for types that cannot be represented in the sql AST
    // Todo: this will not be an issue when we migrate to RowExpression tree for this,which allows arbitrary literals.
    if (optimize && !FunctionRegistry.isSupportedLiteralType(expressionTypes.get(node))) {
        return new Cast(toExpression(value,expressionTypes.get(node.getExpression())),node.isSafe());
    }

    if (value == null) {
        return null;
    }

    Type type = Metadata.getType(parseTypeSignature(node.getType()));
    if (type == null) {
        throw new IllegalArgumentException("Unsupported type: " + node.getType());
    }

    Signature operator = Metadata.getFunctionRegistry().getCoercion(expressionTypes.get(node.getExpression()),type);

    try {
        return invoke(session,Metadata.getFunctionRegistry().getScalarFunctionImplementation(operator),ImmutableList.of(value));
    }
    catch (RuntimeException e) {
        if (node.isSafe()) {
            return null;
        }
        throw e;
    }
}

@VisibleForTesting
@NotNull
public static Expression createFailureFunction(RuntimeException exception,Type type)
{
    requireNonNull(exception,"Exception is null");

    String failureInfo = JsonCodec.jsonCodec(FailureInfo.class).toJson(Failures.toFailure(exception).toFailureInfo());
    FunctionCall jsonParse = new FunctionCall(Qualifiedname.of("json_parse"),ImmutableList.of(new StringLiteral(failureInfo)));
    FunctionCall failureFunction = new FunctionCall(Qualifiedname.of("fail"),ImmutableList.of(jsonParse));

    return new Cast(failureFunction,type.getTypeSignature().toString());
}

@Override
public String visitCast(Cast node,Void context)
{
    return (node.isSafe() ? "TRY_CAST" : "CAST") +
            "(" + process(node.getExpression(),context) + " AS " + node.getType() + ")";
}

@Override
public String visitCast(Cast node,StackableAstVisitorContext<Integer> indent)
{
    return (node.isSafe() ? "TRY_CAST" : "CAST") +
            "(" + process(node.getExpression(),indent) + " AS " + node.getType() + ")";
}

@Override
protected Boolean visitCast(Cast node,Void context)
{
    return process(node.getExpression(),context);
}

@Override
public Node visitCast(sqlbaseParser.CastContext context)
{
    boolean isTryCast = context.TRY_CAST() != null;
    return new Cast(getLocation(context),(Expression) visit(context.expression()),getType(context.type()),isTryCast);
}

@Override
public String visitCast(Cast node,Boolean unmangleNames)
{
    return (node.isSafe() ? "TRY_CAST" : "CAST") +
            "(" + process(node.getExpression(),unmangleNames) + " AS " + node.getType() + ")";
}

private static void assertCast(String type,String expected)
{
    assertExpression("CAST(null AS " + type + ")",new Cast(new NullLiteral(),expected));
}

@Override
public String visitCast(Cast node,unmangleNames) + " AS " + node.getType() + ")";
}