package edu.uci.ics.asterix.api.common;

import java.io.PrintWriter;
import java.rmi.RemoteException;
import java.util.ArrayList;
import java.util.List;

import org.json.JSONException;

import edu.uci.ics.asterix.api.common.Job.SubmissionMode;
import edu.uci.ics.asterix.aql.base.Statement;
import edu.uci.ics.asterix.aql.expression.Query;
import edu.uci.ics.asterix.aql.expression.visitor.AQLPrintVisitor;
import edu.uci.ics.asterix.aql.rewrites.AqlRewriter;
import edu.uci.ics.asterix.aql.translator.DdlTranslator;
import edu.uci.ics.asterix.common.config.GlobalConfig;
import edu.uci.ics.asterix.common.config.OptimizationConfUtil;
import edu.uci.ics.asterix.common.exceptions.AsterixException;
import edu.uci.ics.asterix.dataflow.data.common.AqlExpressionTypeComputer;
import edu.uci.ics.asterix.dataflow.data.common.AqlMergeAggregationExpressionFactory;
import edu.uci.ics.asterix.dataflow.data.common.AqlNullableTypeComputer;
import edu.uci.ics.asterix.dataflow.data.common.AqlPartialAggregationTypeComputer;
import edu.uci.ics.asterix.file.DatasetOperations;
import edu.uci.ics.asterix.file.FeedOperations;
import edu.uci.ics.asterix.file.IndexOperations;
import edu.uci.ics.asterix.formats.base.IDataFormat;
import edu.uci.ics.asterix.jobgen.AqlLogicalExpressionJobGen;
import edu.uci.ics.asterix.metadata.MetadataManager;
import edu.uci.ics.asterix.metadata.MetadataTransactionContext;
import edu.uci.ics.asterix.metadata.declared.AqlCompiledMetadataDeclarations;
import edu.uci.ics.asterix.metadata.declared.AqlMetadataProvider;
import edu.uci.ics.asterix.optimizer.base.RuleCollections;
import edu.uci.ics.asterix.runtime.job.listener.JobEventListenerFactory;
import edu.uci.ics.asterix.transaction.management.exception.ACIDException;
import edu.uci.ics.asterix.transaction.management.service.transaction.TransactionIDFactory;
import edu.uci.ics.asterix.transaction.management.service.transaction.TransactionManagementConstants.LockManagerConstants.LockMode;
import edu.uci.ics.asterix.translator.AqlExpressionToPlanTranslator;
import edu.uci.ics.asterix.translator.DmlTranslator;
import edu.uci.ics.asterix.translator.DmlTranslator.CompiledBeginFeedStatement;
import edu.uci.ics.asterix.translator.DmlTranslator.CompiledControlFeedStatement;
import edu.uci.ics.asterix.translator.DmlTranslator.CompiledCreateIndexStatement;
import edu.uci.ics.asterix.translator.DmlTranslator.CompiledDeleteStatement;
import edu.uci.ics.asterix.translator.DmlTranslator.CompiledInsertStatement;
import edu.uci.ics.asterix.translator.DmlTranslator.CompiledLoadFromFileStatement;
import edu.uci.ics.asterix.translator.DmlTranslator.CompiledWriteFromQueryResultStatement;
import edu.uci.ics.asterix.translator.DmlTranslator.ICompiledDmlStatement;
import edu.uci.ics.hyracks.algebricks.compiler.api.HeuristicCompilerFactoryBuilder;
import edu.uci.ics.hyracks.algebricks.compiler.api.ICompiler;
import edu.uci.ics.hyracks.algebricks.compiler.api.ICompilerFactory;
import edu.uci.ics.hyracks.algebricks.compiler.rewriter.rulecontrollers.SequentialFixpointRuleController;
import edu.uci.ics.hyracks.algebricks.compiler.rewriter.rulecontrollers.SequentialOnceRuleController;
import edu.uci.ics.hyracks.algebricks.core.algebra.base.ILogicalPlanAndMetadata;
import edu.uci.ics.hyracks.algebricks.core.algebra.base.IOptimizationContext;
import edu.uci.ics.hyracks.algebricks.core.algebra.expressions.IExpressionEvalSizeComputer;
import edu.uci.ics.hyracks.algebricks.core.algebra.expressions.IExpressionTypeComputer;
import edu.uci.ics.hyracks.algebricks.core.algebra.expressions.IMergeAggregationExpressionFactory;
import edu.uci.ics.hyracks.algebricks.core.algebra.expressions.INullableTypeComputer;
import edu.uci.ics.hyracks.algebricks.core.algebra.prettyprint.LogicalOperatorPrettyPrintVisitor;
import edu.uci.ics.hyracks.algebricks.core.algebra.prettyprint.PlanPrettyPrinter;
import edu.uci.ics.hyracks.algebricks.core.api.constraints.AlgebricksPartitionConstraint;
import edu.uci.ics.hyracks.algebricks.core.api.exceptions.AlgebricksException;
import edu.uci.ics.hyracks.algebricks.core.rewriter.base.AbstractRuleController;
import edu.uci.ics.hyracks.algebricks.core.rewriter.base.AlgebricksOptimizationContext;
import edu.uci.ics.hyracks.algebricks.core.rewriter.base.IAlgebraicRewriteRule;
import edu.uci.ics.hyracks.algebricks.core.rewriter.base.IOptimizationContextFactory;
import edu.uci.ics.hyracks.algebricks.core.rewriter.base.PhysicalOptimizationConfig;
import edu.uci.ics.hyracks.algebricks.core.utils.Pair;
import edu.uci.ics.hyracks.api.client.IHyracksClientConnection;
import edu.uci.ics.hyracks.api.job.JobId;
import edu.uci.ics.hyracks.api.job.JobSpecification;

