/**
 * Creates a table based on a select on another type and inserts the data (currently not possible to
 * perform this action 'WITH NO DATA'.
 * @param sql
 * @param createAsSelect
 * @param index
 * @return the number of rows inserted
 * @throws sqlException
 */
public int execute(String sql,CreateTableAsSelect createAsSelect,String index) throws sqlException {
    if(!createAsSelect.isWithData()) throw new sqlException("Not yet possible to create table as select without data (create emtpy table,"
            + "insert data and delete it will have the same effect");
    // first create the index
    sqlParser parser = new sqlParser();
    int queryIdx = sql.toLowerCase().indexOf(" as ");
    try{
        String createsql = sql.substring(0,queryIdx)+" (_id String)" ;
        CreateTable create = (CreateTable)parser.createStatement(createsql);
        this.execute(createsql,create,index);

    }catch(sqlException sqle) {
        throw sqle;
    }catch(Exception e){
        throw new sqlException("Unable to create table due to: "+e.getMessage(),e);
    }

    // secondly add the documents from the query
    String insertsql = "INSERT INTO "+createAsSelect.getName().toString()+" "+sql.substring(queryIdx+4);
    Insert insert = (Insert)parser.createStatement(insertsql);
    int res = this.execute(insertsql,insert,index);
    this.statement.getConnection().getTypeMap(); // trigger a reload of the table&column set for the connection
    return res;
}

public InterpretedFilterFunction(
        Expression predicate,Map<Symbol,Type> symbolTypes,Integer> symbolToInputMappings,Metadata Metadata,sqlParser sqlParser,Session session)
{
    // pre-compute symbol -> input mappings and replace the corresponding nodes in the tree
    Expression rewritten = ExpressionTreeRewriter.rewriteWith(new SymbolToInputRewriter(symbolToInputMappings),predicate);

    // analyze expression so we can kNow the type of every expression in the tree
    ImmutableMap.Builder<Integer,Type> inputTypes = ImmutableMap.builder();
    for (Map.Entry<Symbol,Integer> entry : symbolToInputMappings.entrySet()) {
        inputTypes.put(entry.getValue(),symbolTypes.get(entry.getKey()));
    }
    IdentityHashMap<Expression,Type> expressionTypes = getExpressionTypesFromInput(session,Metadata,sqlParser,inputTypes.build(),rewritten);

    evaluator = ExpressionInterpreter.expressionInterpreter(rewritten,session,expressionTypes);
}

public InterpretedProjectionFunction(
        Expression expression,expression);

    // analyze expression so we can kNow the type of every expression in the tree
    ImmutableMap.Builder<Integer,rewritten);
    this.type = requireNonNull(expressionTypes.get(rewritten),"type is null");

    evaluator = ExpressionInterpreter.expressionInterpreter(rewritten,expressionTypes);
    InputReferenceExtractor inputReferenceExtractor = new InputReferenceExtractor();
    inputReferenceExtractor.process(rewritten,null);
    this.inputChannels = inputReferenceExtractor.getInputChannels();
    this.deterministic = DeterminismEvaluator.isDeterministic(expression);
}

public static ExpressionAnalysis analyzeExpressionsWithSymbols(
        Session session,Type> types,Iterable<? extends Expression> expressions)
{
    List<Field> fields = DependencyExtractor.extractUnique(expressions).stream()
            .map(symbol -> {
                Type type = types.get(symbol);
                checkArgument(type != null,"No type for symbol %s",symbol);
                return Field.newUnqualified(symbol.getName(),type);
            })
            .collect(toImmutableList());

    return analyzeExpressions(session,new RelationType(fields),expressions);
}

private static ExpressionAnalysis analyzeExpressions(
        Session session,RelationType tupleDescriptor,Iterable<? extends Expression> expressions)
{
    // expressions at this point can not have sub queries so deny all access checks
    // in the future,we will need a full access controller here to verify access to functions
    ExpressionAnalyzer analyzer = create(new Analysis(),new DenyAllAccessControl(),false);
    for (Expression expression : expressions) {
        analyzer.analyze(expression,tupleDescriptor,new AnalysisContext());
    }

    return new ExpressionAnalysis(
            analyzer.getExpressionTypes(),analyzer.getExpressionCoercions(),analyzer.getSubqueryInPredicates(),analyzer.getScalarsubqueries(),analyzer.getResolvednames().keySet());
}

public StatementAnalyzer(
        Analysis analysis,AccessControl accessControl,Session session,boolean experimentalSyntaxEnabled,Optional<QueryExplainer> queryExplainer)
{
    this.analysis = requireNonNull(analysis,"analysis is null");
    this.Metadata = requireNonNull(Metadata,"Metadata is null");
    this.sqlParser = requireNonNull(sqlParser,"sqlParser is null");
    this.accessControl = requireNonNull(accessControl,"accessControl is null");
    this.session = requireNonNull(session,"session is null");
    this.experimentalSyntaxEnabled = experimentalSyntaxEnabled;
    this.queryExplainer = requireNonNull(queryExplainer,"queryExplainer is null");
}

@Inject
public QueryExplainer(
        List<Planoptimizer> planoptimizers,Map<Class<? extends Statement>,DataDeFinitionTask<?>> dataDeFinitionTask,FeaturesConfig featuresConfig)
{
    this(planoptimizers,accessControl,dataDeFinitionTask,featuresConfig.isExperimentalSyntaxEnabled());
}

public static LocalExecutionPlanner createTestingPlanner()
{
    MetadataManager Metadata = MetadataManager.createTestMetadataManager();

    PageSourceManager pageSourceManager = new PageSourceManager();
    pageSourceManager.addConnectorPageSourceProvider("test",new TestingPageSourceProvider());
    return new LocalExecutionPlanner(
            Metadata,new sqlParser(),pageSourceManager,new IndexManager(),new PageSinkManager(),new MockExchangeClientsupplier(),new ExpressionCompiler(Metadata),new IndexJoinLookupStats(),new CompilerConfig(),new TaskManagerConfig());
}

@Test
public void testSimpleQuery()
        throws Exception {
    Statement statement = new sqlParser().createStatement("select * from testcollection");

    assertEquals(formatQuery(statement,name -> "dummy",'"').trim(),"SELECT *\n" +
            "   FROM\n" +
            "     dummy");
}

@Test(enabled = false)
    public void testName()
            throws Exception {
        String s = "select count(*) from test where ali = 5";
        Statement statement = new sqlParser().createStatement(s);
        Analysis analysis = new Analysis(statement,ImmutableList.of(),false);
        Session build = Session.builder(new SessionPropertyManager())
                .setQueryId(QueryId.valueOf("test"))
                .setCatalog("test")
                .setCatalog("test")
                .setCatalog("test")
                .build();
        QueryPlanner queryPlanner = new QueryPlanner(analysis,new SymbolAllocator(),new PlanNodeIdAllocator(),null,build);
        RelationPlan plan = queryPlanner.plan((Query) statement);
//        EffectivePredicateExtractor.extract(plan.getRoot(),ImmutableMap.of(new Symbol("ali"),BIGINT));
        EffectivePredicateExtractor.extract(plan.getRoot());
    }

/**
 * Executes the {@link BulkRequest} being hold by this state.
 * @return an integer indicator for each executed request: Statement.SUCCESS_NO_INFO for success,* else Statement.EXECUTE_Failed)
 */
public int[] executeBulk(){
    int[] result = new int[bulkList.size()];
    sqlParser parser = new sqlParser();
    for(int i=0; i<bulkList.size(); i++) try{
        String sql = bulkList.get(i);
        com.facebook.presto.sql.tree.Statement st = parser.createStatement(sql);
        if(st instanceof DropTable){
            this.execute(sql,(DropTable)st);
        }else if(st instanceof DropView){
            this.execute(sql,(DropView)st);
        }else if(st instanceof CreateTable){
            this.execute(sql,(CreateTable)st,this.statement.getConnection().getSchema());
        }else if(st instanceof CreateTableAsSelect){
            this.execute(sql,(CreateTableAsSelect)st,this.statement.getConnection().getSchema());
        }else if(st instanceof CreateView){
            this.execute(sql,(CreateView)st,this.statement.getConnection().getSchema());
        }else if(st instanceof Delete){
            this.execute(sql,(Delete)st,this.statement.getConnection().getSchema());
        }else  if(st instanceof Insert){
            this.execute(sql,(Insert)st,this.statement.getConnection().getSchema());
        }
        result[i]= Statement.SUCCESS_NO_INFO;
    }catch (Exception e){
        result[i] = Statement.EXECUTE_Failed;
    }
    this.clearBulk();
    return result;
}

@Inject
public CreateViewTask(
        JsonCodec<ViewDeFinition> codec,FeaturesConfig featuresConfig)
{
    this.codec = requireNonNull(codec,"codec is null");
    this.sqlParser = requireNonNull(sqlParser,"accessControl is null");
    requireNonNull(featuresConfig,"featuresConfig is null");
    this.experimentalSyntaxEnabled = featuresConfig.isExperimentalSyntaxEnabled();
}

