class QueryProcedureProducer {

    public QueryProcedure createQueryProcedure(String sql) {
        RelNode originalLogicalPlan = buildLogicalPlan(sql);
        RelNode optimizedPlan = optimizeLogicalPlan(originalLogicalPlan);
        TreeNode treeNode = optimizedPlan.accept(new LogicalViewShuttle());

        SubtreeSyncopator subtreeSyncopator = new SubtreeSyncopator(optimizedPlan, funcTable, builder);
        Map<RelNode, AbstractMap.SimpleEntry<String, RelOptTable>> resultRelNode =
            subtreeSyncopator.rootNodeSchemas;

        LoadProcedure procedure = createLoadProcedure();

        TransformProcedure transformProcedure =
            new DataSetTransformProcedure(procedure, subtreeSyncopator.getRoot());

        List<ExtractProcedure> extractProcedures = new ArrayList<>();
        for (Map.Entry<RelNode, AbstractMap.SimpleEntry<String, RelOptTable>> entry :
            resultRelNode.entrySet()) {
            extractProcedures.add(PreparedExtractProcedure.createSpecificProcedure(
                transformProcedure, ((RelOptTableImpl) entry.getValue().getValue()),
                config, entry.getKey(),
                entry.getValue().getKey(),
                (output instanceof SqlInsertOutput)
                    ? ((SqlInsertOutput) output).getSelect() : output, JdbcSourceInfo.getSourceMap()));
            //if table is mongo table and set connection information properties to builder object so as to set
            // parameters when do spark-submit job
            if (((RelOptTableImpl) entry.getValue().getValue()).getTable() instanceof MongoTable) {
                builder.setProperties(((MongoTable) ((RelOptTableImpl) entry.getValue().getValue()).getTable())
                    .getProperties());
            }
        }
        QueryProcedure queryProcedure = new ProcedurePortFire(extractProcedures).optimize();
        queryProcedure.setTreeNode(treeNode);
        return queryProcedure;
    }

    public  QueryProcedureProducer(String jsonPath, Builder builder);
    public  QueryProcedureProducer(String jsonPath);

}