public class APIFramework {

    private static List<Pair<AbstractRuleController, List<IAlgebraicRewriteRule>>> buildDefaultLogicalRewrites() {
        List<Pair<AbstractRuleController, List<IAlgebraicRewriteRule>>> defaultLogicalRewrites = new ArrayList<Pair<AbstractRuleController, List<IAlgebraicRewriteRule>>>();
        SequentialFixpointRuleController seqCtrlNoDfs = new SequentialFixpointRuleController(false);
        SequentialFixpointRuleController seqCtrlFullDfs = new SequentialFixpointRuleController(true);
        SequentialOnceRuleController seqOnceCtrl = new SequentialOnceRuleController(true);
        defaultLogicalRewrites.add(new Pair<AbstractRuleController, List<IAlgebraicRewriteRule>>(seqOnceCtrl,
                RuleCollections.buildTypeInferenceRuleCollection()));
        defaultLogicalRewrites.add(new Pair<AbstractRuleController, List<IAlgebraicRewriteRule>>(seqCtrlFullDfs,
                RuleCollections.buildNormalizationRuleCollection()));
        defaultLogicalRewrites.add(new Pair<AbstractRuleController, List<IAlgebraicRewriteRule>>(seqCtrlNoDfs,
                RuleCollections.buildCondPushDownAndJoinInferenceRuleCollection()));
        defaultLogicalRewrites.add(new Pair<AbstractRuleController, List<IAlgebraicRewriteRule>>(seqCtrlFullDfs,
                RuleCollections.buildLoadFieldsRuleCollection()));
        // fj
        defaultLogicalRewrites.add(new Pair<AbstractRuleController, List<IAlgebraicRewriteRule>>(seqCtrlFullDfs,
                RuleCollections.buildFuzzyJoinRuleCollection()));
        //
        defaultLogicalRewrites.add(new Pair<AbstractRuleController, List<IAlgebraicRewriteRule>>(seqCtrlFullDfs,
                RuleCollections.buildNormalizationRuleCollection()));
        defaultLogicalRewrites.add(new Pair<AbstractRuleController, List<IAlgebraicRewriteRule>>(seqCtrlNoDfs,
                RuleCollections.buildCondPushDownAndJoinInferenceRuleCollection()));
        defaultLogicalRewrites.add(new Pair<AbstractRuleController, List<IAlgebraicRewriteRule>>(seqCtrlFullDfs,
                RuleCollections.buildLoadFieldsRuleCollection()));
        defaultLogicalRewrites.add(new Pair<AbstractRuleController, List<IAlgebraicRewriteRule>>(seqOnceCtrl,
                RuleCollections.buildDataExchangeRuleCollection()));
        defaultLogicalRewrites.add(new Pair<AbstractRuleController, List<IAlgebraicRewriteRule>>(seqCtrlNoDfs,
                RuleCollections.buildConsolidationRuleCollection()));
        defaultLogicalRewrites.add(new Pair<AbstractRuleController, List<IAlgebraicRewriteRule>>(seqCtrlNoDfs,
                RuleCollections.buildOpPushDownRuleCollection()));
        return defaultLogicalRewrites;
    }