@Inject
sqlQueryExecutionFactory(QueryManagerConfig config,FeaturesConfig featuresConfig,LocationFactory locationFactory,SplitManager splitManager,NodeScheduler nodeScheduler,List<Planoptimizer> planoptimizers,RemoteTaskFactory remoteTaskFactory,TransactionManager transactionManager,@ForQueryExecution ExecutorService executor,NodeTaskMap nodeTaskMap,QueryExplainer queryExplainer,Map<String,ExecutionPolicy> executionPolicies)
{
    requireNonNull(config,"config is null");
    this.scheduleSplitBatchSize = config.getScheduleSplitBatchSize();
    this.Metadata = requireNonNull(Metadata,"Metadata is null");
    this.accessControl = requireNonNull(accessControl,"accessControl is null");
    this.sqlParser = requireNonNull(sqlParser,"sqlParser is null");
    this.locationFactory = requireNonNull(locationFactory,"locationFactory is null");
    this.splitManager = requireNonNull(splitManager,"splitManager is null");
    this.nodeScheduler = requireNonNull(nodeScheduler,"nodeScheduler is null");
    this.planoptimizers = requireNonNull(planoptimizers,"planoptimizers is null");
    this.remoteTaskFactory = requireNonNull(remoteTaskFactory,"remoteTaskFactory is null");
    this.transactionManager = requireNonNull(transactionManager,"transactionManager is null");
    requireNonNull(featuresConfig,"featuresConfig is null");
    this.experimentalSyntaxEnabled = featuresConfig.isExperimentalSyntaxEnabled();
    this.executor = requireNonNull(executor,"executor is null");
    this.nodeTaskMap = requireNonNull(nodeTaskMap,"nodeTaskMap is null");
    this.queryExplainer = requireNonNull(queryExplainer,"queryExplainer is null");

    this.executionPolicies = requireNonNull(executionPolicies,"schedulerPolicies is null");
}

@Inject
public LocalExecutionPlanner(
        Metadata Metadata,PageSourceProvider pageSourceProvider,IndexManager indexManager,PageSinkManager pageSinkManager,ExchangeClientsupplier exchangeClientsupplier,ExpressionCompiler compiler,IndexJoinLookupStats indexJoinLookupStats,CompilerConfig compilerConfig,TaskManagerConfig taskManagerConfig)
{
    requireNonNull(compilerConfig,"compilerConfig is null");
    this.pageSourceProvider = requireNonNull(pageSourceProvider,"pageSourceProvider is null");
    this.indexManager = requireNonNull(indexManager,"indexManager is null");
    this.exchangeClientsupplier = exchangeClientsupplier;
    this.Metadata = requireNonNull(Metadata,"sqlParser is null");
    this.pageSinkManager = requireNonNull(pageSinkManager,"pageSinkManager is null");
    this.compiler = requireNonNull(compiler,"compiler is null");
    this.indexJoinLookupStats = requireNonNull(indexJoinLookupStats,"indexJoinLookupStats is null");
    this.maxIndexMemorySize = requireNonNull(taskManagerConfig,"taskManagerConfig is null").getMaxIndexMemoryUsage();
    this.maxPartialAggregationMemorySize = taskManagerConfig.getMaxPartialAggregationMemoryUsage();

    interpreterEnabled = compilerConfig.isInterpreterEnabled();
}

private Rewriter(
        SymbolAllocator symbolAllocator,PlanNodeIdAllocator idAllocator,Session session)
{
    this.symbolAllocator = requireNonNull(symbolAllocator,"symbolAllocator is null");
    this.idAllocator = requireNonNull(idAllocator,"idAllocator is null");
    this.Metadata = requireNonNull(Metadata,"sqlParser is null");
    this.session = requireNonNull(session,"session is null");
}

public Visitor(Metadata Metadata,sqlParser parser)
{
    this.Metadata = Metadata;
    this.session = session;
    this.types = types;
    this.parser = parser;
}

public Rewriter(Metadata Metadata,PlanNodeIdAllocator idAllocator)
{
    this.Metadata = Metadata;
    this.sqlParser = sqlParser;
    this.session = session;
    this.types = types;
    this.idAllocator = idAllocator;
}

/**
 * Extract a normalized simple comparison between a QualifiednameReference and a native value if possible.
 */
private static Optional<normalizedSimpleComparison> tonormalizedSimpleComparison(Session session,ComparisonExpression comparison)
{
    IdentityHashMap<Expression,Type> expressionTypes = ExpressionAnalyzer.getExpressionTypes(session,types,comparison);
    Object left = ExpressionInterpreter.expressionoptimizer(comparison.getLeft(),expressionTypes).optimize(NoOpSymbolResolver.INSTANCE);
    Object right = ExpressionInterpreter.expressionoptimizer(comparison.getRight(),expressionTypes).optimize(NoOpSymbolResolver.INSTANCE);

    if (left instanceof QualifiednameReference && !(right instanceof Expression)) {
        return Optional.of(new normalizedSimpleComparison((QualifiednameReference) left,comparison.getType(),new NullableValue(expressionTypes.get(comparison.getRight()),right)));
    }
    if (right instanceof QualifiednameReference && !(left instanceof Expression)) {
        return Optional.of(new normalizedSimpleComparison((QualifiednameReference) right,flipComparison(comparison.getType()),new NullableValue(expressionTypes.get(comparison.getLeft()),left)));
    }
    return Optional.empty();
}

public Analyzer(Session session,Optional<QueryExplainer> queryExplainer,boolean experimentalSyntaxEnabled)
{
    this.session = requireNonNull(session,"session is null");
    this.Metadata = requireNonNull(Metadata,"accessControl is null");
    this.queryExplainer = requireNonNull(queryExplainer,"query explainer is null");
    this.experimentalSyntaxEnabled = experimentalSyntaxEnabled;
}

public static IdentityHashMap<Expression,Type> getExpressionTypes(
        Session session,Expression expression)
{
    return getExpressionTypes(session,ImmutableList.of(expression));
}

public static IdentityHashMap<Expression,Iterable<? extends Expression> expressions)
{
    return analyzeExpressionsWithSymbols(session,expressions).getExpressionTypes();
}

