/*
 * Copyright 2009-2010 by The Regents of the University of California
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * you may obtain a copy of the License from
 * 
 *     http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package edu.uci.ics.hyracks.control.cc;

import java.util.ArrayList;
import java.util.EnumSet;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.logging.Logger;

import jol.core.Runtime;
import jol.types.basic.BasicTupleSet;
import jol.types.basic.Tuple;
import jol.types.basic.TupleSet;
import jol.types.exception.BadKeyException;
import jol.types.exception.UpdateException;
import jol.types.table.BasicTable;
import jol.types.table.EventTable;
import jol.types.table.Function;
import jol.types.table.Key;
import jol.types.table.TableName;

import org.json.JSONArray;
import org.json.JSONObject;

import edu.uci.ics.hyracks.api.comm.Endpoint;
import edu.uci.ics.hyracks.api.constraints.AbsoluteLocationConstraint;
import edu.uci.ics.hyracks.api.constraints.ChoiceLocationConstraint;
import edu.uci.ics.hyracks.api.constraints.ExplicitPartitionConstraint;
import edu.uci.ics.hyracks.api.constraints.LocationConstraint;
import edu.uci.ics.hyracks.api.constraints.PartitionConstraint;
import edu.uci.ics.hyracks.api.constraints.PartitionCountConstraint;
import edu.uci.ics.hyracks.api.dataflow.ActivityNodeId;
import edu.uci.ics.hyracks.api.dataflow.ConnectorDescriptorId;
import edu.uci.ics.hyracks.api.dataflow.Direction;
import edu.uci.ics.hyracks.api.dataflow.IActivityGraphBuilder;
import edu.uci.ics.hyracks.api.dataflow.IActivityNode;
import edu.uci.ics.hyracks.api.dataflow.IConnectorDescriptor;
import edu.uci.ics.hyracks.api.dataflow.IOperatorDescriptor;
import edu.uci.ics.hyracks.api.dataflow.OperatorDescriptorId;
import edu.uci.ics.hyracks.api.dataflow.PortInstanceId;
import edu.uci.ics.hyracks.api.job.JobFlag;
import edu.uci.ics.hyracks.api.job.JobPlan;
import edu.uci.ics.hyracks.api.job.JobSpecification;
import edu.uci.ics.hyracks.api.job.JobStatus;

public class JOLJobManagerImpl implements IJobManager {
    private static final Logger LOGGER = Logger.getLogger(JOLJobManagerImpl.class.getName());

    public static final String JOL_SCOPE = "hyrackscc";

    private static final String SCHEDULER_OLG_FILE = "edu/uci/ics/hyracks/control/cc/scheduler.olg";

    private final Runtime jolRuntime;

    private final LinkedBlockingQueue<Runnable> jobQueue;

    private final JobTable jobTable;

    private final JobQueueThread jobQueueThread;

    private final OperatorDescriptorTable odTable;

    private final OperatorLocationTable olTable;

    private final OperatorCloneCountTable ocTable;

    private final ConnectorDescriptorTable cdTable;

    private final ActivityNodeTable anTable;

    private final ActivityConnectionTable acTable;

    private final ActivityBlockedTable abTable;

    private final JobStartTable jobStartTable;

    private final JobCleanUpTable jobCleanUpTable;

    private final JobCleanUpCompleteTable jobCleanUpCompleteTable;

    private final StartMessageTable startMessageTable;

    private final StageletCompleteTable stageletCompleteTable;

    private final StageletFailureTable stageletFailureTable;

    private final AvailableNodesTable availableNodesTable;

    private final RankedAvailableNodesTable rankedAvailableNodesTable;

    private final FailedNodesTable failedNodesTable;

    private final AbortMessageTable abortMessageTable;

    private final AbortNotifyTable abortNotifyTable;

    private final ExpandPartitionCountConstraintTableFunction expandPartitionCountConstraintFunction;

    private final ProfileUpdateTable puTable;

    private final List<String> rankedAvailableNodes;

    private final IJobManagerQueryInterface qi;

    public JOLJobManagerImpl(final ClusterControllerService ccs, final Runtime jolRuntime) throws Exception {
        this.jolRuntime = jolRuntime;
        jobQueue = new LinkedBlockingQueue<Runnable>();
        jobQueueThread = new JobQueueThread();
        jobQueueThread.start();

        this.jobTable = new JobTable(jolRuntime);
        this.odTable = new OperatorDescriptorTable(jolRuntime);
        this.olTable = new OperatorLocationTable(jolRuntime);
        this.ocTable = new OperatorCloneCountTable(jolRuntime);
        this.cdTable = new ConnectorDescriptorTable(jolRuntime);
        this.anTable = new ActivityNodeTable(jolRuntime);
        this.acTable = new ActivityConnectionTable(jolRuntime);
        this.abTable = new ActivityBlockedTable(jolRuntime);
        this.jobStartTable = new JobStartTable();
        this.jobCleanUpTable = new JobCleanUpTable(jolRuntime);
        this.jobCleanUpCompleteTable = new JobCleanUpCompleteTable();
        this.startMessageTable = new StartMessageTable(jolRuntime);
        this.stageletCompleteTable = new StageletCompleteTable(jolRuntime);
        this.stageletFailureTable = new StageletFailureTable(jolRuntime);
        this.availableNodesTable = new AvailableNodesTable(jolRuntime);
        this.rankedAvailableNodesTable = new RankedAvailableNodesTable(jolRuntime);
        this.failedNodesTable = new FailedNodesTable(jolRuntime);
        this.abortMessageTable = new AbortMessageTable(jolRuntime);
        this.abortNotifyTable = new AbortNotifyTable(jolRuntime);
        this.expandPartitionCountConstraintFunction = new ExpandPartitionCountConstraintTableFunction();
        this.puTable = new ProfileUpdateTable();
        this.rankedAvailableNodes = new ArrayList<String>();

        jolRuntime.catalog().register(jobTable);
        jolRuntime.catalog().register(odTable);
        jolRuntime.catalog().register(olTable);
        jolRuntime.catalog().register(ocTable);
        jolRuntime.catalog().register(cdTable);
        jolRuntime.catalog().register(anTable);
        jolRuntime.catalog().register(acTable);
        jolRuntime.catalog().register(abTable);
        jolRuntime.catalog().register(jobStartTable);
        jolRuntime.catalog().register(jobCleanUpTable);
        jolRuntime.catalog().register(jobCleanUpCompleteTable);
        jolRuntime.catalog().register(startMessageTable);
        jolRuntime.catalog().register(stageletCompleteTable);
        jolRuntime.catalog().register(stageletFailureTable);
        jolRuntime.catalog().register(availableNodesTable);
        jolRuntime.catalog().register(rankedAvailableNodesTable);
        jolRuntime.catalog().register(failedNodesTable);
        jolRuntime.catalog().register(abortMessageTable);
        jolRuntime.catalog().register(abortNotifyTable);
        jolRuntime.catalog().register(expandPartitionCountConstraintFunction);
        jolRuntime.catalog().register(puTable);

        jobTable.register(new JobTable.Callback() {
            @Override
            public void deletion(TupleSet arg0) {
                jobTable.notifyAll();
            }

            @Override
            public void insertion(TupleSet arg0) {
                jobTable.notifyAll();
            }
        });

        startMessageTable.register(new StartMessageTable.Callback() {
            @Override
            public void deletion(TupleSet tuples) {

            }

            @SuppressWarnings("unchecked")
            @Override
            public void insertion(TupleSet tuples) {
                for (final Tuple t : tuples) {
                    jobQueue.add(new Runnable() {
                        @Override
                        public void run() {
                            try {
                                Object[] data = t.toArray();
                                UUID jobId = (UUID) data[0];
                                UUID stageId = (UUID) data[1];
                                Integer attempt = (Integer) data[2];
                                String appName = (String) data[3];
                                JobPlan plan = (JobPlan) data[4];
                                Set<List> ts = (Set<List>) data[5];
                                Map<OperatorDescriptorId, Set<Integer>> opPartitions = new HashMap<OperatorDescriptorId, Set<Integer>>();
                                for (List t2 : ts) {
                                    Object[] t2Data = t2.toArray();
                                    Set<List> activityInfoSet = (Set<List>) t2Data[1];
                                    for (List l : activityInfoSet) {
                                        Object[] lData = l.toArray();
                                        ActivityNodeId aid = (ActivityNodeId) lData[0];
                                        Set<Integer> opParts = opPartitions.get(aid.getOperatorDescriptorId());
                                        if (opParts == null) {
                                            opParts = new HashSet<Integer>();
                                            opPartitions.put(aid.getOperatorDescriptorId(), opParts);
                                        }
                                        opParts.add((Integer) lData[1]);
                                    }
                                }
                                ClusterControllerService.Phase1Installer[] p1is = new ClusterControllerService.Phase1Installer[ts
                                        .size()];
                                int i = 0;
                                for (List t2 : ts) {
                                    Object[] t2Data = t2.toArray();
                                    Map<ActivityNodeId, Set<Integer>> tasks = new HashMap<ActivityNodeId, Set<Integer>>();
                                    Set<List> activityInfoSet = (Set<List>) t2Data[1];
                                    for (List l : activityInfoSet) {
                                        Object[] lData = l.toArray();
                                        ActivityNodeId aid = (ActivityNodeId) lData[0];
                                        Set<Integer> aParts = tasks.get(aid);
                                        if (aParts == null) {
                                            aParts = new HashSet<Integer>();
                                            tasks.put(aid, aParts);
                                        }
                                        aParts.add((Integer) lData[1]);
                                    }
                                    p1is[i++] = new ClusterControllerService.Phase1Installer((String) t2Data[0], jobId,
                                            appName, plan, stageId, attempt, tasks, opPartitions);
                                }
                                LOGGER.info("Stage start - Phase 1");
                                Map<PortInstanceId, Endpoint> globalPortMap = ccs.runRemote(p1is,
                                        new ClusterControllerService.PortMapMergingAccumulator());

                                ClusterControllerService.Phase2Installer[] p2is = new ClusterControllerService.Phase2Installer[ts
                                        .size()];
                                ClusterControllerService.Phase3Installer[] p3is = new ClusterControllerService.Phase3Installer[ts
                                        .size()];
                                ClusterControllerService.StageStarter[] ss = new ClusterControllerService.StageStarter[ts
                                        .size()];
                                i = 0;
                                for (List t2 : ts) {
                                    Object[] t2Data = t2.toArray();
                                    Map<ActivityNodeId, Set<Integer>> tasks = new HashMap<ActivityNodeId, Set<Integer>>();
                                    Set<List> activityInfoSet = (Set<List>) t2Data[1];
                                    for (List l : activityInfoSet) {
                                        Object[] lData = l.toArray();
                                        ActivityNodeId aid = (ActivityNodeId) lData[0];
                                        Set<Integer> aParts = tasks.get(aid);
                                        if (aParts == null) {
                                            aParts = new HashSet<Integer>();
                                            tasks.put(aid, aParts);
                                        }
                                        aParts.add((Integer) lData[1]);
                                    }
                                    p2is[i] = new ClusterControllerService.Phase2Installer((String) t2Data[0], jobId,
                                            appName, plan, stageId, tasks, opPartitions, globalPortMap);
                                    p3is[i] = new ClusterControllerService.Phase3Installer((String) t2Data[0], jobId,
                                            stageId);
                                    ss[i] = new ClusterControllerService.StageStarter((String) t2Data[0], jobId,
                                            stageId);
                                    ++i;
                                }
                                LOGGER.info("Stage start - Phase 2");
                                ccs.runRemote(p2is, null);
                                LOGGER.info("Stage start - Phase 3");
                                ccs.runRemote(p3is, null);
                                LOGGER.info("Stage start");
                                ccs.runRemote(ss, null);
                                LOGGER.info("Stage started");
                            } catch (Exception e) {
                            }
                        }
                    });
                }
            }
        });

        jobCleanUpTable.register(new JobCleanUpTable.Callback() {
            @Override
            public void deletion(TupleSet tuples) {
            }

            @SuppressWarnings("unchecked")
            @Override
            public void insertion(TupleSet tuples) {
                for (final Tuple t : tuples) {
                    jobQueue.add(new Runnable() {
                        @Override
                        public void run() {
                            try {
                                Object[] data = t.toArray();
                                UUID jobId = (UUID) data[0];
                                Set<String> ts = (Set<String>) data[1];
                                ClusterControllerService.JobCompleteNotifier[] jcns = new ClusterControllerService.JobCompleteNotifier[ts
                                        .size()];
                                int i = 0;
                                for (String n : ts) {
                                    jcns[i++] = new ClusterControllerService.JobCompleteNotifier(n, jobId);
                                }
                                try {
                                    ccs.runRemote(jcns, null);
                                } finally {
                                    BasicTupleSet jccTuples = new BasicTupleSet(JobCleanUpCompleteTable
                                            .createTuple(jobId));
                                    jolRuntime.schedule(JOL_SCOPE, JobCleanUpCompleteTable.TABLE_NAME, jccTuples, null);
                                    jolRuntime.evaluate();
                                }
                            } catch (Exception e) {
                            }
                        }
                    });
                }
            }
        });

        abortMessageTable.register(new AbortMessageTable.Callback() {
            @Override
            public void deletion(TupleSet tuples) {

            }

            @SuppressWarnings("unchecked")
            @Override
            public void insertion(TupleSet tuples) {
                for (final Tuple t : tuples) {
                    jobQueue.add(new Runnable() {
                        @Override
                        public void run() {
                            try {
                                Object[] data = t.toArray();
                                UUID jobId = (UUID) data[0];
                                UUID stageId = (UUID) data[1];
                                Integer attempt = (Integer) data[2];
                                Set<List> ts = (Set<List>) data[4];
                                ClusterControllerService.JobletAborter[] jas = new ClusterControllerService.JobletAborter[ts
                                        .size()];
                                int i = 0;
                                BasicTupleSet notificationTuples = new BasicTupleSet();
                                for (List t2 : ts) {
                                    Object[] t2Data = t2.toArray();
                                    String nodeId = (String) t2Data[0];
                                    jas[i++] = new ClusterControllerService.JobletAborter(nodeId, jobId, stageId,
                                            attempt);
                                    notificationTuples.add(AbortNotifyTable
                                            .createTuple(jobId, stageId, nodeId, attempt));
                                }
                                try {
                                    ccs.runRemote(jas, null);
                                } finally {
                                    jolRuntime.schedule(JOL_SCOPE, AbortNotifyTable.TABLE_NAME, notificationTuples,
                                            null);
                                    jolRuntime.evaluate();
                                }
                            } catch (Exception e) {
                            }
                        }
                    });
                }
            }
        });

        jolRuntime.install(JOL_SCOPE, ClassLoader.getSystemResource(SCHEDULER_OLG_FILE));
        jolRuntime.evaluate();

        qi = new QueryInterfaceImpl();
    }

    @Override
    public UUID createJob(String appName, JobSpecification jobSpec, EnumSet<JobFlag> jobFlags) throws Exception {
        final UUID jobId = UUID.randomUUID();

        final JobPlanBuilder builder = new JobPlanBuilder();
        builder.init(jobSpec, jobFlags);

        final BasicTupleSet anTuples = new BasicTupleSet();
        final BasicTupleSet acTuples = new BasicTupleSet();
        final BasicTupleSet abTuples = new BasicTupleSet();
        IActivityGraphBuilder gBuilder = new IActivityGraphBuilder() {
            @Override
            public void addTask(IActivityNode task) {
                anTuples.add(ActivityNodeTable.createTuple(jobId, task));
                builder.addTask(task);
            }

            @Override
            public void addTargetEdge(int operatorOutputIndex, IActivityNode task, int taskOutputIndex) {
                acTuples.add(ActivityConnectionTable.createTuple(jobId, task, Direction.OUTPUT, operatorOutputIndex,
                        taskOutputIndex));
                builder.addTargetEdge(operatorOutputIndex, task, taskOutputIndex);
            }

            @Override
            public void addSourceEdge(int operatorInputIndex, IActivityNode task, int taskInputIndex) {
                acTuples.add(ActivityConnectionTable.createTuple(jobId, task, Direction.INPUT, operatorInputIndex,
                        taskInputIndex));
                builder.addSourceEdge(operatorInputIndex, task, taskInputIndex);
            }

            @Override
            public void addBlockingEdge(IActivityNode blocker, IActivityNode blocked) {
                abTuples.add(ActivityBlockedTable.createTuple(jobId, blocker, blocked));
                builder.addBlockingEdge(blocker, blocked);
            }
        };

        BasicTupleSet odTuples = new BasicTupleSet();
        BasicTupleSet olTuples = new BasicTupleSet();
        BasicTupleSet ocTuples = new BasicTupleSet();
        for (Map.Entry<OperatorDescriptorId, IOperatorDescriptor> e : jobSpec.getOperatorMap().entrySet()) {
            IOperatorDescriptor od = e.getValue();
            int nPartitions = addPartitionConstraintTuples(jobId, od, olTuples, ocTuples);
            odTuples.add(OperatorDescriptorTable.createTuple(jobId, nPartitions, od));
            od.contributeTaskGraph(gBuilder);
        }

        BasicTupleSet cdTuples = new BasicTupleSet();
        for (Map.Entry<ConnectorDescriptorId, IConnectorDescriptor> e : jobSpec.getConnectorMap().entrySet()) {
            cdTuples.add(ConnectorDescriptorTable.createTuple(jobId, jobSpec, e.getValue()));
        }

        BasicTupleSet jobTuples = new BasicTupleSet(JobTable.createInitialJobTuple(jobId, appName, jobSpec,
                builder.getPlan()));

        jolRuntime.schedule(JOL_SCOPE, JobTable.TABLE_NAME, jobTuples, null);
        jolRuntime.schedule(JOL_SCOPE, OperatorDescriptorTable.TABLE_NAME, odTuples, null);
        jolRuntime.schedule(JOL_SCOPE, OperatorLocationTable.TABLE_NAME, olTuples, null);
        jolRuntime.schedule(JOL_SCOPE, OperatorCloneCountTable.TABLE_NAME, ocTuples, null);
        jolRuntime.schedule(JOL_SCOPE, ConnectorDescriptorTable.TABLE_NAME, cdTuples, null);
        jolRuntime.schedule(JOL_SCOPE, ActivityNodeTable.TABLE_NAME, anTuples, null);
        jolRuntime.schedule(JOL_SCOPE, ActivityConnectionTable.TABLE_NAME, acTuples, null);
        jolRuntime.schedule(JOL_SCOPE, ActivityBlockedTable.TABLE_NAME, abTuples, null);

        jolRuntime.evaluate();

        return jobId;
    }

    private int addPartitionConstraintTuples(UUID jobId, IOperatorDescriptor od, BasicTupleSet olTuples,
            BasicTupleSet ocTuples) {
        PartitionConstraint pc = od.getPartitionConstraint();

        switch (pc.getPartitionConstraintType()) {
            case COUNT:
                int count = ((PartitionCountConstraint) pc).getCount();
                ocTuples.add(OperatorCloneCountTable.createTuple(jobId, od.getOperatorId(), count));
                return count;

            case EXPLICIT:
                LocationConstraint[] locationConstraints = ((ExplicitPartitionConstraint) pc).getLocationConstraints();
                for (int i = 0; i < locationConstraints.length; ++i) {
                    addLocationConstraintTuple(olTuples, jobId, od.getOperatorId(), i, locationConstraints[i], 0);
                }
                return locationConstraints.length;
        }
        throw new IllegalArgumentException();
    }

    private void addLocationConstraintTuple(BasicTupleSet olTuples, UUID jobId, OperatorDescriptorId opId, int i,
            LocationConstraint locationConstraint, int benefit) {
        switch (locationConstraint.getConstraintType()) {
            case ABSOLUTE:
                String nodeId = ((AbsoluteLocationConstraint) locationConstraint).getLocationId();
                olTuples.add(OperatorLocationTable.createTuple(jobId, opId, nodeId, i, benefit));
                break;

            case CHOICE:
                int index = 0;
                for (LocationConstraint lc : ((ChoiceLocationConstraint) locationConstraint).getChoices()) {
                    addLocationConstraintTuple(olTuples, jobId, opId, i, lc, benefit - index);
                    index++;
                }
        }
    }

    @Override
    public void reportProfile(String id, Map<UUID, Map<String, Long>> counterDump) throws Exception {
        BasicTupleSet puTuples = new BasicTupleSet();
        for (Map.Entry<UUID, Map<String, Long>> e : counterDump.entrySet()) {
            puTuples.add(ProfileUpdateTable.createTuple(e.getKey(), id, e.getValue()));
        }
        jolRuntime.schedule(JOL_SCOPE, ProfileUpdateTable.TABLE_NAME, puTuples, null);
    }

    @Override
    public JobStatus getJobStatus(UUID jobId) {
        synchronized (jobTable) {
            try {
                Tuple jobTuple = jobTable.lookupJob(jobId);
                if (jobTuple == null) {
                    return null;
                }
                return (JobStatus) jobTuple.value(JobTable.JOBSTATUS_FIELD_INDEX);
            } catch (BadKeyException e) {
                throw new RuntimeException(e);
            }
        }
    }

    @Override
    public synchronized void notifyNodeFailure(String nodeId) throws Exception {
        int len = rankedAvailableNodes.size();
        int delIndex = -1;
        for (int i = 0; i < len; ++i) {
            if (nodeId.equals(rankedAvailableNodes.get(i))) {
                delIndex = i;
                break;
            }
        }
        if (delIndex < 0) {
            return;
        }
        BasicTupleSet delRANTuples = new BasicTupleSet();
        delRANTuples.add(RankedAvailableNodesTable.createTuple(nodeId, delIndex));

        BasicTupleSet insRANTuples = new BasicTupleSet();
        for (int i = delIndex + 1; i < len; ++i) {
            insRANTuples.add(RankedAvailableNodesTable.createTuple(rankedAvailableNodes.get(i), i - 1));
        }

        rankedAvailableNodes.remove(delIndex);

        jolRuntime.schedule(JOL_SCOPE, RankedAvailableNodesTable.TABLE_NAME, insRANTuples, delRANTuples);

        BasicTupleSet unavailableTuples = new BasicTupleSet(AvailableNodesTable.createTuple(nodeId));

        jolRuntime.schedule(JOL_SCOPE, AvailableNodesTable.TABLE_NAME, null, unavailableTuples);

        jolRuntime.evaluate();

        BasicTupleSet failedTuples = new BasicTupleSet(FailedNodesTable.createTuple(nodeId));

        jolRuntime.schedule(JOL_SCOPE, FailedNodesTable.TABLE_NAME, failedTuples, null);

        jolRuntime.evaluate();
    }

    @Override
    public synchronized void notifyStageletComplete(UUID jobId, UUID stageId, int attempt, String nodeId,
            Map<String, Long> statistics) throws Exception {
        BasicTupleSet scTuples = new BasicTupleSet();
        scTuples.add(StageletCompleteTable.createTuple(jobId, stageId, nodeId, attempt, statistics));

        jolRuntime.schedule(JOL_SCOPE, StageletCompleteTable.TABLE_NAME, scTuples, null);

        jolRuntime.evaluate();
    }

    @Override
    public synchronized void notifyStageletFailure(UUID jobId, UUID stageId, int attempt, String nodeId)
            throws Exception {
        BasicTupleSet sfTuples = new BasicTupleSet();
        sfTuples.add(StageletFailureTable.createTuple(jobId, stageId, nodeId, attempt));

        jolRuntime.schedule(JOL_SCOPE, StageletFailureTable.TABLE_NAME, sfTuples, null);

        jolRuntime.evaluate();
    }

    @Override
    public void start(UUID jobId) throws Exception {
        BasicTupleSet jsTuples = new BasicTupleSet();
        jsTuples.add(JobStartTable.createTuple(jobId, System.currentTimeMillis()));

        jolRuntime.schedule(JOL_SCOPE, JobStartTable.TABLE_NAME, jsTuples, null);

        jolRuntime.evaluate();
    }

    @Override
    public synchronized void registerNode(String nodeId) throws Exception {
        rankedAvailableNodes.add(nodeId);
        BasicTupleSet insRANTuples = new BasicTupleSet();
        insRANTuples.add(RankedAvailableNodesTable.createTuple(nodeId, rankedAvailableNodes.size() - 1));

        jolRuntime.schedule(JOL_SCOPE, RankedAvailableNodesTable.TABLE_NAME, insRANTuples, null);

        BasicTupleSet availableTuples = new BasicTupleSet(AvailableNodesTable.createTuple(nodeId));

        jolRuntime.schedule(JOL_SCOPE, AvailableNodesTable.TABLE_NAME, availableTuples, null);

        jolRuntime.evaluate();

        BasicTupleSet unfailedTuples = new BasicTupleSet(FailedNodesTable.createTuple(nodeId));

        jolRuntime.schedule(JOL_SCOPE, FailedNodesTable.TABLE_NAME, null, unfailedTuples);

        jolRuntime.evaluate();
    }

    @Override
    public void waitForCompletion(UUID jobId) throws Exception {
        synchronized (jobTable) {
            Tuple jobTuple = null;
            while ((jobTuple = jobTable.lookupJob(jobId)) != null
                    && jobTuple.value(JobTable.JOBSTATUS_FIELD_INDEX) != JobStatus.TERMINATED) {
                jobTable.wait();
            }
        }
    }

    @Override
    public IJobManagerQueryInterface getQueryInterface() {
        return qi;
    }

    public void visitJobs(ITupleProcessor processor) throws Exception {
        for (Tuple t : jobTable.tuples()) {
            processor.process(t);
        }
    }

    public void visitJob(UUID jobId, ITupleProcessor processor) throws Exception {
        Tuple job = jobTable.lookupJob(jobId);
        if (job != null) {
            processor.process(job);
        }
    }

    /*
     * declare(job, keys(0), {JobId, Status, JobSpec, JobPlan})
     */
    private static class JobTable extends BasicTable {
        private static TableName TABLE_NAME = new TableName(JOL_SCOPE, "job");

        private static Key PRIMARY_KEY = new Key(0);

        @SuppressWarnings("unchecked")
        private static final Class[] SCHEMA = new Class[] { UUID.class, String.class, JobStatus.class,
                JobSpecification.class, JobPlan.class, Map.class };

        public static final int JOBID_FIELD_INDEX = 0;
        public static final int APPNAME_FIELD_INDEX = 1;
        public static final int JOBSTATUS_FIELD_INDEX = 2;
        public static final int JOBSPEC_FIELD_INDEX = 3;
        public static final int JOBPLAN_FIELD_INDEX = 4;
        public static final int STATISTICS_FIELD_INDEX = 5;

        public JobTable(Runtime context) {
            super(context, TABLE_NAME, PRIMARY_KEY, SCHEMA);
        }

        @SuppressWarnings("unchecked")
        static Tuple createInitialJobTuple(UUID jobId, String appName, JobSpecification jobSpec, JobPlan plan) {
            return new Tuple(jobId, appName, JobStatus.INITIALIZED, jobSpec, plan, new HashMap());
        }

        Tuple lookupJob(UUID jobId) throws BadKeyException {
            TupleSet set = primary().lookupByKey(jobId);
            if (set.isEmpty()) {
                return null;
            }
            return (Tuple) set.toArray()[0];
        }
    }

    /*
     * declare(operatordescriptor, keys(0, 1), {JobId, ODId, OperatorDescriptor})
     */
    private static class OperatorDescriptorTable extends BasicTable {
        private static TableName TABLE_NAME = new TableName(JOL_SCOPE, "operatordescriptor");

        private static Key PRIMARY_KEY = new Key(0, 1);

        @SuppressWarnings("unchecked")
        private static final Class[] SCHEMA = new Class[] { UUID.class, OperatorDescriptorId.class, Integer.class,
                IOperatorDescriptor.class };

        public OperatorDescriptorTable(Runtime context) {
            super(context, TABLE_NAME, PRIMARY_KEY, SCHEMA);
        }

        static Tuple createTuple(UUID jobId, int nPartitions, IOperatorDescriptor od) {
            return new Tuple(jobId, od.getOperatorId(), nPartitions, od);
        }
    }

    /*
     * declare(operatorlocation, keys(0, 1), {JobId, ODId, NodeId})
     */
    private static class OperatorLocationTable extends BasicTable {
        private static TableName TABLE_NAME = new TableName(JOL_SCOPE, "operatorlocation");

        private static Key PRIMARY_KEY = new Key();

        @SuppressWarnings("unchecked")
        private static final Class[] SCHEMA = new Class[] { UUID.class, OperatorDescriptorId.class, String.class,
                Integer.class, Integer.class };

        public OperatorLocationTable(Runtime context) {
            super(context, TABLE_NAME, PRIMARY_KEY, SCHEMA);
        }

        static Tuple createTuple(UUID jobId, OperatorDescriptorId opId, String nodeId, int partition, int benefit) {
            return new Tuple(jobId, opId, nodeId, partition, benefit);
        }
    }

    /*
     * declare(operatorclonecount, keys(0, 1), {JobId, ODId, Count})
     */
    private static class OperatorCloneCountTable extends BasicTable {
        private static TableName TABLE_NAME = new TableName(JOL_SCOPE, "operatorclonecount");

        private static Key PRIMARY_KEY = new Key();

        @SuppressWarnings("unchecked")
        private static final Class[] SCHEMA = new Class[] { UUID.class, OperatorDescriptorId.class, Integer.class };

        public OperatorCloneCountTable(Runtime context) {
            super(context, TABLE_NAME, PRIMARY_KEY, SCHEMA);
        }

        static Tuple createTuple(UUID jobId, OperatorDescriptorId opId, int cloneCount) {
            return new Tuple(jobId, opId, cloneCount);
        }
    }

    /*
     * declare(connectordescriptor, keys(0, 1), {JobId, CDId, SrcODId, SrcPort, DestODId, DestPort, ConnectorDescriptor})
     */
    private static class ConnectorDescriptorTable extends BasicTable {
        private static TableName TABLE_NAME = new TableName(JOL_SCOPE, "connectordescriptor");

        private static Key PRIMARY_KEY = new Key(0, 1);

        @SuppressWarnings("unchecked")
        private static final Class[] SCHEMA = new Class[] { UUID.class, ConnectorDescriptorId.class,
                OperatorDescriptorId.class, Integer.class, OperatorDescriptorId.class, Integer.class,
                IConnectorDescriptor.class };

        public ConnectorDescriptorTable(Runtime context) {
            super(context, TABLE_NAME, PRIMARY_KEY, SCHEMA);
        }

        static Tuple createTuple(UUID jobId, JobSpecification jobSpec, IConnectorDescriptor conn) {
            IOperatorDescriptor srcOD = jobSpec.getProducer(conn);
            int srcPort = jobSpec.getProducerOutputIndex(conn);
            IOperatorDescriptor destOD = jobSpec.getConsumer(conn);
            int destPort = jobSpec.getConsumerInputIndex(conn);
            Tuple cdTuple = new Tuple(jobId, conn.getConnectorId(), srcOD.getOperatorId(), srcPort,
                    destOD.getOperatorId(), destPort, conn);
            return cdTuple;
        }
    }

    /*
     * declare(activitynode, keys(0, 1, 2), {JobId, OperatorId, ActivityId, ActivityNode})
     */
    private static class ActivityNodeTable extends BasicTable {
        private static TableName TABLE_NAME = new TableName(JOL_SCOPE, "activitynode");

        private static Key PRIMARY_KEY = new Key(0, 1, 2);

        @SuppressWarnings("unchecked")
        private static final Class[] SCHEMA = new Class[] { UUID.class, OperatorDescriptorId.class,
                ActivityNodeId.class, IActivityNode.class };

        public ActivityNodeTable(Runtime context) {
            super(context, TABLE_NAME, PRIMARY_KEY, SCHEMA);
        }

        static Tuple createTuple(UUID jobId, IActivityNode aNode) {
            return new Tuple(jobId, aNode.getActivityNodeId().getOperatorDescriptorId(), aNode.getActivityNodeId(),
                    aNode);
        }
    }

    /*
     * declare(activityconnection, keys(0, 1, 2, 3), {JobId, OperatorId, Integer, Direction, ActivityNodeId, Integer})
     */
    private static class ActivityConnectionTable extends BasicTable {
        private static TableName TABLE_NAME = new TableName(JOL_SCOPE, "activityconnection");

        private static Key PRIMARY_KEY = new Key(0, 1, 2, 3);

        @SuppressWarnings("unchecked")
        private static final Class[] SCHEMA = new Class[] { UUID.class, OperatorDescriptorId.class, Integer.class,
                Direction.class, ActivityNodeId.class, Integer.class };

        public ActivityConnectionTable(Runtime context) {
            super(context, TABLE_NAME, PRIMARY_KEY, SCHEMA);
        }

        static Tuple createTuple(UUID jobId, IActivityNode aNode, Direction direction, int odPort, int activityPort) {
            return new Tuple(jobId, aNode.getActivityNodeId().getOperatorDescriptorId(), odPort, direction,
                    aNode.getActivityNodeId(), activityPort);
        }
    }

    /*
     * declare(activityblocked, keys(0, 1, 2, 3), {JobId, OperatorId, BlockerActivityId, BlockedActivityId})
     */
    private static class ActivityBlockedTable extends BasicTable {
        private static TableName TABLE_NAME = new TableName(JOL_SCOPE, "activityblocked");

        private static Key PRIMARY_KEY = new Key(0, 1, 2, 3, 4);

        @SuppressWarnings("unchecked")
        private static final Class[] SCHEMA = new Class[] { UUID.class, OperatorDescriptorId.class,
                ActivityNodeId.class, OperatorDescriptorId.class, ActivityNodeId.class };

        public ActivityBlockedTable(Runtime context) {
            super(context, TABLE_NAME, PRIMARY_KEY, SCHEMA);
        }

        static Tuple createTuple(UUID jobId, IActivityNode blocker, IActivityNode blocked) {
            ActivityNodeId blockerANId = blocker.getActivityNodeId();
            OperatorDescriptorId blockerODId = blockerANId.getOperatorDescriptorId();
            ActivityNodeId blockedANId = blocked.getActivityNodeId();
            OperatorDescriptorId blockedODId = blockedANId.getOperatorDescriptorId();
            return new Tuple(jobId, blockerODId, blockerANId, blockedODId, blockedANId);
        }
    }

    /*
     * declare(jobstart, keys(0), {JobId, SubmitTime})
     */
    private static class JobStartTable extends EventTable {
        private static TableName TABLE_NAME = new TableName(JOL_SCOPE, "jobstart");

        @SuppressWarnings("unchecked")
        private static final Class[] SCHEMA = new Class[] { UUID.class, Long.class };

        public JobStartTable() {
            super(TABLE_NAME, SCHEMA);
        }

        static Tuple createTuple(UUID jobId, long submitTime) {
            return new Tuple(jobId, submitTime);
        }
    }

    /*
     * declare(startmessage, keys(0, 1), {JobId, StageId, JobPlan, TupleSet})
     */
    private static class StartMessageTable extends BasicTable {
        private static TableName TABLE_NAME = new TableName(JOL_SCOPE, "startmessage");

        private static Key PRIMARY_KEY = new Key(0, 1, 2);

        @SuppressWarnings("unchecked")
        private static final Class[] SCHEMA = new Class[] { UUID.class, UUID.class, Integer.class, String.class,
                JobPlan.class, Set.class };

        public StartMessageTable(Runtime context) {
            super(context, TABLE_NAME, PRIMARY_KEY, SCHEMA);
        }
    }

    /*
     * declare(jobcleanup, keys(0), {JobId, Set<NodeId>})
     */
    private static class JobCleanUpTable extends BasicTable {
        private static TableName TABLE_NAME = new TableName(JOL_SCOPE, "jobcleanup");

        private static Key PRIMARY_KEY = new Key(0);

        @SuppressWarnings("unchecked")
        private static final Class[] SCHEMA = new Class[] { UUID.class, Set.class };

        public JobCleanUpTable(Runtime context) {
            super(context, TABLE_NAME, PRIMARY_KEY, SCHEMA);
        }
    }

    /*
     * declare(jobcleanupcomplete, keys(0), {JobId})
     */
    private static class JobCleanUpCompleteTable extends EventTable {
        private static TableName TABLE_NAME = new TableName(JOL_SCOPE, "jobcleanupcomplete");

        @SuppressWarnings("unchecked")
        private static final Class[] SCHEMA = new Class[] { UUID.class };

        public JobCleanUpCompleteTable() {
            super(TABLE_NAME, SCHEMA);
        }

        public static Tuple createTuple(UUID jobId) {
            return new Tuple(jobId);
        }
    }

    /*
     * declare(stageletcomplete, keys(0, 1, 2, 3), {JobId, StageId, NodeId, Attempt})
     */
    private static class StageletCompleteTable extends BasicTable {
        private static TableName TABLE_NAME = new TableName(JOL_SCOPE, "stageletcomplete");

        private static Key PRIMARY_KEY = new Key(0, 1, 2, 3);

        @SuppressWarnings("unchecked")
        private static final Class[] SCHEMA = new Class[] { UUID.class, UUID.class, String.class, Integer.class,
                Map.class };

        public StageletCompleteTable(Runtime context) {
            super(context, TABLE_NAME, PRIMARY_KEY, SCHEMA);
        }

        public static Tuple createTuple(UUID jobId, UUID stageId, String nodeId, int attempt,
                Map<String, Long> statistics) {
            return new Tuple(jobId, stageId, nodeId, attempt, statistics);
        }
    }

    /*
     * declare(stageletfailure, keys(0, 1, 2, 3), {JobId, StageId, NodeId, Attempt})
     */
    private static class StageletFailureTable extends BasicTable {
        private static TableName TABLE_NAME = new TableName(JOL_SCOPE, "stageletfailure");

        private static Key PRIMARY_KEY = new Key(0, 1, 2, 3);

        @SuppressWarnings("unchecked")
        private static final Class[] SCHEMA = new Class[] { UUID.class, UUID.class, String.class, Integer.class };

        public StageletFailureTable(Runtime context) {
            super(context, TABLE_NAME, PRIMARY_KEY, SCHEMA);
        }

        public static Tuple createTuple(UUID jobId, UUID stageId, String nodeId, int attempt) {
            return new Tuple(jobId, stageId, nodeId, attempt);
        }
    }

    /*
     * declare(availablenodes, keys(0), {NodeId})
     */
    private static class AvailableNodesTable extends BasicTable {
        private static TableName TABLE_NAME = new TableName(JOL_SCOPE, "availablenodes");

        private static Key PRIMARY_KEY = new Key(0);

        @SuppressWarnings("unchecked")
        private static final Class[] SCHEMA = new Class[] { String.class };

        public AvailableNodesTable(Runtime context) {
            super(context, TABLE_NAME, PRIMARY_KEY, SCHEMA);
        }

        public static Tuple createTuple(String nodeId) {
            return new Tuple(nodeId);
        }
    }

    /*
     * declare(rankedavailablenodes, keys(0), {NodeId, Integer})
     */
    private static class RankedAvailableNodesTable extends BasicTable {
        private static TableName TABLE_NAME = new TableName(JOL_SCOPE, "rankedavailablenodes");

        private static Key PRIMARY_KEY = new Key(0);

        @SuppressWarnings("unchecked")
        private static final Class[] SCHEMA = new Class[] { String.class, Integer.class };

        public RankedAvailableNodesTable(Runtime context) {
            super(context, TABLE_NAME, PRIMARY_KEY, SCHEMA);
        }

        public static Tuple createTuple(String nodeId, int rank) {
            return new Tuple(nodeId, rank);
        }
    }

    /*
     * declare(failednodes, keys(0), {NodeId})
     */
    private static class FailedNodesTable extends BasicTable {
        private static TableName TABLE_NAME = new TableName(JOL_SCOPE, "failednodes");

        private static Key PRIMARY_KEY = new Key(0);

        @SuppressWarnings("unchecked")
        private static final Class[] SCHEMA = new Class[] { String.class };

        public FailedNodesTable(Runtime context) {
            super(context, TABLE_NAME, PRIMARY_KEY, SCHEMA);
        }

        public static Tuple createTuple(String nodeId) {
            return new Tuple(nodeId);
        }
    }

    /*
     * declare(abortmessage, keys(0, 1), {JobId, StageId, Attempt, JobPlan, TupleSet})
     */
    private static class AbortMessageTable extends BasicTable {
        private static TableName TABLE_NAME = new TableName(JOL_SCOPE, "abortmessage");

        private static Key PRIMARY_KEY = new Key(0, 1, 2);

        @SuppressWarnings("unchecked")
        private static final Class[] SCHEMA = new Class[] { UUID.class, UUID.class, Integer.class, JobPlan.class,
                Set.class };

        public AbortMessageTable(Runtime context) {
            super(context, TABLE_NAME, PRIMARY_KEY, SCHEMA);
        }
    }

    /*
     * declare(abortnotify, keys(0, 1, 2, 3), {JobId, StageId, NodeId, Attempt, StageletStatistics})
     */
    private static class AbortNotifyTable extends BasicTable {
        private static TableName TABLE_NAME = new TableName(JOL_SCOPE, "abortnotify");

        private static Key PRIMARY_KEY = new Key(0, 1, 2, 3);

        @SuppressWarnings("unchecked")
        private static final Class[] SCHEMA = new Class[] { UUID.class, UUID.class, String.class, Integer.class };

        public AbortNotifyTable(Runtime context) {
            super(context, TABLE_NAME, PRIMARY_KEY, SCHEMA);
        }

        public static Tuple createTuple(UUID jobId, UUID stageId, String nodeId, int attempt) {
            return new Tuple(jobId, stageId, nodeId, attempt);
        }
    }

    private static class ExpandPartitionCountConstraintTableFunction extends Function {
        private static final String TABLE_NAME = "expandpartitioncountconstraint";

        @SuppressWarnings("unchecked")
        private static final Class[] SCHEMA = new Class[] { UUID.class, OperatorDescriptorId.class, Integer.class,
                Integer.class };

        public static final int JOBID_FIELD_INDEX = 0;
        public static final int ODID_FIELD_INDEX = 1;
        public static final int NPARTITIONS_FIELD_INDEX = 2;
        public static final int ORDINAL_FIELD_INDEX = 3;

        public ExpandPartitionCountConstraintTableFunction() {
            super(TABLE_NAME, SCHEMA);
        }

        @Override
        public TupleSet insert(TupleSet tuples, TupleSet conflicts) throws UpdateException {
            TupleSet result = new BasicTupleSet();
            int counter = 0;
            for (Tuple t : tuples) {
                int nPartitions = (Integer) t.value(NPARTITIONS_FIELD_INDEX);
                for (int i = 0; i < nPartitions; ++i) {
                    result.add(new Tuple(t.value(JOBID_FIELD_INDEX), t.value(ODID_FIELD_INDEX), i, counter++));
                }
            }
            return result;
        }
    }

    /*
     * declare(profileupdate, keys(0, 1), {JobId, NodeId, Map})
     */
    private static class ProfileUpdateTable extends EventTable {
        private static TableName TABLE_NAME = new TableName(JOL_SCOPE, "profileupdate");

        @SuppressWarnings("unchecked")
        private static final Class[] SCHEMA = new Class[] { UUID.class, String.class, Map.class };

        public ProfileUpdateTable() {
            super(TABLE_NAME, SCHEMA);
        }

        public static Tuple createTuple(UUID jobId, String nodeId, Map<String, Long> statistics) {
            return new Tuple(jobId, nodeId, statistics);
        }
    }

    private class JobQueueThread extends Thread {
        public JobQueueThread() {
            setDaemon(true);
        }

        public void run() {
            Runnable r;
            while (true) {
                try {
                    r = jobQueue.take();
                } catch (InterruptedException e) {
                    continue;
                }
                try {
                    r.run();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }

    public interface ITupleProcessor {
        public void process(Tuple t) throws Exception;
    }

    private class QueryInterfaceImpl implements IJobManagerQueryInterface {
        @Override
        public JSONArray getAllJobSummaries() throws Exception {
            final JSONArray jobs = new JSONArray();
            JOLJobManagerImpl.ITupleProcessor tp = new JOLJobManagerImpl.ITupleProcessor() {
                @Override
                public void process(Tuple t) throws Exception {
                    JSONObject jo = new JSONObject();
                    jo.put("type", "job-summary");
                    jo.put("id", t.value(JobTable.JOBID_FIELD_INDEX).toString());
                    jo.put("status", t.value(JobTable.JOBSTATUS_FIELD_INDEX).toString());
                    jobs.put(jo);
                }
            };
            visitJobs(tp);
            return jobs;
        }

        @Override
        public JSONObject getJobSpecification(UUID jobId) throws Exception {
            final JSONArray jobs = new JSONArray();
            JOLJobManagerImpl.ITupleProcessor tp = new JOLJobManagerImpl.ITupleProcessor() {
                @Override
                public void process(Tuple t) throws Exception {
                    JobSpecification js = (JobSpecification) t.value(JobTable.JOBSPEC_FIELD_INDEX);
                    jobs.put(js.toJSON());
                }
            };
            visitJob(jobId, tp);
            return (JSONObject) (jobs.length() == 0 ? new JSONObject() : jobs.get(0));
        }

        @Override
        public JSONObject getJobPlan(UUID jobId) throws Exception {
            final JSONArray jobs = new JSONArray();
            JOLJobManagerImpl.ITupleProcessor tp = new JOLJobManagerImpl.ITupleProcessor() {
                @Override
                public void process(Tuple t) throws Exception {
                }
            };
            visitJob(jobId, tp);
            return (JSONObject) (jobs.length() == 0 ? new JSONObject() : jobs.get(0));
        }

        @Override
        public JSONObject getJobProfile(UUID jobId) throws Exception {
            final JSONArray jobs = new JSONArray();
            JOLJobManagerImpl.ITupleProcessor tp = new JOLJobManagerImpl.ITupleProcessor() {
                @Override
                public void process(Tuple t) throws Exception {
                    JSONObject jo = new JSONObject();
                    jo.put("type", "profile");
                    jo.put("id", t.value(JobTable.JOBID_FIELD_INDEX).toString());
                    Map<String, Long> profile = (Map<String, Long>) t.value(JobTable.STATISTICS_FIELD_INDEX);
                    if (profile != null) {
                        for (Map.Entry<String, Long> e : profile.entrySet()) {
                            JSONObject jpe = new JSONObject();
                            jpe.put("name", e.getKey());
                            jpe.put("value", e.getValue());
                            jo.accumulate("counters", jpe);
                        }
                    }
                    jobs.put(jo);
                }
            };
            visitJob(jobId, tp);
            return (JSONObject) (jobs.length() == 0 ? new JSONObject() : jobs.get(0));
        }
    }
}