    private static List<Pair<AbstractRuleController, List<IAlgebraicRewriteRule>>> buildDefaultPhysicalRewrites() {
        List<Pair<AbstractRuleController, List<IAlgebraicRewriteRule>>> defaultPhysicalRewrites = new ArrayList<Pair<AbstractRuleController, List<IAlgebraicRewriteRule>>>();
        SequentialOnceRuleController seqOnceCtrl = new SequentialOnceRuleController(true);
        SequentialOnceRuleController seqOnceTopLevel = new SequentialOnceRuleController(false);
        defaultPhysicalRewrites.add(new Pair<AbstractRuleController, List<IAlgebraicRewriteRule>>(seqOnceCtrl,
                RuleCollections.buildPhysicalRewritesAllLevelsRuleCollection()));
        defaultPhysicalRewrites.add(new Pair<AbstractRuleController, List<IAlgebraicRewriteRule>>(seqOnceTopLevel,
                RuleCollections.buildPhysicalRewritesTopLevelRuleCollection()));
        defaultPhysicalRewrites.add(new Pair<AbstractRuleController, List<IAlgebraicRewriteRule>>(seqOnceCtrl,
                RuleCollections.prepareForJobGenRuleCollection()));
        return defaultPhysicalRewrites;
    }

    private static class AqlOptimizationContextFactory implements IOptimizationContextFactory {

        public static final AqlOptimizationContextFactory INSTANCE = new AqlOptimizationContextFactory();

        private AqlOptimizationContextFactory() {
        }

        @Override
        public IOptimizationContext createOptimizationContext(int varCounter, int frameSize,
                IExpressionEvalSizeComputer expressionEvalSizeComputer,
                IMergeAggregationExpressionFactory mergeAggregationExpressionFactory,
                IExpressionTypeComputer expressionTypeComputer, INullableTypeComputer nullableTypeComputer,
                PhysicalOptimizationConfig physicalOptimizationConfig) {
            return new AlgebricksOptimizationContext(varCounter, frameSize, expressionEvalSizeComputer,
                    mergeAggregationExpressionFactory, expressionTypeComputer, nullableTypeComputer,
                    physicalOptimizationConfig);
        }

    }

    public enum DisplayFormat {
        TEXT,
        HTML
    }

    public static String compileDdlStatements(IHyracksClientConnection hcc, Query query, PrintWriter out,
            SessionConfig pc, DisplayFormat pdf) throws AsterixException, AlgebricksException, JSONException,
            RemoteException, ACIDException {
        // Begin a transaction against the metadata.
        // Lock the metadata in X mode to protect against other DDL and DML.
        MetadataTransactionContext mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
        MetadataManager.INSTANCE.lock(mdTxnCtx, LockMode.EXCLUSIVE);
        try {
            DdlTranslator ddlt = new DdlTranslator(mdTxnCtx, query.getPrologDeclList(), out, pc, pdf);
            ddlt.translate(hcc, false);
            MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
            return ddlt.getCompiledDeclarations().getDataverseName();
        } catch (Exception e) {
            MetadataManager.INSTANCE.abortTransaction(mdTxnCtx);
            e.printStackTrace();
            throw new AlgebricksException(e);
        }
    }