public static IdentityHashMap<Expression,Type> getExpressionTypesFromInput(
        Session session,Map<Integer,Expression expression)
{
    return getExpressionTypesFromInput(session,Iterable<? extends Expression> expressions)
{
    return analyzeExpressionsWithInputs(session,expressions).getExpressionTypes();
}

private static ExpressionAnalysis analyzeExpressionsWithInputs(
        Session session,Iterable<? extends Expression> expressions)
{
    Field[] fields = new Field[types.size()];
    for (Entry<Integer,Type> entry : types.entrySet()) {
        fields[entry.getKey()] = Field.newUnqualified(Optional.empty(),entry.getValue());
    }
    RelationType tupleDescriptor = new RelationType(fields);

    return analyzeExpressions(session,expressions);
}

public static ExpressionAnalysis analyzeExpression(
        Session session,Analysis analysis,boolean approximateQueriesEnabled,AnalysisContext context,Expression expression)
{
    ExpressionAnalyzer analyzer = create(analysis,approximateQueriesEnabled);
    analyzer.analyze(expression,context);

    IdentityHashMap<Expression,Type> expressionTypes = analyzer.getExpressionTypes();
    IdentityHashMap<Expression,Type> expressionCoercions = analyzer.getExpressionCoercions();
    IdentityHashMap<FunctionCall,Signature> resolvedFunctions = analyzer.getResolvedFunctions();

    analysis.addTypes(expressionTypes);
    analysis.addCoercions(expressionCoercions);
    analysis.addFunctionSignatures(resolvedFunctions);

    analysis.addResolvednames(analyzer.getResolvednames());

    return new ExpressionAnalysis(
            expressionTypes,expressionCoercions,analyzer.getColumnReferences());
}

public static ExpressionAnalyzer create(
        Analysis analysis,boolean experimentalSyntaxEnabled)
{
    return new ExpressionAnalyzer(
            Metadata.getFunctionRegistry(),Metadata.getTypeManager(),node -> new StatementAnalyzer(analysis,experimentalSyntaxEnabled,Optional.empty()),session);
}

public QueryExplainer(
        List<Planoptimizer> planoptimizers,boolean experimentalSyntaxEnabled)
{
    this.planoptimizers = requireNonNull(planoptimizers,"planoptimizers is null");
    this.Metadata = requireNonNull(Metadata,"sqlParser is null");
    this.experimentalSyntaxEnabled = experimentalSyntaxEnabled;
    this.dataDeFinitionTask = ImmutableMap.copyOf(requireNonNull(dataDeFinitionTask,"dataDeFinitionTask is null"));
}

@Test
public void testUpdateSessionParameters()
        throws Exception
{
    ClientOptions options = new ClientOptions();
    ClientSession session = options.toClientSession();
    sqlParser sqlParser = new sqlParser();

    ImmutableMap<String,String> existingProperties = ImmutableMap.of("query_max_memory","10GB","distributed_join","true");
    session = Console.processSessionParameterChange(sqlParser.createStatement("USE test_catalog.test_schema"),existingProperties);
    assertEquals(session.getCatalog(),"test_catalog");
    assertEquals(session.getSchema(),"test_schema");
    assertEquals(session.getProperties().get("query_max_memory"),"10GB");
    assertEquals(session.getProperties().get("distributed_join"),"true");

    session = Console.processSessionParameterChange(sqlParser.createStatement("USE test_schema_b"),"test_schema_b");
    assertEquals(session.getProperties().get("query_max_memory"),"true");

    session = Console.processSessionParameterChange(sqlParser.createStatement("USE test_catalog_2.test_schema"),"test_catalog_2");
    assertEquals(session.getSchema(),"true");
}

public static void assertFormattedsql(sqlParser sqlParser,Node expected)
{
    String formatted = formatsql(expected);

    // verify round-trip of formatting already-formatted sql
    Statement actual = parseFormatted(sqlParser,formatted,expected);
    assertEquals(formatsql(actual),formatted);

    // compare parsed tree with parsed tree of formatted sql
    if (!actual.equals(expected)) {
        // simplify finding the non-equal part of the tree
        assertListEquals(linearizeTree(actual),linearizeTree(expected));
    }
    assertEquals(actual,expected);
}

private static Statement parseFormatted(sqlParser sqlParser,String sql,Node tree)
{
    try {
        return sqlParser.createStatement(sql);
    }
    catch (ParsingException e) {
        throw new AssertionError(format(
                "Failed to parse formatted sql: %s\nerror: %s\ntree: %s",sql,e.getMessage(),tree));
    }
}

@Test
public void testSimpleExpression()
        throws Exception {
    Expression expression = new sqlParser().createExpression("test = 'test'");

    assertEquals(formatExpression(expression,name -> {
                throw new UnsupportedOperationException();
            },name -> "\"dummy\".\"" + name + "\"",'"'),"(\"dummy\".\"test\" = 'test')");
}

@Test
public void testJoinQuery()
        throws Exception {
    Statement statement = new sqlParser().createStatement
            ("select * from testcollection join anothercollection on (anothercollection.test = testcollection.test)");

    // Todo: decide if we should also format expressions in QueryFormatter
    assertEquals(formatQuery(statement,"SELECT *\n" +
            "   FROM\n" +
            "     (dummy\n" +
            "   INNER JOIN dummy ON (\"anothercollection\".\"test\" = \"testcollection\".\"test\"))");
}

@Test
public void testQueryWithCTE()
        throws Exception {
    Statement statement = new sqlParser().createStatement("with test as (select * from collection) select * from test");

    assertEquals(formatQuery(statement,"WITH\n" +
            "     \"test\" AS (\n" +
            "      SELECT *\n" +
            "      FROM\n" +
            "        dummy\n" +
            "   ) \n" +
            "   SELECT *\n" +
            "   FROM\n" +
            "     test");
}

@Test
public void testAlias()
        throws Exception {
    Statement statement = new sqlParser().createStatement("select a as b from test");

    assertEquals(formatQuery(statement,"SELECT \"a\" \"b\"\n" +
            "   FROM\n" +
            "     dummy");
}

@Test
public void testQueryWithCTEDuplicateName()
        throws Exception {
    Statement statement = new sqlParser().createStatement("with test as (select * from collection) select * from collection.test");

    assertEquals(formatQuery(statement,"WITH\n" +
            "     \"test\" AS (\n" +
            "      SELECT *\n" +
            "      FROM\n" +
            "        dummy\n" +
            "   ) \n" +
            "   SELECT *\n" +
            "   FROM\n" +
            "     dummy");
}

@Test
public void testExpressionFormatterFormatTable()
        throws Exception {
    Expression expression = new sqlParser().createExpression("test in (select id from testcollection)");

    assertEquals(formatExpression(expression,name -> "\"schema\"." + name.getParts().stream().map(e -> formatIdentifier(e,'"')).collect(Collectors.joining(".")),name -> '"' + name + '"',"(\"test\" IN (SELECT \"id\"\n" +
            "FROM\n" +
            "  \"schema\".\"testcollection\"\n" +
            "))");
}

@Test
public void testName() throws Exception {
    String sql = "select selami:timestamp,melami:varchar from deneme where ali:timestamp is not null and veli is null group by demo";
    sqlParserOptions options = new sqlParserOptions().allowIdentifierSymbol(IdentifierSymbol.COLON);
    Statement statement = new sqlParser(options).createStatement(sql);

    String s = RakamsqlFormatter.formatsql(statement,name -> String.format("(SELECT * FROM events WHERE collection_name = '%s')",name.toString()),name -> "\"$data\"['" + name + "']",'"');
    System.out.println(s);
}

public sqlStatement parse(sqlSession session,String sql) {
    log.debug("Parsing sql: {}",sql);

    Statement statement = sqlParser.createStatement(sql);

    QueryExplainer queryExplainer = new QueryExplainer(session.prestoSession,planoptimizers,MetadataManager,periodicImportManager,storageManager);
    // analyze query
    Analyzer analyzer = new Analyzer(session.prestoSession,Optional.of(queryExplainer));

    Analysis analysis = analyzer.analyze(statement);

    // System.out.println("analysis: " + analysis);

    PlanNodeIdAllocator idAllocator = new PlanNodeIdAllocator();
    // plan query
    LogicalPlanner logicalPlanner = new LogicalPlanner(session.prestoSession,idAllocator,storageManager);
    Plan plan = logicalPlanner.plan(analysis);

    return new sqlStatement(session,plan);
    //
    // TableScanCountVisitor visitor = new TableScanCountVisitor();
    // plan.getRoot().accept(visitor,0);
    // Assert.assertEquals(1,visitor.count);
    // String p = PlanPrinter.textLogicalPlan(plan.getRoot(),plan.getTypes());
    //
    // System.out.println("plan: " + p);
}

private Plan parse(String sql) {
    InMemoryNodeManager nodeManager = new InMemoryNodeManager();

    MetadataManager Metadata = buildMetadata();
    StorageManager storageManager = new MockStorageManager();
    PeriodicImportManager periodicImportManager = new StubPeriodicImportManager();

    SplitManager splitManager = buildSplitManager(nodeManager);
    List<Planoptimizer> planoptimizers = buildplanoptimizers(Metadata,splitManager);

    Statement statement = sqlParser.createStatement(sql);

    // System.out.println("Statement: " + statement);

    Session session = buildSession();
    QueryExplainer queryExplainer = new QueryExplainer(session,storageManager);
    // analyze query
    Analyzer analyzer = new Analyzer(session,Optional.of(queryExplainer));

    Analysis analysis = analyzer.analyze(statement);

    // System.out.println("analysis: " + analysis);

    PlanNodeIdAllocator idAllocator = new PlanNodeIdAllocator();
    // plan query
    LogicalPlanner logicalPlanner = new LogicalPlanner(session,storageManager);
    Plan plan = logicalPlanner.plan(analysis);
    return plan;
}

@SuppressWarnings("unchecked")
public int execute(String update) throws sqlException {
    Matcher matcher = updateRegex.matcher(update);
    if(!matcher.find()) throw new sqlException("Unable to parse UPDATE statement");

    // get index and type to update
    String index = statement.getConnection().getSchema();
    String type = matcher.group(1);
    if(matcher.group(2) != null){
        index = type;
        type = matcher.group(2);
    }

    // get fields and values to update
    try{
        Map<String,Object> fieldValues = new HashMap<String,Object>();
        sqlParser parser = new sqlParser();
        String[] parts = matcher.group(3).replaceAll(",\\s*([\"|\\w|\\.]+\\s*=)","<-SPLIT->$1").split("<-SPLIT->");
        for(String p : parts){
            ComparisonExpression comparison = (ComparisonExpression) parser.createExpression(p);
            String field = comparison.getLeft().toString().replaceAll("\"","");
            field = heading.findOriginal(matcher.group(3),field,"","\\s*=");
            Object value = getLiteralValue(comparison.getRight());

            if(field.indexOf('.') == -1) {
                fieldValues.put(field,value);
                continue;
            }

            // create nested object
            Map<String,Object> map = fieldValues; 
            String[] objectDef = field.split("\\.");
            for(int k=0; k<objectDef.length; k++){
                String key = objectDef[k];
                if(k == objectDef.length-1) map.put(key,value);
                else{
                    if(!map.containsKey(key)) map.put(key,new HashMap<String,Object>());
                        map = (Map<String,Object>)map.get(key);
                }
            }

        }

        // get ID's for documents to be updated

        String select = "SELECT _id FROM "+type+" WHERE "+matcher.group(4);
        Query query = (Query)new sqlParser().createStatement(select);
        this.queryState.buildrequest(select,query.getQueryBody(),index);
        ResultSet rs = this.queryState.execute();

        // execute updates in batch mode based on id's returned
        int maxRequestsPerBulk = Utils.getIntProp(props,Utils.PROP_FETCH_SIZE,2500);
        List<UpdateRequestBuilder> indexReqs = new ArrayList<UpdateRequestBuilder>();
        int updateCount = 0;
        while(rs != null){
            while(rs.next()){
                String id = rs.getString(1);
                indexReqs.add(
                    client.prepareUpdate(index,type,id).setDoc(fieldValues)
                );
                if(indexReqs.size() >= maxRequestsPerBulk){
                    updateCount += this.execute(indexReqs,maxRequestsPerBulk);
                    indexReqs.clear();
                }
            }
            rs.close();
            rs = queryState.moreResults(true);
        }
        if(indexReqs.size() > 0)  updateCount += this.execute(indexReqs,maxRequestsPerBulk);
        return updateCount;
    }catch(Exception e){
        throw new sqlException("Unable to execute UPDATE due to "+e.getMessage(),e);
    }

}

public static final <T> T parSEObject(String input,Type clazz,ParserConfig config,ParseProcess processor,int featureValues,Feature... features) {
    if (input == null) {
        return null;
    }
    for (Feature featrue : features) {
        featureValues = Feature.config(featureValues,featrue,true);
    }
    DefaultJSONParser parser = new DefaultJSONParser(input,config,featureValues);
    if (processor instanceof ExtraTypeProvider) {
        parser.getExtraTypeProviders().add((ExtraTypeProvider) processor);
    }
    if (processor instanceof ExtraProcessor) {
        parser.getExtraProcessors().add((ExtraProcessor) processor);
    }
    T value = parser.parSEObject(clazz);
    parser.handleResovleTask(value);
    parser.close();
    return value;
}

@SuppressWarnings("unchecked")
public static <T> T parSEObject(String input,Feature... features) {
    if (input == null) {
        return null;
    }

    if (features != null) {
        for (Feature feature : features) {
            featureValues |= feature.mask;
        }
    }

    DefaultJSONParser parser = new DefaultJSONParser(input,featureValues);

    if (processor != null) {
        if (processor instanceof ExtraTypeProvider) {
            parser.getExtraTypeProviders().add((ExtraTypeProvider) processor);
        }

        if (processor instanceof ExtraProcessor) {
            parser.getExtraProcessors().add((ExtraProcessor) processor);
        }

        if (processor instanceof FieldTypeResolver) {
            parser.setFieldTypeResolver((FieldTypeResolver) processor);
        }
    }

    T value = (T) parser.parSEObject(clazz,null);

    parser.handleResovleTask(value);

    parser.close();

    return (T) value;
}

@SuppressWarnings("unchecked")
public static final <T> T parSEObject(String input,Feature... features) {
    if (input == null) {
        return null;
    }

    for (Feature featrue : features) {
        featureValues = Feature.config(featureValues,true);
    }

    DefaultJSONParser parser = new DefaultJSONParser(input,featureValues);

    if (processor instanceof ExtraTypeProvider) {
        parser.getExtraTypeProviders().add((ExtraTypeProvider) processor);
    }

    if (processor instanceof ExtraProcessor) {
        parser.getExtraProcessors().add((ExtraProcessor) processor);
    }

    T value = (T) parser.parSEObject(clazz);

    parser.handleResovleTask(value);

    parser.close();

    return (T) value;
}

@SuppressWarnings("unchecked")
public static final <T> T parSEObject(String input,featureValues);

    if (processor instanceof ExtraTypeProvider) {
        parser.getExtraTypeProviders().add((ExtraTypeProvider) processor);
    }

    if (processor instanceof ExtraProcessor) {
        parser.getExtraProcessors().add((ExtraProcessor) processor);
    }

    T value = (T) parser.parSEObject(clazz);

    handleResovleTask(parser,value);

    parser.close();

    return (T) value;
}

@SuppressWarnings("unchecked")
public static <T> T parSEObject(String text,Class<T> clazz,Feature... features) {
    return (T) parSEObject(text,(Type) clazz,ParserConfig.global,processor,DEFAULT_PARSER_FEATURE,features);
}

@SuppressWarnings("unchecked")
public static <T> T parSEObject(String input,Feature... features) {
    return (T) parSEObject(input,clazz,features);
}

public static final <T> T parSEObject(String text,Feature... features) {
    return parSEObject(text,ParserConfig.getGlobalInstance(),features);
}

public static final <T> T parSEObject(String input,Feature... features) {
    return parSEObject(input,features);
}

@SuppressWarnings("unchecked")
public static final <T> T parSEObject(String text,features);
}

@SuppressWarnings("unchecked")
public static final <T> T parSEObject(String input,features);
}

@SuppressWarnings("unchecked")
public static final <T> T parSEObject(String text,features);
}

@SuppressWarnings("unchecked")
public static final <T> T parSEObject(String input,features);
}

public static EthereumBlockRange fromMarkers(Marker low,Marker high) {
    long startBlock;
    long endBlock;
    if (low.isLowerUnbounded()) {
        startBlock = 1L;
    } else if (low.getBound() == Marker.Bound.EXACTLY) {
        startBlock = (long) low.getValue();
    } else if (low.getBound() == Marker.Bound.ABOVE) {
        startBlock = (long) low.getValue() + 1L;
    } else {
        throw new IllegalArgumentException("Low bound cannot be BELOW");
    }

    if (high.isUpperUnbounded()) {
        endBlock = -1L;
    } else if (high.getBound() == Marker.Bound.EXACTLY) {
        endBlock = (long) high.getValue();
    } else if (high.getBound() == Marker.Bound.BELOW) {
        endBlock = (long) high.getValue() - 1L;
    } else {
        throw new IllegalArgumentException("High bound cannot be ABOVE");
    }

    if (startBlock > endBlock && endBlock != -1L) {
        throw new IllegalArgumentException("Low bound is greater than high bound");
    }

    return new EthereumBlockRange(startBlock,endBlock);
}

private static boolean isBetween(Range range)
{
    return !range.getLow().isLowerUnbounded() && range.getLow().getBound() == Marker.Bound.EXACTLY
            && !range.getHigh().isUpperUnbounded() && range.getHigh().getBound() == Marker.Bound.EXACTLY;
}

@Test
public void testGetRecordSet()
        throws Exception
{
    ExampleRecordSetProvider recordSetProvider = new ExampleRecordSetProvider(new ExampleConnectorId("test"));
    RecordSet recordSet = recordSetProvider.getRecordSet(SESSION,new ExampleSplit("test","schema","table",dataUri),ImmutableList.of(
            new ExampleColumnHandle("test","text",VARCHAR,0),new ExampleColumnHandle("test","value",BIGINT,1)));
    assertNotNull(recordSet,"recordSet is null");

    RecordCursor cursor = recordSet.cursor();
    assertNotNull(cursor,"cursor is null");

    Map<String,Long> data = new LinkedHashMap<>();
    while (cursor.advanceNextPosition()) {
        data.put(cursor.getSlice(0).toStringUtf8(),cursor.getLong(1));
    }
    assertEquals(data,ImmutableMap.<String,Long>builder()
            .put("ten",10L)
            .put("eleven",11L)
            .put("twelve",12L)
            .build());
}

@Test
public void testCursorSimple()
        throws Exception
{
    RecordSet recordSet = new ExampleRecordSet(new ExampleSplit("test",1)));
    RecordCursor cursor = recordSet.cursor();

    assertEquals(cursor.getType(0),VARCHAR);
    assertEquals(cursor.getType(1),BIGINT);

    Map<String,cursor.getLong(1));
        assertFalse(cursor.isNull(0));
        assertFalse(cursor.isNull(1));
    }
    assertEquals(data,12L)
            .build());
}

@Test
public void testCursorMixedOrder()
        throws Exception
{
    RecordSet recordSet = new ExampleRecordSet(new ExampleSplit("test",1),0)));
    RecordCursor cursor = recordSet.cursor();

    Map<String,Long> data = new LinkedHashMap<>();
    while (cursor.advanceNextPosition()) {
        assertEquals(cursor.getLong(0),cursor.getLong(1));
        data.put(cursor.getSlice(2).toStringUtf8(),cursor.getLong(0));
    }
    assertEquals(data,12L)
            .build());
}

private static IndexedTable indexTable(RecordSet recordSet,final List<String> outputColumns,List<String> keyColumns)
{
    List<Integer> keyPositions = FluentIterable.from(keyColumns)
            .transform(columnName -> {
                int position = outputColumns.indexOf(columnName);
                checkState(position != -1);
                return position;
            })
            .toList();

    ImmutableListMultimap.Builder<MaterializedTuple,MaterializedTuple> indexedValuesBuilder = ImmutableListMultimap.builder();

    List<Type> outputTypes = recordSet.getColumnTypes();
    List<Type> keyTypes = extractPositionValues(outputTypes,keyPositions);

    RecordCursor cursor = recordSet.cursor();
    while (cursor.advanceNextPosition()) {
        List<Object> values = extractValues(cursor,outputTypes);
        List<Object> keyvalues = extractPositionValues(values,keyPositions);

        indexedValuesBuilder.put(new MaterializedTuple(keyvalues),new MaterializedTuple(values));
    }

    return new IndexedTable(keyColumns,keyTypes,outputColumns,outputTypes,indexedValuesBuilder.build());
}