    public static Job[] compileDmlStatements(String dataverseName, Query query, PrintWriter out, SessionConfig pc,
            DisplayFormat pdf) throws AsterixException, AlgebricksException, JSONException, RemoteException,
            ACIDException {

        // Begin a transaction against the metadata.
        // Lock the metadata in S mode to protect against other DDL
        // modifications.
        MetadataTransactionContext mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
        MetadataManager.INSTANCE.lock(mdTxnCtx, LockMode.SHARED);
        try {
            DmlTranslator dmlt = new DmlTranslator(mdTxnCtx, query.getPrologDeclList());
            dmlt.translate();

            if (dmlt.getCompiledDmlStatements().size() == 0) {
                // There is no DML to run. Consider the transaction against the
                // metadata successful.
                MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
                return new Job[] {};
            }

            List<Job> dmlJobs = new ArrayList<Job>();
            AqlCompiledMetadataDeclarations metadata = dmlt.getCompiledDeclarations();

            if (!metadata.isConnectedToDataverse())
                metadata.connectToDataverse(metadata.getDataverseName());

            for (ICompiledDmlStatement stmt : dmlt.getCompiledDmlStatements()) {
                switch (stmt.getKind()) {
                    case LOAD_FROM_FILE: {
                        CompiledLoadFromFileStatement stmtLoad = (CompiledLoadFromFileStatement) stmt;
                        dmlJobs.addAll(DatasetOperations.createLoadDatasetJobSpec(stmtLoad, metadata));
                        break;
                    }
                    case WRITE_FROM_QUERY_RESULT: {
                        CompiledWriteFromQueryResultStatement stmtLoad = (CompiledWriteFromQueryResultStatement) stmt;
                        SessionConfig sc2 = new SessionConfig(pc.getPort(), true, pc.isPrintExprParam(),
                                pc.isPrintRewrittenExprParam(), pc.isPrintLogicalPlanParam(),
                                pc.isPrintOptimizedLogicalPlanParam(), pc.isPrintPhysicalOpsOnly(), pc.isPrintJob());
                        sc2.setGenerateJobSpec(true);
                        Pair<AqlCompiledMetadataDeclarations, JobSpecification> mj = compileQueryInternal(mdTxnCtx,
                                dataverseName, stmtLoad.getQuery(), stmtLoad.getVarCounter(),
                                stmtLoad.getDatasetName(), metadata, sc2, out, pdf,
                                Statement.Kind.WRITE_FROM_QUERY_RESULT);
                        dmlJobs.add(new Job(mj.second));
                        break;
                    }
                    case INSERT: {
                        CompiledInsertStatement stmtLoad = (CompiledInsertStatement) stmt;
                        SessionConfig sc2 = new SessionConfig(pc.getPort(), true, pc.isPrintExprParam(),
                                pc.isPrintRewrittenExprParam(), pc.isPrintLogicalPlanParam(),
                                pc.isPrintOptimizedLogicalPlanParam(), pc.isPrintPhysicalOpsOnly(), pc.isPrintJob());
                        sc2.setGenerateJobSpec(true);
                        Pair<AqlCompiledMetadataDeclarations, JobSpecification> mj = compileQueryInternal(mdTxnCtx,
                                dataverseName, stmtLoad.getQuery(), stmtLoad.getVarCounter(),
                                stmtLoad.getDatasetName(), metadata, sc2, out, pdf, Statement.Kind.INSERT);
                        dmlJobs.add(new Job(mj.second));
                        break;
                    }
                    case DELETE: {
                        CompiledDeleteStatement stmtLoad = (CompiledDeleteStatement) stmt;
                        SessionConfig sc2 = new SessionConfig(pc.getPort(), true, pc.isPrintExprParam(),
                                pc.isPrintRewrittenExprParam(), pc.isPrintLogicalPlanParam(),
                                pc.isPrintOptimizedLogicalPlanParam(), pc.isPrintPhysicalOpsOnly(), pc.isPrintJob());
                        sc2.setGenerateJobSpec(true);
                        Pair<AqlCompiledMetadataDeclarations, JobSpecification> mj = compileQueryInternal(mdTxnCtx,
                                dataverseName, stmtLoad.getQuery(), stmtLoad.getVarCounter(),
                                stmtLoad.getDatasetName(), metadata, sc2, out, pdf, Statement.Kind.DELETE);
                        dmlJobs.add(new Job(mj.second));
                        break;
                    }
                    case CREATE_INDEX: {
                        CompiledCreateIndexStatement cis = (CompiledCreateIndexStatement) stmt;
                        JobSpecification jobSpec = IndexOperations.buildCreateIndexJobSpec(cis, metadata);
                        dmlJobs.add(new Job(jobSpec));
                        break;
                    }

                    case BEGIN_FEED: {
                        CompiledBeginFeedStatement cbfs = (CompiledBeginFeedStatement) stmt;
                        SessionConfig sc2 = new SessionConfig(pc.getPort(), true, pc.isPrintExprParam(),
                                pc.isPrintRewrittenExprParam(), pc.isPrintLogicalPlanParam(),
                                pc.isPrintOptimizedLogicalPlanParam(), pc.isPrintPhysicalOpsOnly(), pc.isPrintJob());
                        sc2.setGenerateJobSpec(true);
                        Pair<AqlCompiledMetadataDeclarations, JobSpecification> mj = compileQueryInternal(mdTxnCtx,
                                dataverseName, cbfs.getQuery(), cbfs.getVarCounter(), cbfs.getDatasetName().getValue(),
                                metadata, sc2, out, pdf, Statement.Kind.BEGIN_FEED);
                        dmlJobs.add(new Job(mj.second));
                        break;

                    }

                    case CONTROL_FEED: {
                        CompiledControlFeedStatement cfs = (CompiledControlFeedStatement) stmt;
                        Job job = new Job(FeedOperations.buildControlFeedJobSpec(cfs, metadata),
                                SubmissionMode.ASYNCHRONOUS);
                        dmlJobs.add(job);
                        break;
                    }
                    default: {
                        throw new IllegalArgumentException();
                    }
                }
            }
            if (pc.isPrintJob()) {
                int i = 0;
                for (Job js : dmlJobs) {
                    out.println("<H1>Hyracks job number " + i + ":</H1>");
                    out.println("<PRE>");
                    out.println(js.getJobSpec().toJSON().toString(1));
                    out.println(js.getJobSpec().getUserConstraints());
                    out.println(js.getSubmissionMode());
                    out.println("</PRE>");
                    i++;
                }
            }
            // close connection to dataverse
            if (metadata.isConnectedToDataverse())
                metadata.disconnectFromDataverse();

            Job[] jobs = dmlJobs.toArray(new Job[0]);
            MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
            return jobs;
        } catch (AsterixException e) {
            MetadataManager.INSTANCE.abortTransaction(mdTxnCtx);
            throw e;
        } catch (AlgebricksException e) {
            MetadataManager.INSTANCE.abortTransaction(mdTxnCtx);
            throw e;
        } catch (JSONException e) {
            MetadataManager.INSTANCE.abortTransaction(mdTxnCtx);
            throw e;
        } catch (Exception e) {
            MetadataManager.INSTANCE.abortTransaction(mdTxnCtx);
            throw new AsterixException(e);
        }
    }