private static Object extractObject(RecordCursor cursor,int field,Type type)
{
    if (cursor.isNull(field)) {
        return null;
    }

    Class<?> javaType = type.getJavaType();
    if (javaType == boolean.class) {
        return cursor.getBoolean(field);
    }
    else if (javaType == long.class) {
        return cursor.getLong(field);
    }
    else if (javaType == double.class) {
        return cursor.getDouble(field);
    }
    else if (javaType == Slice.class) {
        return cursor.getSlice(field).toStringUtf8();
    }
    throw new AssertionError("Unsupported type: " + type);
}

private static Iterable<MaterializedTuple> tupleIterable(final RecordSet recordSet)
{
    return () -> new AbstractIterator<MaterializedTuple>()
    {
        private final RecordCursor cursor = recordSet.cursor();

        @Override
        protected MaterializedTuple computeNext()
        {
            if (!cursor.advanceNextPosition()) {
                return endOfData();
            }
            return new MaterializedTuple(extractValues(cursor,recordSet.getColumnTypes()));
        }
    };
}

@Test
public void testRecordSetProvider()
        throws Exception
{
    for (SchemaTableName schemaTableName : Metadata.listTables(SESSION,"jmx")) {
        JmxTableHandle tableHandle = Metadata.getTableHandle(SESSION,schemaTableName);
        List<ColumnHandle> columnHandles = ImmutableList.copyOf(Metadata.getColumnHandles(SESSION,tableHandle).values());

        ConnectorTableLayoutHandle layout = new JmxTableLayoutHandle(tableHandle,TupleDomain.all());
        ConnectorSplitSource splitSource = splitManager.getSplits(JmxTransactionHandle.INSTANCE,SESSION,layout);
        List<ConnectorSplit> allSplits = getAllSplits(splitSource);
        assertEquals(allSplits.size(),nodes.size());
        ConnectorSplit split = allSplits.get(0);

        RecordSet recordSet = recordSetProvider.getRecordSet(JmxTransactionHandle.INSTANCE,split,columnHandles);
        try (RecordCursor cursor = recordSet.cursor()) {
            while (cursor.advanceNextPosition()) {
                for (int i = 0; i < recordSet.getColumnTypes().size(); i++) {
                    cursor.isNull(i);
                }
            }
        }
    }
}

@Test
public void testGetRecordSet()
        throws Exception
{
    JdbcRecordSetProvider recordSetProvider = new JdbcRecordSetProvider(jdbcclient);
    RecordSet recordSet = recordSetProvider.getRecordSet(SESSION,ImmutableList.of(textColumn,valueColumn));
    assertNotNull(recordSet,Long>builder()
            .put("one",1L)
            .put("two",2L)
            .put("three",3L)
            .put("ten",12L)
            .build());
}

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

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

@Override
public RecordCursor cursor(ConnectorTransactionHandle transactionHandle,ConnectorSession session,TupleDomain<Integer> constraint)
{
    Builder table = InMemoryRecordSet.builder(transactionsTable);
    for (TransactionInfo info : transactionManager.getAllTransactionInfos()) {
        table.addRow(
                info.getTransactionId().toString(),info.getIsolationLevel().toString(),info.isReadOnly(),info.isAutoCommitContext(),info.getCreateTime().getMillis(),(long) info.getIdleTime().getValue(TimeUnit.SECONDS),info.getWrittenConnectorId().orElse(null),createStringsBlock(info.getConnectorIds()));
    }
    return table.build().cursor();
}

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

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

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

@Override
public RecordCursor cursor(ConnectorTransactionHandle transactionHandle,TupleDomain<Integer> constraint)
{
    Builder table = InMemoryRecordSet.builder(TABLE_PROPERTIES_TABLE);
    Map<String,Map<String,PropertyMetadata<?>>> tableProperties = new TreeMap<>(Metadata.getTablePropertyManager().getAllTableProperties());
    for (Entry<String,PropertyMetadata<?>>> entry : tableProperties.entrySet()) {
        String catalog = entry.getKey();
        Map<String,PropertyMetadata<?>> properties = new TreeMap<>(entry.getValue());
        for (PropertyMetadata<?> propertyMetadata : properties.values()) {
            table.addRow(
                    catalog,propertyMetadata.getName(),firstNonNull(propertyMetadata.getDefaultValue(),"").toString(),propertyMetadata.getsqlType().toString(),propertyMetadata.getDescription());
        }
    }
    return table.build().cursor();
}

@Override
public RecordCursor cursor(ConnectorTransactionHandle transactionHandle,TupleDomain<Integer> constraint)
{
    Builder table = InMemoryRecordSet.builder(QUERY_TABLE);
    for (QueryInfo queryInfo : queryManager.getAllQueryInfo()) {
        QueryStats queryStats = queryInfo.getQueryStats();
        table.addRow(
                nodeId,queryInfo.getQueryId().toString(),queryInfo.getState().toString(),queryInfo.getSession().getUser(),queryInfo.getSession().getSource().orElse(null),queryInfo.getQuery(),toMillis(queryStats.getQueuedTime()),toMillis(queryStats.getAnalysisTime()),toMillis(queryStats.getdistributedplanningTime()),toTimeStamp(queryStats.getCreateTime()),toTimeStamp(queryStats.getExecutionStartTime()),toTimeStamp(queryStats.getLastHeartbeat()),toTimeStamp(queryStats.getEndTime()));
    }
    return table.build().cursor();
}

@Override
public int process(ConnectorSession session,RecordCursor cursor,int count,PageBuilder pageBuilder)
{
    checkArgument(!pageBuilder.isFull(),"page builder can't be full");
    checkArgument(count > 0,"count must be > 0");

    int position = 0;
    for (; position < count; position++) {
        if (pageBuilder.isFull()) {
            break;
        }

        if (!cursor.advanceNextPosition()) {
            break;
        }

        if (filterFunction.filter(cursor)) {
            pageBuilder.declarePosition();
            for (int channel = 0; channel < projections.size(); channeL++) {
                // todo: if the projection function increases the size of the data significantly,this Could cause the servers to OOM
                projections.get(channel).project(cursor,pageBuilder.getBlockBuilder(channel));
            }
        }
    }
    return position;
}

private void generateProjectMethod(ClassDeFinition classDeFinition,CallSiteBinder callSiteBinder,CachedInstanceBinder cachedInstanceBinder,String methodName,RowExpression projection)
{
    Parameter session = arg("session",ConnectorSession.class);
    Parameter cursor = arg("cursor",RecordCursor.class);
    Parameter output = arg("output",BlockBuilder.class);
    MethodDeFinition method = classDeFinition.declareMethod(a(PUBLIC),methodName,type(void.class),cursor,output);

    method.comment("Projection: %s",projection.toString());

    Scope scope = method.getScope();
    Variable wasNullVariable = scope.declareVariable(type(boolean.class),"wasNull");

    BytecodeBlock body = method.getBody()
            .comment("boolean wasNull = false;")
            .putvariable(wasNullVariable,false);

    BytecodeExpressionVisitor visitor = new BytecodeExpressionVisitor(callSiteBinder,cachedInstanceBinder,fieldReferenceCompiler(cursor,wasNullVariable),Metadata.getFunctionRegistry());

    body.getvariable(output)
            .comment("evaluate projection: " + projection.toString())
            .append(projection.accept(visitor,scope))
            .append(generateWrite(callSiteBinder,scope,wasNullVariable,projection.getType()))
            .ret();
}

@Test
public void test()
{
    ArrayType arrayOfBigintType = new ArrayType(BIGINT);
    FieldSetFilteringRecordSet fieldSetFilteringRecordSet = new FieldSetFilteringRecordSet(
            FUNCTION_REGISTRY,new InMemoryRecordSet(
                    ImmutableList.of(BIGINT,TIMESTAMP_WITH_TIME_ZONE,arrayOfBigintType,arrayOfBigintType),ImmutableList.of(
                            ImmutableList.of(
                                    100L,100L,// test same time in different time zone to make sure equal check was done properly
                                    packDateTimeWithZone(100,getTimeZoneKeyForOffset(123)),packDateTimeWithZone(100,getTimeZoneKeyForOffset(234)),// test structural type
                                    arrayBlockOf(BIGINT,12,34,56),arrayBlockOf(BIGINT,56)
                            ))),ImmutableList.of(ImmutableSet.of(0,ImmutableSet.of(2,3),ImmutableSet.of(4,5)));
    RecordCursor recordCursor = fieldSetFilteringRecordSet.cursor();
    assertTrue(recordCursor.advanceNextPosition());
}

@Override
public RecordCursor cursor() {
    EthBlock block = null;
    try {
        block = web3j.ethGetBlockByNumber(DefaultBlockParameter.valueOf(BigInteger.valueOf(split.getBlockId())),true).send();
    } catch (IOException e) {
        e.printstacktrace();
    }
    return new EthereumRecordCursor(columnHandles,block,split.getTable(),web3j);
}

private static List<MaterializedRow> getMaterializedResults(RecordCursor cursor,List<ColumnMetadata> columns)
{
    List<Type> types = columns.stream().map(ColumnMetadata::getType).collect(toList());

    ImmutableList.Builder<MaterializedRow> rowBuilder = ImmutableList.builder();
    for (int i = 0; i < types.size(); i++) {
        assertEquals(cursor.getType(i),types.get(i));
    }

    while (cursor.advanceNextPosition()) {
        List<Object> values = new ArrayList<>(types.size());
        for (int i = 0; i < columns.size(); i++) {
            Type type = columns.get(i).getType();
            Class<?> javaType = type.getJavaType();
            if (cursor.isNull(i)) {
                continue;
            }
            if (javaType == boolean.class) {
                values.add(i,cursor.getBoolean(i));
            }
            else if (javaType == long.class) {
                values.add(i,cursor.getLong(i));
            }
            else if (javaType == double.class) {
                values.add(i,cursor.getDouble(i));
            }
            else if (javaType == Slice.class) {
                values.add(i,cursor.getSlice(i));
            }
        }
        rowBuilder.add(new MaterializedRow(DEFAULT_PRECISION,values));
    }
    return rowBuilder.build();
}

private AppendingRecordCursor(RecordCursor delegate,int delegateFieldCount,List<Object> appendedValues,List<Type> appendedTypes)
{
    this.delegate = requireNonNull(delegate,"delegate is null");
    this.delegateFieldCount = delegateFieldCount;
    checkArgument(delegateFieldCount >= 0,"delegateFieldCount must be greater than or equal to zero");
    this.appendedValues = requireNonNull(appendedValues,"appendedValues is null"); // May contain null elements
    this.appendedTypes = ImmutableList.copyOf(requireNonNull(appendedTypes,"appendedTypes is null"));
    checkArgument(appendedValues.size() == appendedTypes.size(),"appendedValues must have the same size as appendedTypes");
}

@Override
public RecordCursor cursor()
{
    // NOTE: the ConcatRecordCursor implementation relies on the fact that the
    // cursor creation in the Iterable is lazy so DO NOT materialize this into
    // an ImmutableList
    Iterable<RecordCursor> recordCursors = transform(recordSets,RecordSet::cursor);
    return new ConcatRecordCursor(recordCursors.iterator(),types);
}

private ConcatRecordCursor(Iterator<RecordCursor> iterator,List<Type> types)
{
    // NOTE: this cursor implementation relies on the fact that the
    // cursor creation in the Iterable is lazy so DO NOT materialize this into
    // an ImmutableList
    this.iterator = requireNonNull(iterator,"iterator is null");
    this.types = ImmutableList.copyOf(requireNonNull(types,"types is null"));
}

private static List<Object> extractValues(RecordCursor cursor,List<Type> types)
{
    List<Object> list = new ArrayList<>(types.size());
    for (int i = 0; i < types.size(); i++) {
        list.add(extractObject(cursor,i,types.get(i)));
    }
    return list;
}

@Test
public void testCursorSimple()
        throws Exception
{
    RecordSet recordSet = new JdbcRecordSet(jdbcclient,ImmutableList.of(
            columnHandles.get("text"),columnHandles.get("value")));

    try (RecordCursor cursor = recordSet.cursor()) {
        assertEquals(cursor.getType(0),VARCHAR);
        assertEquals(cursor.getType(1),BIGINT);

        Map<String,Long> data = new LinkedHashMap<>();
        while (cursor.advanceNextPosition()) {
            data.put(cursor.getSlice(0).toStringUtf8(),cursor.getLong(1));
            assertFalse(cursor.isNull(0));
            assertFalse(cursor.isNull(1));
        }

        assertEquals(data,Long>builder()
                .put("one",1L)
                .put("two",2L)
                .put("three",3L)
                .put("ten",10L)
                .put("eleven",11L)
                .put("twelve",12L)
                .build());
    }
}

@Test
public void testCursorMixedOrder()
        throws Exception
{
    RecordSet recordSet = new JdbcRecordSet(jdbcclient,ImmutableList.of(
            columnHandles.get("value"),columnHandles.get("value"),columnHandles.get("text")));

    try (RecordCursor cursor = recordSet.cursor()) {
        assertEquals(cursor.getType(0),BIGINT);
        assertEquals(cursor.getType(1),BIGINT);
        assertEquals(cursor.getType(2),VARCHAR);

        Map<String,Long> data = new LinkedHashMap<>();
        while (cursor.advanceNextPosition()) {
            assertEquals(cursor.getLong(0),cursor.getLong(1));
            data.put(cursor.getSlice(2).toStringUtf8(),cursor.getLong(0));
        }

        assertEquals(data,12L)
                .build());
    }
}

@Test
public void testIdempotentClose()
{
    RecordSet recordSet = new JdbcRecordSet(jdbcclient,columnHandles.get("text")));

    RecordCursor cursor = recordSet.cursor();
    cursor.close();
    cursor.close();
}

private RecordCursor getCursor(JdbcTableHandle jdbcTableHandle,List<JdbcColumnHandle> columns,TupleDomain<ColumnHandle> domain)
        throws InterruptedException
{
    JdbcTableLayoutHandle layoutHandle = new JdbcTableLayoutHandle(jdbcTableHandle,domain);
    ConnectorSplitSource splits = jdbcclient.getSplits(layoutHandle);
    JdbcSplit split = (JdbcSplit) getonlyElement(getFutureValue(splits.getNextBatch(1000)));

    JdbcRecordSetProvider recordSetProvider = new JdbcRecordSetProvider(jdbcclient);
    RecordSet recordSet = recordSetProvider.getRecordSet(SESSION,columns);

    return recordSet.cursor();
}

@Override
public RecordCursor cursor(ConnectorTransactionHandle transactionHandle,TupleDomain<Integer> constraint)
{
    Builder table = InMemoryRecordSet.builder(MetaDATA);
    for (String name : Metadata.getCatalogNames().keySet()) {
        table.addRow(name);
    }
    return table.build().cursor();
}

@Override
public RecordCursor cursor(ConnectorTransactionHandle transactionHandle,connectorSession);
    Optional<String> catalogFilter = FilterUtil.stringFilter(constraint,1);

    Builder table = InMemoryRecordSet.builder(MetaDATA);
    for (String catalog : filter(Metadata.getCatalogNames().keySet(),catalogFilter)) {
        for (String schema : Metadata.listSchemaNames(session,catalog)) {
            table.addRow(schema,catalog);
        }
    }
    return table.build().cursor();
}

@Override
public RecordCursor cursor(ConnectorTransactionHandle transactionHandle,TupleDomain<Integer> constraint)
{
    return InMemoryRecordSet.builder(MetaDATA)
            .addRow("TABLE")
            .addRow("VIEW")
            .build().cursor();
}

@Override
public RecordCursor cursor(ConnectorTransactionHandle transactionHandle,TupleDomain<Integer> constraint)
{
    Builder table = InMemoryRecordSet.builder(CATALOG_TABLE);
    for (Map.Entry<String,String> entry : Metadata.getCatalogNames().entrySet()) {
        table.addRow(entry.getKey(),entry.getValue());
    }
    return table.build().cursor();
}

@Override
public RecordCursor cursor(ConnectorTransactionHandle transactionHandle,TupleDomain<Integer> constraint)
{
    Builder table = InMemoryRecordSet.builder(TASK_TABLE);
    for (TaskInfo taskInfo : taskManager.getAllTaskInfo()) {
        TaskStats stats = taskInfo.getStats();
        table.addRow(
                nodeId,taskInfo.getTaskId().toString(),taskInfo.getTaskId().getStageId().toString(),taskInfo.getTaskId().getQueryId().toString(),taskInfo.getState().toString(),(long) stats.getTotalDrivers(),(long) stats.getQueuedDrivers(),(long) stats.getRunningDrivers(),(long) stats.getCompletedDrivers(),toMillis(stats.getTotalScheduledTime()),toMillis(stats.getTotalcpuTime()),toMillis(stats.getTotalUserTime()),toMillis(stats.getTotalBlockedTime()),toBytes(stats.getRawInputDataSize()),stats.getRawInputPositions(),toBytes(stats.getProcessedInputDataSize()),stats.getProcessedInputPositions(),toBytes(stats.getoutputDataSize()),stats.getoutputPositions(),toTimeStamp(stats.getCreateTime()),toTimeStamp(stats.getFirstStartTime()),toTimeStamp(taskInfo.getLastHeartbeat()),toTimeStamp(stats.getEndTime()));
    }
    return table.build().cursor();
}