    public static Pair<AqlCompiledMetadataDeclarations, JobSpecification> compileQuery(String dataverseName, Query q,
            int varCounter, String outputDatasetName, AqlCompiledMetadataDeclarations metadataDecls, SessionConfig pc,
            PrintWriter out, DisplayFormat pdf, Statement.Kind dmlKind) throws AsterixException, AlgebricksException,
            JSONException, RemoteException, ACIDException {
        MetadataTransactionContext mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
        try {
            MetadataManager.INSTANCE.lock(mdTxnCtx, LockMode.SHARED);
            Pair<AqlCompiledMetadataDeclarations, JobSpecification> result = compileQueryInternal(mdTxnCtx,
                    dataverseName, q, varCounter, outputDatasetName, metadataDecls, pc, out, pdf, dmlKind);
            MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
            return result;
        } catch (AsterixException e) {
            MetadataManager.INSTANCE.abortTransaction(mdTxnCtx);
            throw e;
        } catch (AlgebricksException e) {
            MetadataManager.INSTANCE.abortTransaction(mdTxnCtx);
            throw e;
        } catch (JSONException e) {
            MetadataManager.INSTANCE.abortTransaction(mdTxnCtx);
            throw e;
        } catch (RemoteException e) {
            MetadataManager.INSTANCE.abortTransaction(mdTxnCtx);
            throw e;
        } catch (ACIDException e) {
            MetadataManager.INSTANCE.abortTransaction(mdTxnCtx);
            throw e;
        } catch (Exception e) {
            MetadataManager.INSTANCE.abortTransaction(mdTxnCtx);
            throw new AsterixException(e);
        }
    }