@Override
public RecordCursor cursor(ConnectorTransactionHandle transactionHandle,TupleDomain<Integer> constraint)
{
    Builder table = InMemoryRecordSet.builder(NODES_TABLE);
    AllNodes allNodes = nodeManager.getAllNodes();
    addRows(table,allNodes.getActiveNodes(),ACTIVE);
    addRows(table,allNodes.getInactiveNodes(),INACTIVE);
    addRows(table,allNodes.getShuttingDownNodes(),SHUTTING_DOWN);
    return table.build().cursor();
}

@Override
public void project(RecordCursor cursor,BlockBuilder output)
{
    // record cursors have each value in a separate field
    if (cursor.isNull(channelIndex)) {
        output.appendNull();
    }
    else {
        Class<?> javaType = columnType.getJavaType();
        if (javaType == boolean.class) {
            columnType.writeBoolean(output,cursor.getBoolean(channelIndex));
        }
        else if (javaType == long.class) {
            columnType.writeLong(output,cursor.getLong(channelIndex));
        }
        else if (javaType == double.class) {
            columnType.writeDouble(output,cursor.getDouble(channelIndex));
        }
        else if (javaType == Slice.class) {
            Slice slice = cursor.getSlice(channelIndex);
            columnType.writeSlice(output,slice,slice.length());
        }
        else {
            throw new UnsupportedOperationException("not yet implemented: " + javaType);
        }
    }
}

public RecordProjectOperator(OperatorContext operatorContext,List<Type> columnTypes,RecordCursor cursor)
{
    this.operatorContext = requireNonNull(operatorContext,"operatorContext is null");
    this.cursor = requireNonNull(cursor,"cursor is null");

    ImmutableList.Builder<Type> types = ImmutableList.builder();
    for (Type columnType : columnTypes) {
        types.add(columnType);
    }
    this.types = types.build();

    pageBuilder = new PageBuilder(getTypes());
}

private static boolean fieldSetsEqual(RecordCursor cursor,List<Set<Field>> fieldSets)
{
    for (Set<Field> fieldSet : fieldSets) {
        if (!fieldsEquals(cursor,fieldSet)) {
            return false;
        }
    }
    return true;
}

private static boolean fieldsEquals(RecordCursor cursor,Set<Field> fields)
{
    if (fields.size() < 2) {
        return true; // nothing to compare
    }
    Iterator<Field> fieldIterator = fields.iterator();
    Field firstField = fieldIterator.next();
    while (fieldIterator.hasNext()) {
        if (!fieldEquals(cursor,firstField,fieldIterator.next())) {
            return false;
        }
    }
    return true;
}

private static boolean fieldEquals(RecordCursor cursor,Field field1,Field field2)
{
    checkArgument(cursor.getType(field1.getField()).equals(cursor.getType(field2.getField())),"Should only be comparing fields of the same type");

    if (cursor.isNull(field1.getField()) || cursor.isNull(field2.getField())) {
        return false;
    }

    Class<?> javaType = cursor.getType(field1.getField()).getJavaType();
    try {
        if (javaType == long.class) {
            return (boolean) field1.getEqualsMethodHandle().invokeExact(cursor.getLong(field1.getField()),cursor.getLong(field2.getField()));
        }
        else if (javaType == double.class) {
            return (boolean) field1.getEqualsMethodHandle().invokeExact(cursor.getDouble(field1.getField()),cursor.getDouble(field2.getField()));
        }
        else if (javaType == boolean.class) {
            return (boolean) field1.getEqualsMethodHandle().invokeExact(cursor.getBoolean(field1.getField()),cursor.getBoolean(field2.getField()));
        }
        else if (javaType == Slice.class) {
            return (boolean) field1.getEqualsMethodHandle().invokeExact(cursor.getSlice(field1.getField()),cursor.getSlice(field2.getField()));
        }
        else {
            return (boolean) field1.getEqualsMethodHandle().invoke(cursor.getobject(field1.getField()),cursor.getobject(field2.getField()));
        }
    }
    catch (Throwable throwable) {
        throw Throwables.propagate(throwable);
    }
}

private void generateFilterMethod(ClassDeFinition classDeFinition,RowExpression filter)
{
    Parameter session = arg("session",RecordCursor.class);
    MethodDeFinition method = classDeFinition.declareMethod(a(PUBLIC),"filter",type(boolean.class),cursor);

    method.comment("Filter: %s",filter);

    Scope scope = method.getScope();
    Variable wasNullVariable = scope.declareVariable(type(boolean.class),"wasNull");

    BytecodeExpressionVisitor visitor = new BytecodeExpressionVisitor(callSiteBinder,Metadata.getFunctionRegistry());

    LabelNode end = new LabelNode("end");
    method.getBody()
            .comment("boolean wasNull = false;")
            .putvariable(wasNullVariable,false)
            .comment("evaluate filter: " + filter)
            .append(filter.accept(visitor,scope))
            .comment("if (wasNull) return false;")
            .getvariable(wasNullVariable)
            .ifFalseGoto(end)
            .pop(boolean.class)
            .push(false)
            .visitLabel(end)
            .retBoolean();
}

@Override
public void project(RecordCursor cursor,BlockBuilder output)
{
    if (cursor.isNull(channelIndex)) {
        output.appendNull();
    }
    else {
        BIGINT.writeLong(output,cursor.getLong(channelIndex) + 5);
    }
}

private static Object evaluate(Expression expression)
{
    IdentityHashMap<Expression,Type> expressionTypes = getExpressionTypes(TEST_SESSION,MetaDATA,sql_PARSER,SYMBOL_TYPES,expression);
    ExpressionInterpreter interpreter = expressionInterpreter(expression,TEST_SESSION,expressionTypes);

    return interpreter.evaluate((RecordCursor) null);
}

@Test
public void testReverse()
{
    assertFunction("REVERSE('')",VARCHAR,"");
    assertFunction("REVERSE('hello')","olleh");
    assertFunction("REVERSE('Quadratically')","yllacitardauQ");
    assertFunction("REVERSE('racecar')","racecar");
    // Test REVERSE for non-ASCII
    assertFunction("REVERSE('\u4FE1\u5FF5,\u7231,\u5E0C\u671B')","\u671B\u5E0C,\u5FF5\u4FE1");
    assertFunction("REVERSE('\u00D6sterreich')","hcierrets\u00D6");
    assertFunction("REVERSE('na\u00EFve')","ev\u00EFan");
    assertFunction("REVERSE('\uD801\uDC2Dend')","dne\uD801\uDC2D");

    assertFunction("CAST(REVERSE(utf8(from_hex('CE'))) AS VARBINARY)",VARBINARY,new sqlVarbinary(new byte[] {(byte) 0xCE}));
    assertFunction("CAST(REVERSE('hello' || utf8(from_hex('CE'))) AS VARBINARY)",new sqlVarbinary(new byte[] {(byte) 0xCE,'o','l','e','h'}));
}

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

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

private static sqlVarbinary byteArrayToVarbinary(byte[] input)
{
    if (input == null) {
        return null;
    }
    return new sqlVarbinary(input);
}

private static sqlVarbinary expectedChecksum(Type type,Block block)
{
    long result = 0;
    for (int i = 0; i < block.getPositionCount(); i++) {
        if (block.isNull(i)) {
            result += PRIME64;
        }
        else {
            result += type.hash(block,i) * PRIME64;
        }
    }
    return new sqlVarbinary(wrappedLongArray(result).getBytes());
}

@Test
public void testReplace()
{
    assertFunction("REPLACE('aaa','a','aa')","aaaaaa");
    assertFunction("REPLACE('abcdefabcdef','cd','XX')","abXXefabXXef");
    assertFunction("REPLACE('abcdefabcdef','cd')","abefabef");
    assertFunction("REPLACE('123123tech','123')","tech");
    assertFunction("REPLACE('123tech123',"tech");
    assertFunction("REPLACE('222tech','2','3')","333tech");
    assertFunction("REPLACE('0000123','0')","123");
    assertFunction("REPLACE('0000123','0',' ')","    123");
    assertFunction("REPLACE('foo','')","foo");
    assertFunction("REPLACE('foo','','foo',"");
    assertFunction("REPLACE('abc','xx')","xxaxxbxxcxx");
    assertFunction("REPLACE('',"xx");
    assertFunction("REPLACE('',"");
    assertFunction("REPLACE('',"");
    assertFunction("REPLACE('\u4FE1\u5FF5,\u5E0C\u671B',','\u2014')","\u4FE1\u5FF5\u2014\u7231\u2014\u5E0C\u671B");
    assertFunction("REPLACE('::\uD801\uDC2D::',':',"\uD801\uDC2D");
    assertFunction("REPLACE('\u00D6sterreich','\u00D6','Oe')","Oesterreich");

    assertFunction("CAST(REPLACE(utf8(from_hex('CE')),'X') AS VARBINARY)",new sqlVarbinary(new byte[] {'X',(byte) 0xCE,'X'}));

    assertFunction("CAST(REPLACE('abc' || utf8(from_hex('CE')),'X','b','c','X'}));

    assertFunction("CAST(REPLACE(utf8(from_hex('CE')) || 'xyz','x','y','z','X'}));

    assertFunction("CAST(REPLACE('abc' || utf8(from_hex('CE')) || 'xyz','X'}));
}

@Test
public void testLower()
{
    assertFunction("LOWER('')","");
    assertFunction("LOWER('Hello World')","hello world");
    assertFunction("LOWER('WHAT!!')","what!!");
    assertFunction("LOWER('\u00D6STERREICH')",lowerByCodePoint("\u00D6sterreich"));
    assertFunction("LOWER('From\uD801\uDC2DTo')",lowerByCodePoint("from\uD801\uDC2Dto"));

    assertFunction("CAST(LOWER(utf8(from_hex('CE'))) AS VARBINARY)",new sqlVarbinary(new byte[] {(byte) 0xCE}));
    assertFunction("CAST(LOWER('HELLO' || utf8(from_hex('CE'))) AS VARBINARY)",new sqlVarbinary(new byte[] {'h',(byte) 0xCE}));
    assertFunction("CAST(LOWER(utf8(from_hex('CE')) || 'HELLO') AS VARBINARY)",'h','o'}));
    assertFunction("CAST(LOWER(utf8(from_hex('C8BAFF'))) AS VARBINARY)",new sqlVarbinary(new byte[] {(byte) 0xE2,(byte) 0xB1,(byte) 0xA5,(byte) 0xFF}));
}

@Test
public void testUpper()
{
    assertFunction("UPPER('')","");
    assertFunction("UPPER('Hello World')","HELLO WORLD");
    assertFunction("UPPER('what!!')","WHAT!!");
    assertFunction("UPPER('\u00D6sterreich')",upperByCodePoint("\u00D6STERREICH"));
    assertFunction("UPPER('From\uD801\uDC2DTo')",upperByCodePoint("FROM\uD801\uDC2DTO"));

    assertFunction("CAST(UPPER(utf8(from_hex('CE'))) AS VARBINARY)",new sqlVarbinary(new byte[] {(byte) 0xCE}));
    assertFunction("CAST(UPPER('hello' || utf8(from_hex('CE'))) AS VARBINARY)",new sqlVarbinary(new byte[] {'H','E','L','O',(byte) 0xCE}));
    assertFunction("CAST(UPPER(utf8(from_hex('CE')) || 'hello') AS VARBINARY)",'H','O'}));
}

@Test
public void smokedtest()
        throws Exception
{
    assertExecute("cast(true as boolean)",BOOLEAN,true);
    assertExecute("true",true);
    assertExecute("false",false);
    assertExecute("42",BIGINT,42L);
    assertExecute("'foo'","foo");
    assertExecute("4.2",DOUBLE,4.2);
    assertExecute("1 + 1",2L);
    assertExecute("X' 1 f'",new sqlVarbinary(Slices.wrappedBuffer((byte) 0x1f).getBytes()));
    assertExecute("X' '",new sqlVarbinary(new byte[0]));
    assertExecute("bound_long",1234L);
    assertExecute("bound_string","hello");
    assertExecute("bound_double",12.34);
    assertExecute("bound_boolean",true);
    assertExecute("bound_timestamp",new DateTime(2001,8,22,3,4,5,321,UTC).getMillis());
    assertExecute("bound_pattern","%el%");
    assertExecute("bound_null_string",null);
    assertExecute("bound_timestamp_with_timezone",TIMESTAMP_WITH_TIME_ZONE,new sqlTimestampWithTimeZone(new DateTime(1970,1,999,DateTimeZone.UTC).getMillis(),TimeZoneKey.getTimeZoneKey("Z")));
    assertExecute("bound_binary_literal",new sqlVarbinary(new byte[]{(byte) 0xab}));

    // todo enable when null output type is supported
    // assertExecute("null",null);

    Futures.allAsList(futures).get();
}

@Test
public void testMapKeys()
        throws Exception
{
    assertFunction("MAP_KEYS(MAP(ARRAY['1','3'],ARRAY['2','4']))",new ArrayType(VARCHAR),ImmutableList.of("1","3"));
    assertFunction("MAP_KEYS(MAP(ARRAY[1.0,2.0],ARRAY[ARRAY[1,2],ARRAY[3]]))",new ArrayType(DOUBLE),ImmutableList.of(1.0,2.0));
    assertFunction("MAP_KEYS(MAP(ARRAY['puppies'],ARRAY['kittens']))",ImmutableList.of("puppies"));
    assertFunction("MAP_KEYS(MAP(ARRAY[TRUE],ARRAY[2]))",new ArrayType(BOOLEAN),ImmutableList.of(true));
    assertFunction("MAP_KEYS(MAP(ARRAY[from_unixtime(1)],ARRAY[1.0]))",new ArrayType(TIMESTAMP),ImmutableList.of(new sqlTimestamp(1000,TEST_SESSION.getTimeZoneKey())));
    assertFunction("MAP_KEYS(MAP(ARRAY[CAST('puppies' as varbinary)],new ArrayType(VARBINARY),ImmutableList.of(new sqlVarbinary("puppies".getBytes(UTF_8))));
    assertFunction("MAP_KEYS(MAP(ARRAY[1,new ArrayType(BIGINT),ImmutableList.of(1L,2L));
    assertFunction("MAP_KEYS(MAP(ARRAY[1,4],ARRAY[MAP(ARRAY[2],ARRAY[3]),MAP(ARRAY[5],ARRAY[6])]))",4L));
    assertFunction("MAP_KEYS(MAP(ARRAY [ARRAY [1],ARRAY [2,3]],ARRAY [ARRAY [3,ARRAY [5]]))",new ArrayType(new ArrayType(BIGINT)),ImmutableList.of(ImmutableList.of(1L),ImmutableList.of(2L,3L)));
}

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

@Override
public Object getobjectValue(ConnectorSession session,block.getSliceLength(position)).getBytes());
}

private static sqlVarbinary sqlBinary(byte[] bytes)
{
    return new sqlVarbinary(bytes);
}

@Test
public void testBinaryLiteral()
        throws Exception
{
    assertFunction("X'58F7'",new sqlVarbinary(new byte[]{(byte) 0x58,(byte) 0xF7}));
}

private static sqlVarbinary sqlVarbinary(String value)
{
    return new sqlVarbinary(value.getBytes(UTF_8));
}

private static sqlVarbinary sqlVarbinaryHex(String value)
{
    return new sqlVarbinary(base16().decode(value));
}

public TestvarbinaryType()
{
    super(VARBINARY,sqlVarbinary.class,createTestBlock());
}

protected void checkPageSource(ConnectorPageSource pageSource,List<TestColumn> testColumns,List<Type> types)
        throws IOException
{
    try {
        MaterializedResult result = materializeSourceDataStream(SESSION,pageSource,types);

        for (MaterializedRow row : result) {
            for (int i = 0,testColumnsSize = testColumns.size(); i < testColumnsSize; i++) {
                TestColumn testColumn = testColumns.get(i);
                Type type = types.get(i);

                Object actualValue = row.getField(i);
                Object expectedValue = testColumn.getExpectedValue();
                if (actualValue == null) {
                    assertEquals(null,expectedValue,String.format("Expected non-null for column %d",i));
                }
                else if (testColumn.getobjectInspector().getTypeName().equals("float") ||
                        testColumn.getobjectInspector().getTypeName().equals("double")) {
                    assertEquals((double) actualValue,(double) expectedValue,EPSILON);
                }
                else if (testColumn.getobjectInspector().getTypeName().equals("date")) {
                    sqlDate expectedDate = new sqlDate(((Long) expectedValue).intValue());
                    assertEquals(actualValue,expectedDate);
                }
                else if (testColumn.getobjectInspector().getTypeName().equals("timestamp")) {
                    sqlTimestamp expectedTimestamp = new sqlTimestamp((Long) expectedValue,SESSION.getTimeZoneKey());
                    assertEquals(actualValue,expectedTimestamp);
                }
                else if (testColumn.getobjectInspector().getCategory() == Category.PRIMITIVE) {
                    if (expectedValue instanceof Slice) {
                        expectedValue = ((Slice) expectedValue).toStringUtf8();
                    }

                    if (actualValue instanceof Slice) {
                        actualValue = ((Slice) actualValue).toStringUtf8();
                    }
                    if (actualValue instanceof sqlVarbinary) {
                        actualValue = new String(((sqlVarbinary) actualValue).getBytes(),UTF_8);
                    }
                    assertEquals(actualValue,String.format("Wrong value for column %d",i));
                }
                else {
                    BlockBuilder builder = type.createBlockBuilder(new BlockBuilderStatus(),1);
                    type.writeObject(builder,expectedValue);
                    expectedValue = type.getobjectValue(SESSION,builder.build(),0);
                    assertEquals(actualValue,String.format("Wrong value for column %s",testColumn.getName()));
                }
            }
        }
    }
    finally {
        pageSource.close();
    }
}