    public static Pair<AqlCompiledMetadataDeclarations, JobSpecification> compileQueryInternal(
            MetadataTransactionContext mdTxnCtx, String dataverseName, Query q, int varCounter,
            String outputDatasetName, AqlCompiledMetadataDeclarations metadataDecls, SessionConfig pc, PrintWriter out,
            DisplayFormat pdf, Statement.Kind dmlKind) throws AsterixException, AlgebricksException, JSONException,
            RemoteException, ACIDException {

        if (!pc.isPrintPhysicalOpsOnly() && pc.isPrintExprParam()) {
            out.println();
            switch (pdf) {
                case HTML: {
                    out.println("<H1>Expression tree:</H1>");
                    out.println("<PRE>");
                    break;
                }
                case TEXT: {
                    out.println("----------Expression tree:");
                    break;
                }
            }
            if (q != null) {
                q.accept(new AQLPrintVisitor(out), 0);
            }
            switch (pdf) {
                case HTML: {
                    out.println("</PRE>");
                    break;
                }
            }
        }
        AqlRewriter rw = new AqlRewriter(q, varCounter, mdTxnCtx, dataverseName);
        rw.rewrite();
        Query rwQ = rw.getExpr();
        if (!pc.isPrintPhysicalOpsOnly() && pc.isPrintRewrittenExprParam()) {
            out.println();

            switch (pdf) {
                case HTML: {
                    out.println("<H1>Rewriten expression tree:</H1>");
                    out.println("<PRE>");
                    break;
                }
                case TEXT: {
                    out.println("----------Rewritten expression:");
                    break;
                }
            }

            if (q != null) {
                rwQ.accept(new AQLPrintVisitor(out), 0);
            }

            switch (pdf) {
                case HTML: {
                    out.println("</PRE>");
                    break;
                }
            }

        }
        long txnId = TransactionIDFactory.generateTransactionId();
        AqlExpressionToPlanTranslator t = new AqlExpressionToPlanTranslator(txnId, mdTxnCtx, rw.getVarCounter(),
                outputDatasetName, dmlKind);

        ILogicalPlanAndMetadata planAndMetadata = t.translate(rwQ, metadataDecls);
        boolean isWriteTransaction = false;
        AqlMetadataProvider mp = (AqlMetadataProvider) planAndMetadata.getMetadataProvider();
        if (metadataDecls == null) {
            metadataDecls = mp.getMetadataDeclarations();
        }
        isWriteTransaction = mp.isWriteTransaction();

        if (outputDatasetName == null && metadataDecls.getOutputFile() == null) {
            throw new AlgebricksException("Unknown output file: `write output to nc:\"file\"' statement missing.");
        }

        LogicalOperatorPrettyPrintVisitor pvisitor = new LogicalOperatorPrettyPrintVisitor();
        if (!pc.isPrintPhysicalOpsOnly() && pc.isPrintLogicalPlanParam()) {

            switch (pdf) {
                case HTML: {
                    out.println("<H1>Logical plan:</H1>");
                    out.println("<PRE>");
                    break;
                }
                case TEXT: {
                    out.println("----------Logical plan:");
                    break;
                }
            }

            if (q != null) {
                StringBuilder buffer = new StringBuilder();
                PlanPrettyPrinter.printPlan(planAndMetadata.getPlan(), buffer, pvisitor, 0);
                out.print(buffer);
            }

            switch (pdf) {
                case HTML: {
                    out.println("</PRE>");
                    break;
                }
            }
        }

        int frameSize = GlobalConfig.DEFAULT_FRAME_SIZE;
        String frameSizeStr = System.getProperty(GlobalConfig.FRAME_SIZE_PROPERTY);
        if (frameSizeStr != null) {
            int fz = -1;
            try {
                fz = Integer.parseInt(frameSizeStr);
            } catch (NumberFormatException nfe) {
                GlobalConfig.ASTERIX_LOGGER.warning("Wrong frame size size argument. Picking default value ("
                        + GlobalConfig.DEFAULT_FRAME_SIZE + ") instead.\n");
                throw new AlgebricksException(nfe);
            }
            if (fz >= 0) {
                frameSize = fz;
            }
        }

        HeuristicCompilerFactoryBuilder builder = new HeuristicCompilerFactoryBuilder(
                AqlOptimizationContextFactory.INSTANCE);
        builder.setLogicalRewrites(buildDefaultLogicalRewrites());
        builder.setPhysicalRewrites(buildDefaultPhysicalRewrites());
        IDataFormat format = metadataDecls.getFormat();
        ICompilerFactory compilerFactory = builder.create();
        builder.setFrameSize(frameSize);
        builder.setExpressionEvalSizeComputer(format.getExpressionEvalSizeComputer());
        builder.setIMergeAggregationExpressionFactory(new AqlMergeAggregationExpressionFactory());
        builder.setPartialAggregationTypeComputer(new AqlPartialAggregationTypeComputer());
        builder.setExpressionTypeComputer(AqlExpressionTypeComputer.INSTANCE);
        builder.setNullableTypeComputer(AqlNullableTypeComputer.INSTANCE);

        OptimizationConfUtil.getPhysicalOptimizationConfig().setFrameSize(frameSize);
        builder.setPhysicalOptimizationConfig(OptimizationConfUtil.getPhysicalOptimizationConfig());
        ICompiler compiler = compilerFactory.createCompiler(planAndMetadata.getPlan(),
                planAndMetadata.getMetadataProvider(), t.getVarCounter());
        if (pc.isOptimize()) {
            compiler.optimize();
            if (true) {
                StringBuilder buffer = new StringBuilder();
                PlanPrettyPrinter.printPhysicalOps(planAndMetadata.getPlan(), buffer, 0);
                out.print(buffer);
            } else if (pc.isPrintOptimizedLogicalPlanParam()) {
                switch (pdf) {
                    case HTML: {
                        out.println("<H1>Optimized logical plan:</H1>");
                        out.println("<PRE>");
                        break;
                    }
                    case TEXT: {
                        out.println("----------Optimized plan ");
                        break;
                    }
                }

                if (q != null) {
                    StringBuilder buffer = new StringBuilder();
                    PlanPrettyPrinter.printPlan(planAndMetadata.getPlan(), buffer, pvisitor, 0);
                    out.print(buffer);
                }
                switch (pdf) {
                    case HTML: {
                        out.println("</PRE>");
                        break;
                    }
                }
            }
        }

        if (!pc.isGenerateJobSpec()) {
            // Job spec not requested. Consider transaction against metadata
            // committed.
            MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
            return null;
        }

        AlgebricksPartitionConstraint clusterLocs = planAndMetadata.getClusterLocations();
        builder.setBinaryBooleanInspector(format.getBinaryBooleanInspector());
        builder.setBinaryIntegerInspector(format.getBinaryIntegerInspector());
        builder.setClusterLocations(clusterLocs);
        builder.setComparatorFactoryProvider(format.getBinaryComparatorFactoryProvider());
        builder.setExprJobGen(AqlLogicalExpressionJobGen.INSTANCE);
        builder.setHashFunctionFactoryProvider(format.getBinaryHashFunctionFactoryProvider());
        builder.setNullWriterFactory(format.getNullWriterFactory());
        builder.setPrinterProvider(format.getPrinterFactoryProvider());
        builder.setSerializerDeserializerProvider(format.getSerdeProvider());
        builder.setTypeTraitProvider(format.getTypeTraitProvider());
        builder.setNormalizedKeyComputerFactoryProvider(format.getNormalizedKeyComputerFactoryProvider());

        JobSpecification spec = compiler.createJob(AsterixAppContextInfoImpl.INSTANCE);
        // set the job event listener
        spec.setJobletEventListenerFactory(new JobEventListenerFactory(txnId, isWriteTransaction));
        if (pc.isPrintJob()) {
            switch (pdf) {
                case HTML: {
                    out.println("<H1>Hyracks job:</H1>");
                    out.println("<PRE>");
                    break;
                }
                case TEXT: {
                    out.println("----------Hyracks job:");
                    break;
                }
            }
            if (q != null) {
                out.println(spec.toJSON().toString(1));
                out.println(spec.getUserConstraints());
            }
            switch (pdf) {
                case HTML: {
                    out.println("</PRE>");
                    break;
                }
            }
        }
        return new Pair<AqlCompiledMetadataDeclarations, JobSpecification>(metadataDecls, spec);
    }

    public static void executeJobArray(IHyracksClientConnection hcc, JobSpecification[] specs, PrintWriter out,
            DisplayFormat pdf) throws Exception {
        for (int i = 0; i < specs.length; i++) {
            specs[i].setMaxReattempts(0);
            JobId jobId = hcc.createJob(GlobalConfig.HYRACKS_APP_NAME, specs[i]);
            long startTime = System.currentTimeMillis();
            hcc.start(jobId);
            hcc.waitForCompletion(jobId);
            long endTime = System.currentTimeMillis();
            double duration = (endTime - startTime) / 1000.00;
            out.println("<PRE>Duration: " + duration + "</PRE>");
        }

    }

    public static void executeJobArray(IHyracksClientConnection hcc, Job[] jobs, PrintWriter out, DisplayFormat pdf)
            throws Exception {
        for (int i = 0; i < jobs.length; i++) {
            jobs[i].getJobSpec().setMaxReattempts(0);
            JobId jobId = hcc.createJob(GlobalConfig.HYRACKS_APP_NAME, jobs[i].getJobSpec());
            long startTime = System.currentTimeMillis();
            try {
                hcc.start(jobId);
                if (jobs[i].getSubmissionMode() == SubmissionMode.ASYNCHRONOUS) {
                    continue;
                }
                hcc.waitForCompletion(jobId);
            } catch (Exception e) {
                e.printStackTrace();
                continue;
            }
            long endTime = System.currentTimeMillis();
            double duration = (endTime - startTime) / 1000.00;
            out.println("<PRE>Duration: " + duration + "</PRE>");
        }

    }

}