/* SQL provenance output to v8 schema.

Copyright (c) 2008-2010 The Regents of the University of California.
All rights reserved.
Permission is hereby granted, without written agreement and without
license or royalty fees, to use, copy, modify, and distribute this
software and its documentation for any purpose, provided that the above
copyright notice and the following two paragraphs appear in all copies
of this software.

IN NO EVENT SHALL THE UNIVERSITY OF CALIFORNIA BE LIABLE TO ANY PARTY
FOR DIRECT, INDIRECT, SPECIAL, INCIDENTAL, OR CONSEQUENTIAL DAMAGES
ARISING OUT OF THE USE OF THIS SOFTWARE AND ITS DOCUMENTATION, EVEN IF
THE UNIVERSITY OF CALIFORNIA HAS BEEN ADVISED OF THE POSSIBILITY OF
SUCH DAMAGE.

THE UNIVERSITY OF CALIFORNIA SPECIFICALLY DISCLAIMS ANY WARRANTIES,
INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE. THE SOFTWARE
PROVIDED HEREUNDER IS ON AN "AS IS" BASIS, AND THE UNIVERSITY OF
CALIFORNIA HAS NO OBLIGATION TO PROVIDE MAINTENANCE, SUPPORT, UPDATES,
ENHANCEMENTS, OR MODIFICATIONS.

*/

package org.kepler.provenance.sql;

import java.io.BufferedInputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.lang.ref.WeakReference;
import java.math.BigInteger;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.WeakHashMap;
import java.util.jar.JarEntry;
import java.util.regex.Matcher;

import org.kepler.kar.KAREntry;
import org.kepler.kar.KARFile;
import org.kepler.kar.ModuleDependencyUtil;
import org.kepler.moml.NamedObjId;
import org.kepler.moml.NamedObjIdReferralList;
import org.kepler.objectmanager.lsid.KeplerLSID;
import org.kepler.provenance.FireState;
import org.kepler.provenance.IOPortRefillEvent;
import org.kepler.provenance.PortConnector;
import org.kepler.provenance.ProvenanceEvent;
import org.kepler.provenance.QueryException;
import org.kepler.provenance.Queryable;
import org.kepler.provenance.RecordingException;
import org.kepler.provenance.RecordingParameters;
import org.kepler.sms.NamedOntClass;
import org.kepler.sms.SMSServices;
import org.kepler.sms.SemanticType;
import org.kepler.tagging.TagEvent;
import org.kepler.util.FileUtil;
import org.kepler.util.RenameUtil;
import org.kepler.util.WorkflowRenameListener;
import org.kepler.util.WorkflowRun;
import org.kepler.util.sql.Schema;

import ptolemy.actor.Actor;
import ptolemy.actor.Director;
import ptolemy.actor.FiringEvent;
import ptolemy.actor.IOPort;
import ptolemy.actor.IORelation;
import ptolemy.actor.TypedIOPort;
import ptolemy.actor.gui.EditorFactory;
import ptolemy.actor.lib.hoc.IterateOverArray;
import ptolemy.actor.parameters.ParameterPort;
import ptolemy.actor.parameters.PortParameter;
import ptolemy.actor.sched.Scheduler;
import ptolemy.data.ArrayToken;
import ptolemy.data.BooleanToken;
import ptolemy.data.IntToken;
import ptolemy.data.RecordToken;
import ptolemy.data.StringToken;
import ptolemy.data.Token;
import ptolemy.data.expr.Parameter;
import ptolemy.domains.sdf.kernel.SDFDirector;
import ptolemy.domains.sdf.lib.SampleDelay;
import ptolemy.kernel.attributes.VersionAttribute;
import ptolemy.kernel.util.Attribute;
import ptolemy.kernel.util.IllegalActionException;
import ptolemy.kernel.util.NameDuplicationException;
import ptolemy.kernel.util.Nameable;
import ptolemy.kernel.util.NamedObj;
import ptolemy.kernel.util.SingletonAttribute;
import ptolemy.kernel.util.StringAttribute;
import ptolemy.moml.MoMLParser;
import ptolemy.util.MessageHandler;


/** SQL provenance output.
 *
 * <b>NOTE: The v8 schema is under development and likely to change.</b>
 *
 * This uses version 8 of the schema. New features:
 * <ul>
 * <li>Saving the workflow contents for each execution.</li>
 * <li>Saving contents of files referenced in tokens.</li>
 * <li>Each workflow and workflow execution has unique LSID.</li>
 * <li>Workflow name no longer required.</li>
 * <li>Explicit mapping from token write to token read.</li>
 * <li>Combining actor firing events into a single "firing cycle event".</li>
 * <li>Recording execution exceptions.</li>
 * <li>Storing associated data for executions.</li>
 * </ul>
 *
 * @author Derik Barseghian, Daniel Crawl, Ben Leinfelder, Sean Riddle
 * @version $Id: SQLRecordingV8.java 34597 2017-08-10 23:34:49Z crawl $
 *
 */
    
public class SQLRecordingV8 extends SQLRecordingV7
{
    public SQLRecordingV8() throws RecordingException
    {
        super();

        _portConnector = new PortConnector<TokenInfo>();

        _queryable = null;

        // start with empty workflow name
        _wfNameStr = "";

        _nextExecLSIDStr = null;
        
        _executionLSIDtoIdMap = new WeakHashMap<KeplerLSID, Integer>();
        
        _v8Set.add(new WeakReference<SQLRecordingV8>(this));
        
        _executionHadAnError = false;
    }
    
    /** Associate the contents of a file with the most recent workflow execution. */
    @Override
    public void addFileForLastExecution(Map<String,String> metadataMap, File file) 
        throws RecordingException 
    {
        addFileForExecution(metadataMap, file, RegEntity.UNKNOWN_ID);
    }
    
    /** Associate the contents of a file with the most recent workflow execution. 
     *  If execId == RegEntity.UNKNOWN_ID, use last execution.
     */
    public void addFileForExecution(Map<String,String> metadataMap, File file, int execId) 
        throws RecordingException
    {   
        int execIdToUse = execId;
        
        if (execId == RegEntity.UNKNOWN_ID){
                if(_wfLastExecId == RegEntity.UNKNOWN_ID) 
                {
                        throw new RecordingException("No previous workflow execution.");
                }
                execIdToUse = _wfLastExecId;
        }
                
        if(metadataMap.size() == 0)
                {
                        throw new RecordingException("Metadata map is empty.");
                }
        // add the file contents to data table if not already there
        String fileMD5Str = _checkData(file);

        try
        {
            synchronized(_psAssocDataInsert)
            {
                for(Map.Entry<String,String> entry : metadataMap.entrySet())
                {
                    _psAssocDataInsert.setString(1, fileMD5Str); 
                    _psAssocDataInsert.setString(2, entry.getKey());
                    _psAssocDataInsert.setString(3, entry.getValue());
                    _psAssocDataInsert.setInt(4, execIdToUse); 
                    _dbType.insert(_psAssocDataInsert, "associated_data", "id");
        
                    if(_debugWriter != null)
                    {
                        _debugWrite("INSERT INTO ASSOCIATED_DATA (" +
                            fileMD5Str + ", " + entry.getKey() + ", " +
                            entry.getValue() + ", " + execIdToUse + ")");
                    }
                }
            }
        }
        catch(SQLException e)
        {
            throw new RecordingException("Error trying to insert into " +
                "associated_data table: " + e.getMessage());
        }
           
    }

    /** React to a change in an attribute. */
    @Override
    public void attributeChanged(Attribute attribute)
        throws IllegalActionException
    {
        boolean notifyParent = true;

        String name = attribute.getName();

        // see if the workflow name has changed
        if(name.equals(SQLRecordingParameters.wfNameParamStr))
        {
            String val = 
                ((StringToken)((Parameter)attribute).getToken()).stringValue();

            try
            {
                // see if name changed and we're already connected to the db
                if(!val.equals(_wfNameStr) && !_needReconnectWF)
                {
                    // change the wf name in the column
                    _changeWorkflowNameColumn(val);
                    notifyParent = false;
                }
                    
            }
            catch(RecordingException e)
            {
                throw new IllegalActionException(e.getMessage());
            }
        }
        else if(name.equals(SQLRecordingParametersV8.maxFileIncludeSizeKBStr))
        {
            notifyParent = false; 

            Token token = ((Parameter)attribute).getToken();
            if(token == null)
            {
                _maxFileIncludeSizeKBVal = _maxFileIncludeSizeKBDefaultVal;
            }
            else
            {
                _maxFileIncludeSizeKBVal = ((IntToken)token).intValue();
            }    
        }
        else if(name.equals(SQLRecordingParametersV8.execAnnotationStr))
        {
            notifyParent = false;

            _execAnnotation = ((StringToken)((Parameter)attribute).getToken()).stringValue();
        }
        else if(name.equals(SQLRecordingParametersV8.nextExecLSIDStr))
        {
            notifyParent = false;
            
            _nextExecLSIDStr = ((StringToken)((Parameter)attribute).getToken()).stringValue();
        }
        else if(name.equals(SQLRecordingParametersV8.watchForLSIDChangesStr))
        {
            notifyParent = false;
            
            _watchForLSIDChanges = ((BooleanToken)((Parameter)attribute).getToken()).booleanValue();
        }
        
        if(notifyParent)
        {
            super.attributeChanged(attribute);
        }
    }

    /** Add Parameters for ProvenanceRecorder. */
    @Override
    public RecordingParameters generateParameters(NamedObj no)
        throws IllegalActionException, NameDuplicationException
    {
        _params = new SQLRecordingParametersV8(no);
        return _params;
    }

    /** Register a port or portparameter.  */
    @Override
    public boolean regPort(TypedIOPort port) throws RecordingException
    {
        boolean retval = super.regPort(port);

        _portConnector.createConnections(port);

        return retval;
    }

    /** Record the starting of workflow execution. */
    @Override
    public void executionStart(KeplerLSID executionLSID, Date timestamp) throws RecordingException
    {
        //System.out.println("execution start");

        if(_wfExecId != RegEntity.UNKNOWN_ID)
        {
            throw new RecordingException("Workflow already running.");
        }

        // NOTE: do not set this to false so that workflow contents
        // will be registered again. this handles cases where entities
        // are renamed between runs.
        //_needWorkflowContents(false);
        
        if(_wfUserStr == null)
        {
            throw new RecordingException("Need workflow user name.");
        }

        if(_machineStr == null)
        {
            throw new RecordingException("Need machine name.");
        }

        // update the db
        try
        {
            byte[] wfBytes = _recorderContainer.exportMoML().getBytes();
            String wfMD5Str = _checkData(wfBytes);

            String annotationStr;
            if(_execAnnotation != null)
            {
                annotationStr = _execAnnotation;
            }
            else
            {
                annotationStr = "";
            }

            // if _nextExecLSID defined, we should use it instead of runLSID
            if(_nextExecLSIDStr != null && _nextExecLSIDStr.length() != 0)
            {
                try
                {
                        executionLSID = new KeplerLSID(_nextExecLSIDStr);
                }
                catch(Exception e)
                {
                    throw new RecordingException("Error converting given LSID " +
                            _nextExecLSIDStr + " to LSID: ", e);
                }
                
                // clear it so that we don't use it again.
                _nextExecLSIDStr = null;
            }

            // get the workflow lsid
            KeplerLSID wfLSID = NamedObjId.getIdFor(_recorderContainer.toplevel());

            synchronized(_psWorkflowExecStart)
            {
                _psWorkflowExecStart.setInt(1, _wfId);
                _psWorkflowExecStart.setString(2, _wfUserStr);
                _psWorkflowExecStart.setTimestamp(3, new Timestamp(timestamp.getTime()));
                _psWorkflowExecStart.setString(4, wfMD5Str);
                _psWorkflowExecStart.setString(5, _machineStr);
                _psWorkflowExecStart.setString(6, annotationStr);
                _psWorkflowExecStart.setString(7, executionLSID.toString());
                _psWorkflowExecStart.setString(8, wfLSID.toString());
                        _psWorkflowExecStart.setString(9, ModuleDependencyUtil.
                                        buildModuleDependenciesString());
                _psWorkflowExecStart.setString(10, WorkflowRun.type.Running.toString());
                _wfExecId = _dbType.insert(_psWorkflowExecStart,
                    "workflow_exec", "id");
            }
            
            _executionLSIDtoIdMap.put(executionLSID, _wfExecId);
            
            //insert any workflow tags here
                insertWorkflowTags(_wfExecId);

            if(_debugWriter != null)
            {
                _debugWrite("INSERT INTO WORKFLOW_EXEC (" + _wfId +
                    ", " + _wfUserStr + ", curTime, md5, " + _machineStr +
                    ", " + annotationStr + ", lsid)");
            }

            _updateParameterExecTable();
            
        }
        catch(SQLException e)
        {
            throw new RecordingException(_getExceptionMessage(e));
        }
    }

    /** An actor threw an exception.  */
    @Override
    public void executionError(Nameable source, Throwable throwable, KeplerLSID executionLSID)
        throws RecordingException
    {

        int entityId = RegEntity.UNKNOWN_ID;
        if(source != null)
        {
            RegEntity entity = _entityCacheTable.get(source);
            
            // XXX what if null?
            if(entity != null)
            {
                entityId = entity.getId();
            }
        }

        Integer execId = _executionLSIDtoIdMap.get(executionLSID);
        if(execId == null)
        {
            throw new RecordingException("No id found for execution " + executionLSID);
        }

        String message = throwable.getMessage();
        int maxLength = 
            _schema.getTable("error").getColumn("message").getLength();

        // if there is a message and it is larger than the column,
        // truncate it.
        if(message != null && message.length() > maxLength)
        {
            message = message.substring(0, maxLength);
        }

        try
        {
            synchronized(_psErrorInsert)
            {
                _psErrorInsert.setInt(1, entityId);
                _psErrorInsert.setInt(2, execId);
                _psErrorInsert.setString(3, message);
                _dbType.insert(_psErrorInsert, "error", "id");
                    
                if(_debugWriter != null)
                {
                    _debugWrite("INSERT INTO ERROR(" + entityId + ", " +
                        execId + ", " + throwable.getMessage() + ")");
                }
            }
        }
        catch(SQLException e)
        {
            throw new RecordingException("Error inserting into error table: ",
                e);
        }
        changeExecutionType(executionLSID, WorkflowRun.type.Error);
        _executionHadAnError = true;
    }

    /** Record the stopping of workflow execution. */
    @Override
    public void executionStop(KeplerLSID executionLSID, Date timestamp) throws RecordingException
    {
        // save the execution id.
        _wfLastExecId = _wfExecId;
        if (!_executionHadAnError)
        {
            changeExecutionType(executionLSID, WorkflowRun.type.Complete);
        }

        super.executionStop(timestamp);
    }

    /** Returns true if workflow contents should be registered. */
    @Override
    public boolean regContents() throws RecordingException
    {
        // see if we're watching for workflow LSID changes and
        // the workflow LSID is not null
        if(_watchForLSIDChanges && _wfLSID != null)
        {
            // see if the workflow LSID has changed.
            KeplerLSID lsid = NamedObjId.getIdFor(_recorderContainer);
            if(! lsid.equalsWithoutRevision(_wfLSID))
            {
                //_debug("regContents: lsid appears to have changed.");
                //_debug("  " + _wfLSID + " --> " + lsid);
                
                // reconnect to the workflow so that we get the new LSID
                _wfLSID = null;
                _reconnectWorkflow();
                
                // return true so the workflow contents get registered.
                return true;
            }
        }

        return super.regContents();
    }
    
    /** Register a link between two endpoints.  */
    @Override
    public boolean regLink(NamedObj endPoint1, NamedObj endPoint2)
        throws RecordingException
    {
        // do nothing
        return false;
    }
    
    /** Register a parameter. A parameter can be any <b>entity</b>
     * stored in the MoML that does not have its own
     * <code>regNNN()</code> method. This can be user-level 
     * parameters (e.g., Parameter, StringParameter, etc.) or
     * internal to Kepler (e.g., _location, semanticType000, etc.).
     * (A "parameter" corresponds to a property in the MoML).
     *
     */
    @Override
    public boolean regParameter(NamedObj parameter) throws RecordingException
    {
        boolean skip = false;
        
        String name = parameter.getName();
        NamedObj container = parameter.getContainer();
        
        // filter out certain types of parameters
        if(parameter instanceof EditorFactory ||
            parameter instanceof Scheduler ||
            parameter instanceof SemanticType ||
            parameter instanceof SingletonAttribute ||
            parameter instanceof VersionAttribute)
        {
            skip = true;
        }
        // do not record class parameters since the
        // type is already recorded and the class lsid is not used.
        else if(name.equals("class") &&
                parameter instanceof StringAttribute)
        {        
            String val = ((StringAttribute)parameter).getExpression();
            String containerClass =
                parameter.getContainer().getClass().getName();
            if(containerClass.equals(val))
            {
                skip = true;
            }
        }
        // do not record special attributes for ports
        else if((container instanceof IOPort) &&
            (name.equals("tokenConsumptionRate") ||
            name.equals("tokenProductionRate") ||
            name.equals("_showName") ||
            name.equals("_hide")))
        {
            skip = true;
        }
        // do not record internal parameter iterationCount in IterateOverArray
        else if((container instanceof IterateOverArray) &&
            name.equals("iterationCount")) {
                skip = true;
        }
        // do not record AUTO and UNBOUNDED in SDF
        else if((container instanceof SDFDirector) &&
                        (name.equals(SDFDirector.AUTO_NAME) ||
                        name.equals(SDFDirector.UNBOUNDED_NAME))) {
                skip = true;
        }
        // do not record start and stop time in Director
        else if((container instanceof Director) &&
                        (name.equals("startTime") ||
                        name.equals("stopTime"))) {
                skip = true;
        }
        
    
        if(skip)
        {
            return false;
        }
        else
        {
            return super.regParameter(parameter);
        }
    }

    /** Register a relation. */
    @Override
    public boolean regRelation(IORelation relation) throws RecordingException
    {
        // do nothing.
        return false;
    }

    /** Record a port refill event. */
    @Override
    public void refillPortEvent(IOPortRefillEvent event) throws RecordingException
    {
        //System.out.println("got refill event: " + event);
        _portConnector.refillId(event.getPort(), event.getChannel());
    }
    
    /** Get a Queryable connected to the Recording output.
     * @param allowReconnectWF - false if you want to force no _reconnectWF
     * @exception QueryException may be thrown if Queryable not implemented
     *  for the Recording type.
     * @throws RecordingException 
     */
    @Override
    public Queryable getQueryable(boolean allowReconnectWF) throws QueryException, RecordingException
    {
        // make sure we're connected to db
        _checkConnection(allowReconnectWF);
        
        return _queryable;
    }
    
    /** A workflow was renamed.
     * 
     * @param namedObj the workflow
     * @param oldLSID the previous LSID
     * @param newLSID the new LSID
     * @param oldName the previous name
     * @param newName the new name
     * @throws RecordingException 
     * @see WorkflowRenameListener
     */
    @Override
    public void renamedWorkflow(NamedObj namedObj, KeplerLSID oldLSID,
        KeplerLSID newLSID, String oldName, String newName) throws RecordingException
    {
        
        _checkConnection(true);
                
        // make sure we're connected to the database
        if(!_needReconnectDB && !_needReconnectWF)
        {
            Matcher matcher = RenameUtil.unnamedIdPattern.matcher(oldName);
            
            // make sure that the id matches the unnamed regex.
            if(oldName != null)
            {

                if (newName != null)
                {
                        if (matcher.matches())
                        {
                                try 
                                {
                                        _changeWorkflowNameColumn(newName, oldLSID);
                                }
                                catch (RecordingException e)
                                {
                                        MessageHandler.error("Error changing workflow name.", e);
                                }
                        }
                        else
                        {
                                // just change this workflow name
                                _wfNameStr = newName;
                                
                                                // note:
                                                // we call _changeWorkflowNameColumn, but with newLSID,
                                                // because workflow has already been recorded at this point. 
                                // An alternative might be to check for NamedObjIdChangeRequest
                                                // in provenanceRecorder changeExecuted and somehow do
                                                // the rename then, before _recordWorkflowContents.
                                try 
                                {
                                        _changeWorkflowNameColumn(newName, newLSID);
                                }
                                catch (RecordingException e)
                                {
                                        MessageHandler.error("Error changing workflow name.", e);
                                }
                        }
                }
            }
        }
    }
    
    /** Delete executions given a list of execution LSIDs.
     * 
     * @param lsidList
     * @return
     * @throws RecordingException
     */
    public int deleteExecutions(List<KeplerLSID> lsidList) throws RecordingException 
    {

        // make sure we're connected to db
        _checkConnection(false);

        int numRowsDeleted = 0;

        String wfExecStr = _dbType.getTableName("workflow_exec");
        String dataStr = _dbType.getTableName("data");

        StringBuilder lsids = new StringBuilder();
        for (Iterator<KeplerLSID> i=lsidList.iterator(); i.hasNext(); ) 
        {
                lsids.append("'"+i.next().toString()+"'");
                if(i.hasNext())
                {
                        lsids.append(",");
                }
        }
        
                // delete the executions
        String deleteExecutions = "DELETE FROM " + wfExecStr + " wfe " +
                "WHERE wfe.lsid IN (" +lsids+ ")";
        try 
        {               
                        _psDeleteExecutions = _dbType.getPrepStatement(deleteExecutions);
                numRowsDeleted = _psDeleteExecutions.executeUpdate();
                } 
        catch (SQLException e) {
            throw new RecordingException("Error deleting executions:", e);
                }
                
        // determine which rows of data table will become cruft
        // after executions deletion occurs
        String md5sToDelete = getDataMD5sToDelete(lsids);

        // see if there are any rows in the data table to delete
        if(md5sToDelete.length() > 0)
        {
                    // now cleanup data table
            String deleteData = "DELETE FROM " + dataStr + " d " +
                            "WHERE d.md5 IN (" +md5sToDelete+ ")";
            try 
            {                   
                    _psDeleteData = _dbType.getPrepStatement(deleteData);
                    _psDeleteData.executeUpdate();
            } 
            catch (SQLException e) {
                    throw new RecordingException("Error deleting data:", e);
            }           
        }
        
        // tags table now automatically taken care of by cascade 
        // now that it uses WF_EXEC_ID fk
        deleteWorkflowsWithNoExecutions();

        return numRowsDeleted;
    }
    
    /** Delete workflows from the database that have no executions.
     *  @return the number of workflows deleted.
     */
    public int deleteWorkflowsWithNoExecutions() throws RecordingException
    {
        _checkConnection(false);
        

        // see which workflows we're going to delete
        List<Integer> wfIdsToBeDeleted = new LinkedList<Integer>();
        synchronized(_psWorkflowIdsForNoExecutions)
        {
            try
            {
                ResultSet result = null;
                try
                {
                    result = _psWorkflowIdsForNoExecutions.executeQuery();
                    while(result.next())
                    {
                        wfIdsToBeDeleted.add(result.getInt(1));
                    }
                }
                finally
                {
                    if(result != null)
                    {
                        result.close();
                    }
                }
            }
            catch(SQLException e)
            {
                throw new RecordingException("Error querying workflows to " + 
                    "be deleted: ", e);
            }
        }
        
        // see if any other recordings have this workflow open
        // if so, tell them to reconnect to the workflow.
        for(Integer wfId : wfIdsToBeDeleted)
        {
            synchronized(_v8Set)
            {
                // use an iterator so we can safely remove items.
                Iterator<WeakReference<SQLRecordingV8>> iterator = _v8Set.iterator();
                while(iterator.hasNext())
                {
                    SQLRecordingV8 recording = iterator.next().get();
                    
                    // see if recording was garbage collected.
                    if(recording == null)
                    {
                        iterator.remove();
                    }
                    else if(recording._wfId == wfId)
                    {
                        recording._needReconnectWF = true;
                    }
                }
            }
        }
        
        // delete the workflows.
        // NOTE: this deletes all information about a workflow
        // if it has no executions.
        synchronized(_psDeleteWorkflowsForNoExecutions)
        {
            try
            {                
                int numDeleted =
                    _psDeleteWorkflowsForNoExecutions.executeUpdate();
                //System.out.println("deleted " + numDeleted + " workflows.");
                return numDeleted;
            }
            catch(SQLException e)
            {
                throw new RecordingException("Unable to delete workflows " +
                    "with no runs: ", e);
            }
        }
    }

    /** A tag was added, insert it */
    @Override
    public void tagAdded(TagEvent event) throws RecordingException
    {
        // make sure we're connected to db
        _checkConnection(false);
        
                String urnStr = event.getTag().getConceptId();
        String tagStr = event.getTag().toString();

        // get the entity id of the tagged object.
        NamedObj namedObj = event.getSource();
        WorkflowRun run = null;
        
        if (namedObj instanceof WorkflowRun)
        {
                run = (WorkflowRun)namedObj;
                boolean alreadyInserted = checkIfTagAlreadyInserted(run.getExecId(), urnStr);

                if(!alreadyInserted)
                {
                        // TODO if we remove the caveat for not updateRevision on workflowRuns in
                        // Tagging.tagAdded, do not rollRevision here:
                        run.rollRevision(run.getExecLSID(), this);
                        insertTag(WorkflowRun.TAG_TYPE_RUN, tagStr, urnStr, run.getExecId());
                }

        }

    }
    
        /**
         * Parse workflow moml for tags, and insert them, associated with execID 
         * @param runLsid
         */
    public void insertWorkflowTags(int execId)
    {
                try 
                {
                    NamedObj workflow = _recorderContainer.toplevel();
                        Iterator<?> iter = SMSServices.getActorSemanticTypes(workflow).iterator();
                        while (iter.hasNext()) 
                        {
                                SemanticType semtype = (SemanticType) iter.next();
                                insertTag(WorkflowRun.TAG_TYPE_WORKFLOW, semtype.getConceptName(),
                                        semtype.getConceptId(), execId);
                        }
                } 
                catch (Exception e) 
                {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                }
    }
    
    /** Insert a new row into the tag table. */
    public void insertTag(String type, String tagStr, String urnStr,
        int execId) throws RecordingException
    {
                synchronized(_psTagInsert)
                {
                        try
                        {
                        _psTagInsert.setString(1, type);
                                _psTagInsert.setString(2, tagStr);
                                _psTagInsert.setString(3, urnStr);
                                _psTagInsert.setInt(4, execId);                         
                                _psTagInsert.execute();
                                
                if(_debugWriter != null)
                {
                    _debugWrite("INSERT INTO TAG(" + type + ", " + tagStr +
                        ", " + urnStr + ", "+ execId +")");
                }

                        }
                        catch(SQLException e)
                        {
                                throw new RecordingException("Could not insert tag:", e);
                        }
                }
    }
    
    
    public boolean checkIfTagAlreadyInserted(int execId, String urn)
    {
        boolean found = false;
                // see if tag is in table        
        synchronized(_psGetTagForExecIdAndURN)
        {
                try
                {
                        ResultSet result = null;
                        try
                        {
                                _psGetTagForExecIdAndURN.setInt(1, execId);
                                _psGetTagForExecIdAndURN.setString(2, urn);
                                result = _psGetTagForExecIdAndURN.executeQuery();
                                if(result.next())
                                {
                                        found = true;
                                }
                        }
                        finally
                        {
                                if(result != null)
                                {
                                        result.close();
                                }
                        }
                }
                catch(SQLException e)
                {
                        e.printStackTrace();
                }
        }
                return found;
    }
    
    
    /** Record a custom provenance event. */
    @Override
    public void customProvEvent(ProvenanceEvent event) throws RecordingException
    {
        if(_wfExecId == RegEntity.UNKNOWN_ID) 
        {
            throw new RecordingException("Workflow must be running to save associated data.");
        }

        final Map<String,String> map = event.getMap();
        
        if(map.isEmpty())
        {
            throw new RecordingException("Metadata map is empty.");
        }

        try
        {
            synchronized(_psAssocDataInsertNoDataId)
            {
                for(Map.Entry<String,String> entry : map.entrySet())
                {
                    _psAssocDataInsertNoDataId.setString(1, entry.getKey());
                    _psAssocDataInsertNoDataId.setString(2, entry.getValue());
                    _psAssocDataInsertNoDataId.setInt(3, _wfExecId); 
                    _dbType.insert(_psAssocDataInsertNoDataId, "associated_data", "id");
        
                    if(_debugWriter != null)
                    {
                        _debugWrite("INSERT INTO ASSOCIATED_DATA (" +
                            entry.getKey() + ", " + 
                            entry.getValue() + ", " +
                            _wfExecId + ")");
                    }
                }
            }
        }
        catch(SQLException e)
        {
            throw new RecordingException("Error trying to insert into " +
                "associated_data table: " + e.getMessage());
        }
    }

    /** A tag was removed. */
    @Override
    public void tagRemoved(TagEvent event) throws RecordingException
    {
        // make sure we're connected to db
        _checkConnection(false);

        String urnStr = event.getTag().getConceptId();
        NamedObj namedObj = event.getSource();
        
        // just dealing with WorkflowRuns here. We don't want to
        // allow the removal of a tag on a workflow that's already run
        //RegEntity entity = _entityCacheTable.get(namedObj);
                
        if (namedObj instanceof WorkflowRun)
        {
                WorkflowRun run = (WorkflowRun)namedObj;
            synchronized(_psDeleteTagForExecIdAndURN)
            {
                try
                {
                        boolean alreadyInserted = checkIfTagAlreadyInserted(run.getExecId(), urnStr);
                        if (alreadyInserted){
                                _psDeleteTagForExecIdAndURN.setInt(1, run.getExecId());
                                _psDeleteTagForExecIdAndURN.setString(2, urnStr);
                                _psDeleteTagForExecIdAndURN.execute();
                                run.rollRevision(run.getExecLSID(), this);
                        }
                }
                catch(SQLException e)
                {
                    throw new RecordingException("Error deleting tag: ", e);
                }
            }
        }
    }
    
    /** If run's exec LSID does not already exist in provenance, insert WorkflowRun
     *  into provenance. This is a special case for 'preview' runs, where no other
     *  associated data (workflow data, reporting files) are inserted.
     *  
     * @param karFile
     * @param run
     * @throws RecordingException
     */
    public boolean insertPreviewRun(WorkflowRun run) throws RecordingException
    {
        
        Queryable q = null;
        try 
        {
            q = getQueryable(false); //this will make sure we're connected to db
        } 
        catch (QueryException e1) 
        {
            // TODO Auto-generated catch block
            e1.printStackTrace();
            return false;
        }

        try 
        {
                // don't insert if some form of this execution already exists in
                // this provenance store.
                
                // don't insert if run lsid is the same as one of the oldest referrals
                // of an execution in this prov store (without rev).
                KeplerLSID runLSID = run.getExecLSID();
                Integer execId = q.getExecutionForOldestReferralExecutionLSIDWithoutRevision(
                                runLSID.toStringWithoutRevision());
                if (execId != null)
                {
                        _warn("SQLRecordingV8 insertHuskRun - Not inserting run! run:"+execId+
                                        " has the same originalExecLSID without rev:"+
                                        runLSID.toStringWithoutRevision());
                        return false;
                }
                
                // don't insert if one of the oldest referrals of an execution in this
                // prov store (without rev) is the same as the oldest referral of this run
                // (without rev)
                KeplerLSID originalExecLSID = null;
                List<KeplerLSID>refs = run.getReferralList();
                if (refs != null && !refs.isEmpty())
                {
                        originalExecLSID = refs.get(refs.size()-1);
                        execId = q.getExecutionForOldestReferralExecutionLSIDWithoutRevision(
                                        originalExecLSID.toStringWithoutRevision());
                }
                if (execId != null)
                {
                        _warn("SQLRecordingV8 insertPreviewRun - Not inserting run! run:"+execId+
                                        " has the same originalExecLSID without rev:"+
                                        originalExecLSID.toStringWithoutRevision());
                        return false;
                }
                
                // don't insert if the execution originated in this prov store
                // (ie oldest referral (without rev) is in this prov store's execution list)
                if (originalExecLSID != null)
                {
                        execId = q.getExecutionForExecutionLSIDWithoutRevision(
                                originalExecLSID.toStringWithoutRevision());
                        if (execId != null)
                        {
                                _warn("SQLRecordingV8 insertPreviewRun - Not inserting run! " +
                                                "Run originated on this machine and still exists - run:"+
                                                execId+"'s lsid is the same as this run's originalExecLSID " +
                                                "without rev:"+originalExecLSID.toStringWithoutRevision());
                                return false;
                        }
                }
                
                // don't insert if the execution (without rev) is in this 
                // prov store's execution list
                execId = q.getExecutionForExecutionLSIDWithoutRevision(
                                run.getExecLSID().toStringWithoutRevision());
                if (execId != null)
                {
                        _warn("SQLRecordingV8 insertPreviewRun - Not inserting run! " +
                                        "run:"+ execId+"'s lsid is same as this run's lsid without rev:"+
                                        run.getExecLSID().toStringWithoutRevision());
                        return false;
                }
                
                
                Integer workflowExecId = null;
                        
                // only insert workflow if not already in provenance
                int workflowId = _getWorkflowId(run.getWorkflowLSID().toStringWithoutRevision());
                if (workflowId == RegEntity.UNKNOWN_ID)
                {
                        synchronized(_psWorkflowInsert)
                        {
                                _psWorkflowInsert.setString(1, run.getWorkflowName());
                                _psWorkflowInsert.setString(2, run.getWorkflowLSID().toStringWithoutRevision());
                                workflowId = _dbType.insert(_psWorkflowInsert, "workflow", "id");
                        }
                }

                // now insert workflow exec start info
                synchronized(_psWorkflowExecStart)
                {
                        // NOTE we do not have the workflow for preview run,
                        // so we use fake workflow data for the Data table.
                        byte[] fakeWfBytes = "".getBytes();
                        String wfMD5Str = _checkData(fakeWfBytes);

                        _psWorkflowExecStart.setInt(1, workflowId);
                        _psWorkflowExecStart.setString(2, run.getUser());
                        _psWorkflowExecStart.setTimestamp(3,
                                        new Timestamp(run.getStartTime().getTime()));
                        _psWorkflowExecStart.setString(4, wfMD5Str);
                        _psWorkflowExecStart.setString(5, run.getHostId());
                        _psWorkflowExecStart.setString(6, run.getAnnotation());
                        _psWorkflowExecStart.setString(7, run.getExecLSID().toString());
                        _psWorkflowExecStart.setString(8, run.getWorkflowLSID().toString());
                        _psWorkflowExecStart.setString(9, run.getModuleDependencies());
                        if (run.getErrorMessages().isEmpty()){
                                _psWorkflowExecStart.setString(10, WorkflowRun.type.Preview.toString());
                        }
                        else{
                                _psWorkflowExecStart.setString(10, WorkflowRun.type.Preview_Error.toString());
                        }
                        workflowExecId = _dbType.insert(_psWorkflowExecStart,
                                        "workflow_exec", "id");
                }

                // if workflow exec start info insert worked, now insert exec stop info
                // and reporting items and error messages if they exist
                if (workflowExecId != RegEntity.UNKNOWN_ID)
                {
                        synchronized(_psWorkflowExecStop)
                        {
                                Timestamp endTime = new Timestamp(run.getStartTime().getTime() + 
                                                (run.getDuration()*numMillisecondsInASecond));
                                _psWorkflowExecStop.setTimestamp(1, endTime);
                                _psWorkflowExecStop.setInt(2, workflowExecId);
                                _psWorkflowExecStop.executeUpdate();
                        }
                        
                                KeplerLSID runLSIDBeforeExecIdChange = run.getExecLSID();
                        
                        // replace original execId. this must be before call to insertAnyHuskRunErrorMessages
                        run.resetExecId(workflowExecId, this);

                        //insert any referrals
                        insertAnyHuskRunReferralList(run);

                        insertAnyHuskRunErrorMessages(run);

                        insertAnyHuskRunTags(run);
                }
                else
                {
                        return false;
                }
                
                return true;
        }
        catch (SQLException e) 
        {
                // TODO Auto-generated catch block
                e.printStackTrace();
                return false;
        } 
        catch (Exception e) 
        {
                // TODO Auto-generated catch block
                e.printStackTrace();
                return false;
        }

    }
    
    /** If run's exec LSID does not already exist in provenance, insert WorkflowRun
     *  into provenance. This is a special case where the run may come from a KAR, 
     *  potentially from a different system.
     *  
     * @param karFile
     * @param run
     * @throws RecordingException
     */
    public boolean insertHuskRun(KARFile karFile, WorkflowRun run) throws RecordingException
    {

        Queryable q = null;
        try 
        {
            q = getQueryable(false); //this will make sure we're connected to db
        } 
        catch (QueryException e1) 
        {
            // TODO Auto-generated catch block
            e1.printStackTrace();
            return false;
        }

        try 
        {
                // don't insert if some form of this execution already exists in
                // this provenance store.
                
                // don't insert if run lsid is the same as one of the oldest referrals
                // of an execution in this prov store (without rev).
                KeplerLSID runLSID = run.getExecLSID();
                Integer execId = q.getExecutionForOldestReferralExecutionLSIDWithoutRevision(
                                runLSID.toStringWithoutRevision());
                if (execId != null)
                {
                        _warn("SQLRecordingV8 insertHuskRun - Not inserting run! run:"+execId+
                                        " has the same originalExecLSID without rev:"+
                                        runLSID.toStringWithoutRevision());
                        return false;
                }
                
                // don't insert if one of the oldest referrals of an execution in this
                // prov store (without rev) is the same as the oldest referral of this run
                // (without rev)
                KeplerLSID originalExecLSID = null;
                List<KeplerLSID>refs = run.getReferralList();
                if (refs != null && !refs.isEmpty())
                {
                        originalExecLSID = refs.get(refs.size()-1);
                        execId = q.getExecutionForOldestReferralExecutionLSIDWithoutRevision(
                                        originalExecLSID.toStringWithoutRevision());
                }
                if (execId != null)
                {
                        _warn("SQLRecordingV8 insertHuskRun - Not inserting run! run:"+execId+
                                        " has the same originalExecLSID without rev:"+
                                        originalExecLSID.toStringWithoutRevision());
                        return false;
                }
                
                // don't insert if the execution originated in this prov store
                // (ie oldest referral (without rev) is in this prov store's execution list)
                if (originalExecLSID != null)
                {
                        execId = q.getExecutionForExecutionLSIDWithoutRevision(
                                originalExecLSID.toStringWithoutRevision());
                        if (execId != null)
                        {
                                _warn("SQLRecordingV8 insertHuskRun - Not inserting run! " +
                                                "Run originated on this machine and still exists - run:"+
                                                execId+"'s lsid is the same as this run's originalExecLSID " +
                                                "without rev:"+originalExecLSID.toStringWithoutRevision());
                                return false;
                        }
                }
                
                // don't insesrt if the execution (without rev) is in this 
                // prov store's execution list
                execId = q.getExecutionForExecutionLSIDWithoutRevision(
                                run.getExecLSID().toStringWithoutRevision());
                if (execId != null)
                {
                        _warn("SQLRecordingV8 insertHuskRun - Not inserting run! " +
                                        "run:"+ execId+"'s lsid is same as this run's lsid without rev:"+
                                        run.getExecLSID().toStringWithoutRevision());
                        return false;
                }
                
                
                Integer workflowExecId = null;
                        
                // only insert workflow if not already in provenance
                int workflowId = _getWorkflowId(run.getWorkflowLSID().toStringWithoutRevision());
                if (workflowId == RegEntity.UNKNOWN_ID)
                {
                        synchronized(_psWorkflowInsert)
                        {
                                _psWorkflowInsert.setString(1, run.getWorkflowName());
                                _psWorkflowInsert.setString(2, run.getWorkflowLSID().toStringWithoutRevision());
                                workflowId = _dbType.insert(_psWorkflowInsert, "workflow", "id");
                        }
                }

                // now insert workflow exec start info
                synchronized(_psWorkflowExecStart)
                {
                        //byte[] wfBytes = om.getHighestObjectRevision(run.getWorkflowLSID()).exportMoML().getBytes();
                                List<KAREntry> karEntries = karFile.karEntries();
                                KAREntry workflowEntry = null;
                                for (KAREntry karEntry: karEntries){
                                        if (karEntry.getLSID().equals(run.getWorkflowLSID())){
                                                workflowEntry = karEntry;
                                                break;
                                        }
                                }
                                // FIXME: workflowEntry could be null
                                // FIXME: close workflowStream?
                        InputStream workflowStream = karFile.getInputStream(workflowEntry);
                        String workflowString = FileUtil.convertStreamToString(workflowStream);
                        //TODO would be better to avoid parsing the workflow (but getting it from OM
                        // is not desirable either, it's not always there) -- see r22173.
                        // Parsing can cause bug related to text provenance type -- a second prov 
                        // recorder is created and tries to write to the same text file.
                        // But discussed w/ Dan, this is rare and text type is only used for debugging. 
                        MoMLParser parser = new MoMLParser();
                                NamedObj workflow = parser.parse(workflowString);
                        byte[] wfBytes = workflow.exportMoML().getBytes();
                        
                        String wfMD5Str = _checkData(wfBytes);

                        _psWorkflowExecStart.setInt(1, workflowId);
                        _psWorkflowExecStart.setString(2, run.getUser());
                        _psWorkflowExecStart.setTimestamp(3,
                                        new Timestamp(run.getStartTime().getTime()));
                        _psWorkflowExecStart.setString(4, wfMD5Str);
                        _psWorkflowExecStart.setString(5, run.getHostId());
                        _psWorkflowExecStart.setString(6, run.getAnnotation());
                        _psWorkflowExecStart.setString(7, run.getExecLSID().toString());
                        _psWorkflowExecStart.setString(8, run.getWorkflowLSID().toString());
                        _psWorkflowExecStart.setString(9, run.getModuleDependencies());
                        if (run.getErrorMessages().isEmpty()){
                                _psWorkflowExecStart.setString(10, WorkflowRun.type.Imported.toString());
                        }
                        else{
                                _psWorkflowExecStart.setString(10, WorkflowRun.type.Imported_Error.toString());
                        }
                        workflowExecId = _dbType.insert(_psWorkflowExecStart,
                                        "workflow_exec", "id");
                }

                // if workflow exec start info insert worked, now insert exec stop info
                // and reporting items and error messages if they exist
                if (workflowExecId != RegEntity.UNKNOWN_ID)
                {
                        synchronized(_psWorkflowExecStop)
                        {
                                Timestamp endTime = new Timestamp(run.getStartTime().getTime() + 
                                                (run.getDuration()*numMillisecondsInASecond));
                                _psWorkflowExecStop.setTimestamp(1, endTime);
                                _psWorkflowExecStop.setInt(2, workflowExecId);
                                _psWorkflowExecStop.executeUpdate();
                        }
                        
                                KeplerLSID runLSIDBeforeExecIdChange = run.getExecLSID();
                        
                        // replace original execId. this must be before call to insertAnyHuskRunErrorMessages
                        run.resetExecId(workflowExecId, this);

                        //insert any referrals
                        insertAnyHuskRunReferralList(run);

                        //insert any reporting files that might be in KAR
                        insertAnyHuskRunReportFiles(run, runLSIDBeforeExecIdChange, karFile);

                        insertAnyHuskRunErrorMessages(run);

                        insertAnyHuskRunTags(run);
                }
                else
                {
                        return false;
                }
                
                return true;
        }
        catch (SQLException e) 
        {
                // TODO Auto-generated catch block
                e.printStackTrace();
                return false;
        } 
        catch (Exception e) 
        {
                // TODO Auto-generated catch block
                e.printStackTrace();
                return false;
        }

    }
    
    
    public void insertAnyHuskRunReferralList(WorkflowRun run)
    {
        StringAttribute referralListAttr = (StringAttribute) run.getAttribute(NamedObjIdReferralList.NAME);
        if (referralListAttr != null)
        {
                try 
                {                       
                        _psChangeExecutionReferralList.setString(1, referralListAttr.getExpression());
                        _psChangeExecutionReferralList.setInt(2, run.getExecId());
                        _psChangeExecutionReferralList.execute();
                } 
                catch (Exception e) 
                {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                }
        }
        }
    
    private void insertAnyHuskRunReportFiles(WorkflowRun run, KeplerLSID runLSIDBeforeExecIdChange, KARFile karFile)
    {
        
        try
        {
                // get all entries that are ReportLayoutKAREntryHandler and ReportInstanceKAREntryHandler
                List<KAREntry> entries = karFile.karEntries();
                Iterator<KAREntry> itr = entries.iterator();
                while (itr.hasNext())
                {
                        KAREntry karEntry = itr.next();
                        String handler = karEntry.getHandler();
                                String name = karEntry.getName();
                                String type = karEntry.getType();

                        //FIXME hardcoded strings
                        if (handler.endsWith("ReportLayoutKAREntryHandler") || 
                                        handler.endsWith("ReportInstanceKAREntryHandler"))
                        {
                                if (karEntry.dependsOn(runLSIDBeforeExecIdChange)){
                                        insertHuskRunReportFile(name, type, karFile, run.getExecId());
                                }
                        }
                        
                }
                } 
        catch (Exception e) 
        {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                }
    }
    
    private void insertAnyHuskRunErrorMessages(WorkflowRun run) throws RecordingException
    {
        try
        {
                Map<Integer, String> errorMsgs = run.getErrorMessages();
                
                Iterator<Integer> itr = errorMsgs.keySet().iterator();
                while (itr.hasNext()){
                        String errorMsg = errorMsgs.get(itr.next());
                        synchronized(_psErrorInsert)
                        {
                                //TODO I believe it makes sense to not include the original entity id
                                        // since we may be on a different system. But is UNKNOWN_ID what we want?
                                _psErrorInsert.setInt(1, RegEntity.UNKNOWN_ID); 
                                _psErrorInsert.setInt(2, run.getExecId());
                                _psErrorInsert.setString(3, errorMsg);
                                _dbType.insert(_psErrorInsert, "error", "id");
                    
                                if(_debugWriter != null)
                                {
                                        _debugWrite("INSERT INTO ERROR(" + RegEntity.UNKNOWN_ID + ", " +
                                                        run.getExecId() + ", " + errorMsg + ")");
                                }
                        }
                }
        }
        catch(SQLException e)
        {
            throw new RecordingException("Error inserting into error table: ",e);
        } 
        catch (RecordingException e) 
        {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                }
    }
    
    
    private void insertAnyHuskRunTags(WorkflowRun run) throws RecordingException
    {
        
        Map<NamedOntClass, String> tags = run.getTags();
        for(Map.Entry<NamedOntClass, String> entry : tags.entrySet())
        {
                NamedOntClass tag = entry.getKey();
            String type = entry.getValue();
                tag.getConceptId();
                type = tags.get(tag);
                insertTag(type, tag.toString(), tag.getConceptId(), run.getExecId());
        }
        
    }
    
    
    private void insertHuskRunReportFile(String reportItem, String type, KARFile karFile, int runExecId)
    {
        
                //any reason file should be gotten out of ObjectManager instead of straight from kar?
                //File reportLayoutFile = karFile.getKAREntry(reportLayoutLSID);
                //NamedObj reportLayout = ObjectManager.getInstance().getHighestObjectRevision(reportLayoutLSID);
        
                JarEntry je = karFile.getJarEntry(reportItem);
                Map<String, String>metadataMap = new HashMap<String, String>();
                byte[] buf = new byte[1024];
                int len = 0;
                
        try 
        {       
                File file = File.createTempFile("keplerTemp", je.getName());
                try(InputStream in = new BufferedInputStream(karFile.getInputStream(je));
                        OutputStream out = new FileOutputStream(file.getAbsolutePath());)
                {
                        while ((len = in.read(buf)) > 0)
                        {
                                out.write(buf, 0, len);
                        }
                }
                
                metadataMap.put("type", type);
                addFileForExecution(metadataMap, file, runExecId);
                boolean deleted = file.delete();
                if (!deleted){
                        System.out.println("SQLRecordingV8 WARN - unable to delete file:"+file.toString());
                }
        } 
        catch (IOException e) 
        {
                // TODO Auto-generated catch block
                e.printStackTrace();
        } 
        catch (RecordingException e) 
        {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                }
    }
    

    //helper method determines which rows in data table can be deleted when deleting workflow executions
    private String getDataMD5sToDelete(StringBuilder lsids) throws RecordingException
    {
        
        StringBuilder md5sToDelete = new StringBuilder();
        
        String wfExecStr = _dbType.getTableName("workflow_exec");
        String associatedDataStr = _dbType.getTableName("associated_data");
        String portEventStr = _dbType.getTableName("port_event");
        String actorFireStr = _dbType.getTableName("actor_fire");
        
        // FIXME: declare all PreparedStatements locally,
        // close all PreperatedStatements and ResultSets.
           
        try 
        {
                // gather md5s (wf_contents_id) for workflow_exec
                String wfContentsIdsToDelete = "SELECT DISTINCT wfe.wf_contents_id FROM " + wfExecStr +
                        " wfe WHERE wfe.lsid IN (" +lsids+ ")";
                        _psGetWfContentsIdsToDeleteQuery = _dbType.getPrepStatement(wfContentsIdsToDelete);
                        ResultSet resultwfContents = _psGetWfContentsIdsToDeleteQuery.executeQuery();
                        
                // gather md5s (wf_contents_id) for workflow_exec that are still referred to and can't be deleted
                String wfContentsIdsUsedByOthers = "SELECT DISTINCT wfe.wf_contents_id FROM " + wfExecStr +
                        " wfe WHERE wfe.lsid NOT IN (" +lsids+ ")";
                _psGetWfContentsIdsToNotDeleteQuery = _dbType.getPrepStatement(wfContentsIdsUsedByOthers);
                ResultSet resultwfContents2 = _psGetWfContentsIdsToNotDeleteQuery.executeQuery();
                        
                        // gather md5s (data_id) from associated_data
                String associatedDataDataIdsToDelete = "SELECT ad.data_id FROM "+associatedDataStr+" ad WHERE ad.wf_exec_id IN " +
                                "(SELECT wfe.id FROM "+wfExecStr+ " wfe WHERE wfe.lsid IN (" +lsids+ "))";
                        _psGetAssociatedDataDataIdsToDeleteQuery = _dbType.getPrepStatement(associatedDataDataIdsToDelete);
                        ResultSet resultAssociatedDataDataIds = _psGetAssociatedDataDataIdsToDeleteQuery.executeQuery();
                        
                        // gather md5s (data_id) from associated_data that are still referred to and can't be deleted
                String associatedDataDataIdsToNotDelete = "SELECT ad.data_id FROM "+associatedDataStr+" ad WHERE ad.wf_exec_id IN " +
                                "(SELECT wfe.id FROM "+wfExecStr+ " wfe WHERE wfe.lsid NOT IN (" +lsids+ "))";
                _psGetAssociatedDataDataIdsToNotDeleteQuery = _dbType.getPrepStatement(associatedDataDataIdsToNotDelete);
                ResultSet resultAssociatedDataDataIds2 = _psGetAssociatedDataDataIdsToNotDeleteQuery.executeQuery();
                        
                        // gather md5s (data_id) from port_event
                        String portEventDataIdsToDelete = "SELECT DISTINCT pe.data_id FROM "+portEventStr+" pe WHERE pe.fire_id IN " +
                                        "(SELECT af.id FROM "+actorFireStr+" af WHERE af.wf_exec_id IN " +
                                        "(SELECT wfe.id FROM "+wfExecStr+" wfe WHERE wfe.lsid IN ("+lsids+")))";
                        // written this way, the query is slower.
                        //String portEventDataIdsToDelete2 = "SELECT DISTINCT pe.data_id FROM "+ portEventStr+" pe,"+ actorFireStr+" af,"+
                        //      wfExecStr+" wfe WHERE pe.fire_id = af.id AND af.wf_exec_id = wfe.id AND wfe.lsid IN ("+lsids+")";
                        
                        _psGetPortEventDataIdsToDeleteQuery = _dbType.getPrepStatement(portEventDataIdsToDelete);
                        ResultSet resultPortEventDataIds = _psGetPortEventDataIdsToDeleteQuery.executeQuery();
                        
                        // gather md5s (data_id) from port_event that are still referred to and can't be deleted
                        String portEventDataIdsToNotDelete = "SELECT DISTINCT pe.data_id FROM "+portEventStr+" pe WHERE pe.fire_id IN " +
                                        "(SELECT af.id FROM "+actorFireStr+" af WHERE af.wf_exec_id IN " +
                                                        "(SELECT wfe.id FROM "+wfExecStr+" wfe WHERE wfe.lsid NOT IN ("+lsids+")))";
                        // written this way, the query is slower.
                        //String portEventDataIdsToNotDelete2 = "SELECT DISTINCT pe.data_id FROM "+ portEventStr+" pe,"+ actorFireStr+" af,"+
                        //      wfExecStr+" wfe WHERE pe.fire_id = af.id AND af.wf_exec_id = wfe.id AND wfe.lsid NOT IN ("+lsids+")";
                        
                        _psGetPortEventDataIdsToNotDeleteQuery = _dbType.getPrepStatement(portEventDataIdsToNotDelete);
                        ResultSet resultPortEventDataIds2 = _psGetPortEventDataIdsToNotDeleteQuery.executeQuery();                      
                        
                        List<String> md5s = new ArrayList<String>();
            while(resultwfContents.next())
            {
                md5s.add(resultwfContents.getString("wf_contents_id"));
            }
            resultwfContents.close();
            while (resultwfContents2.next())
            {
                md5s.remove(resultwfContents2.getString("wf_contents_id"));
            }
            resultwfContents2.close();
            while(resultAssociatedDataDataIds.next())
            {
                md5s.add(resultAssociatedDataDataIds.getString("data_id"));
            }
            resultAssociatedDataDataIds.close();
            while (resultAssociatedDataDataIds2.next())
            {
                md5s.remove(resultAssociatedDataDataIds2.getString("data_id"));
            }
            resultAssociatedDataDataIds2.close();
            while(resultPortEventDataIds.next())
            {
                md5s.add(resultPortEventDataIds.getString("data_id"));
            }
            resultPortEventDataIds.close();
            while (resultPortEventDataIds2.next())
            {
                md5s.remove(resultPortEventDataIds2.getString("data_id"));
            }
            resultPortEventDataIds2.close();
            
            Iterator<String> itr = md5s.iterator();
            while (itr.hasNext())
            {
                md5sToDelete.append("'"+itr.next()+"',");
            }
            
            // remove trailing comma
            if (md5sToDelete.length() > 0)
            {
                md5sToDelete.delete(md5sToDelete.length()-1, md5sToDelete.length());
            }
                                
                } 
        catch (SQLException e) 
                {
            throw new RecordingException("Error while determining which rows in data table to delete:", e);
                }

                return md5sToDelete.toString();
    }
    
    ////////////////////////////////////////////////////////////////////////
    //// protected methods                                              ////

    /** Add a new row to the entity table. */
    @Override
    protected RegEntity _addEntity(int containerId, RegEntity.EntityType type,
        String fullName, String displayName, int prevId)
        throws RecordingException, SQLException
    {
        // see if we've recording the workflow change
        if(_evolId == RegEntity.UNKNOWN_ID)
        {
            _addWorkflowChange();
        }

        // see if display name is different
        String rowDisplayName;
                
        String name = fullName.substring(fullName.lastIndexOf(".") + 1);

        if(name.equals(displayName))
        {
            rowDisplayName = "";
        }
        else
        {
            rowDisplayName = displayName;
        }

        String typeStr = type.toString();
        
        //(wf_change_id, container_id, type, name, prev_id, wf_id) " + 
        //System.out.println("evolId = " + _evolId);
        
        _psEntityInsert.setInt(1, _evolId);
        _psEntityInsert.setString(2, typeStr);
        _psEntityInsert.setString(3, fullName);
        _psEntityInsert.setString(4, rowDisplayName);
        _psEntityInsert.setInt(5, prevId);
        _psEntityInsert.setInt(6, _wfId);
        int newId = _dbType.insert(_psEntityInsert, "entity", "id");
        RegEntity retval =  new RegEntity(newId, true, containerId, type);

        if(_debugWriter != null)
        {
            _debugWrite("INSERT INTO ENTITY (" + _evolId + ", " + type +
                ", " + fullName + ", " + rowDisplayName + ", " + prevId +
                ", " + _wfId + ")");
        }

        return retval;
    }

    /** Add a new row to the workflow table. */
    @Override
    protected void _addWorkflow() throws RecordingException
    {
        if(_wfLSID == null)
        {
            throw new RecordingException("No workflow LSID");
        }

        //System.out.println("INSERTING new workflow " + _wfLSID.toString());
        
        try
        {
            synchronized(_psWorkflowInsert)
            {
                _psWorkflowInsert.setString(1, _wfNameStr);
                _psWorkflowInsert.setString(2, _wfLSID.toStringWithoutRevision());
                _wfId = _dbType.insert(_psWorkflowInsert, "workflow", "id");
                _wfReset();
            }

            if(_debugWriter != null)
            {
                _debugWrite("INSERT INTO WORKFLOW(" + _wfNameStr + ", lsid)");
            }
        }
        catch(SQLException e)
        {
            throw new RecordingException("Error adding row to workflow:", e);
        }
    }

    /** Create a new row in the workflow_change table. */
    @Override
    protected void _addWorkflowChange() throws RecordingException
    {
        if(_wfUserStr == null)
        {
            throw new RecordingException("Need workflow user name.");
        }
    
        if(_machineStr == null)
        {
            throw new RecordingException("Need machine name.");
        }

        try
        {
            synchronized(_psWorkflowChangeInsert)
            {
                _psWorkflowChangeInsert.setString(1, _wfUserStr);
                _psWorkflowChangeInsert.setTimestamp(2, 
                    _workflowChangeTimeStack.peek());
                _psWorkflowChangeInsert.setInt(3, _wfId);
                _psWorkflowChangeInsert.setString(4, _machineStr);
                _evolId = _dbType.insert(_psWorkflowChangeInsert,
                    "workflow_change", "id");
                //System.out.println("got new evolId = " + _evolId);

                if(_debugWriter != null)
                {
                    _debugWrite("INSERT INTO WORKFLOW_CHANGE(" + 
                        _wfUserStr + ", wfChangeTime, " + _wfId + ", " +
                        _machineStr + ")");
                }
            }
        }
        catch(SQLException e)
        {
            throw new RecordingException("Error adding to workflow_change: ",
                e);
        }
    }

    /** Modify a NamedObj's full name. We remove the containing composite's
     *  name, including the period.
     */
    @Override
    protected String _changeEntityFullName(String name)
    {
        //_debug("name = " + name + "  | containerFullName  = " + _containerFullName);
        //_debug("  real container name = " + _recorder.getContainer().getFullName());        
        
        // see if the name is our container's 
        if(name.equals(_containerFullName))
        {
            // see if we have a relative name
            if(_containerName == null || _containerName.length() == 0)
            {
                return "";
            }
            else
            {
                return _containerFullName.substring(_containerFullName.indexOf(".", 1));
            }
        }
        // see if it is the Graph Editor's default blank composite.
        else if(name.equals(".") && _containerFullName.equals(
            ".configuration.effigyFactory.Graph Editor.blank"))
        {
            return "";
        }
        else
        {
            String relativeName;
            
            // NOTE: _containerFullName may not be a substring of name.
            // this can happen, e.g., when using the ModelReference actor
            // see http://bugzilla.ecoinformatics.org/show_bug.cgi?id=4359
            //
            // instead of using _containerFullName, remove everything up to
            // the second period. e.g., .foo.bar becomes .bar
            //
                        
            int index = name.indexOf(".", 1);
            
            // if there is no second period, i.e., name is the top level
            // composite, return an empty string.
            if(index == -1)
            {
                relativeName = "";
            }
            else
            {
                relativeName = name.substring(index);
            }
                
            //_debug("name " + name + " --> " + relativeName +
                //" (container name = " + _containerFullName + ")");
            return relativeName;
        }
    }

    /** Check parameters before (re)connecting to database or workflow. */
    /*
    protected void _checkParameters() throws IllegalActionException,
        RecordingException
    {
        if((_wfNameStr == null || _wfNameStr.length() == 0) &&
            (_containerFullName.equals(".") || _containerFullName.equals(
                ".configuration.effigyFactory.Graph Editor.blank")))
        {
            _wfNameStr = TMP_START_NAME_STR + _random.nextInt(); 

            _debug("checkParameters() setting workflow name to: "
                + _wfNameStr);
            _debug("    container is " + _containerFullName + " hash = "
                + _recorder.getContainer().hashCode());
        }

        super._checkParameters();
    }
    */
    
    /** Check validity of workflow name parameter. */
    @Override
    protected void _checkWorkflowName() throws RecordingException
    {
        // do nothing since workflow name not required.
    }

    /** Initialize the prepared statements. */
    @Override
    protected void _createPreparedStatements() throws SQLException
    {
        if(_psWorkflowInsert == null && _schema.containsTable("workflow"))
        {
            _psWorkflowInsert = _dbType.getSQLInsert("workflow", "id",
                "name, lsid", "?, ?");
        }

        if(_psWorkflowQuery == null && _schema.containsTable("workflow"))
        {
            _psWorkflowQuery = _dbType.getSQLSelect("workflow", "id",
                "lsid = ?");
        }
        
        if(_psActorFireStart == null && _schema.containsTable("actor_fire"))
        {
            String defaultTimeStr = _dbType.getDefaultTimeStr();
            _psActorFireStart = _dbType.getSQLInsert("actor_fire", "id",
                "actor_id, wf_exec_id, start_time, end_time", "?, ?, ?, " +
                defaultTimeStr);
        }

        if(_psEntityInsert == null && _schema.containsTable("entity"))
        {
            _psEntityInsert = _dbType.getSQLInsert("entity", "id",
                "wf_change_id, type, name, display, prev_id, " +
                "wf_id", "?, ?, ?, ?, ?, ?");
        }

        if(_psWorkflowChangeInsert == null && 
            _schema.containsTable("workflow_change"))
        {
            _psWorkflowChangeInsert = _dbType.getSQLInsert("workflow_change",
                "id", "user, time, wf_id, host_id", "?, ?, ?, ?");
        }
       
        if(_psWorkflowExecStart == null &&
            _schema.containsTable("workflow_exec"))
        {
            String defaultTimeStr = _dbType.getDefaultTimeStr();
            _psWorkflowExecStart = _dbType.getSQLInsert("workflow_exec", "id",
                "wf_id, user, start_time, wf_contents_id, end_time, host_id, " +
                "annotation, lsid, wf_full_lsid, module_dependencies, type",
                "?, ?, ?, ?, " + defaultTimeStr + ", ?, ?, ?, ?, ?, ?");
        }

        // port.type was removed.
        if(_psPortInsert == null && _schema.containsTable("port"))
        {
            _psPortInsert = _dbType.getSQLInsert("port",
                "id, direction, multiport", "?, ?, ?");
        }
    
        if(_psPortEventInsertWithChecksum == null && _schema.containsTable("port_event"))
        {
            _psPortEventInsertWithChecksum = _dbType.getSQLInsert("port_event", "id",
                //"channel, data_id, fire_id, file_id, port_id, rw_fire_id, " +
                "channel, data_id, fire_id, file_id, port_id, " +
                "time, type, write_event_id",  
                "?, ?, ?, ?, ?, ?, ?, ?");
        }

        if(_psPortEventInsert == null && _schema.containsTable("port_event"))
        {
            _psPortEventInsert = _dbType.getSQLInsert("port_event", "id",
                "channel, data, fire_id, file_id, port_id, " +
                "time, type, write_event_id",  
                "?, ?, ?, ?, ?, ?, ?, ?");
        }

        if(_psDataInsert == null && _schema.containsTable("data"))
        {
            _psDataInsert = _dbType.getSQLInsert("data",
                "contents, md5, truncated", "?, ?, ?");
        }
        
        if(_psDataMD5Query == null && _schema.containsTable("data"))
        {
            _psDataMD5Query = _dbType.getSQLSelect("data", "md5",
                "md5 = ?");
        }
        /*
        //Note. not setting these up currently, see Note1 in SQLQueryV8.
        if(_psGetWfContentsIdsToDeleteQuery == null && _schema.containsTable("workflow_exec"))
        {
                String wfExecStr = _dbType.getTableName("workflow_exec");
                String wfContentsIdsOfToDelete = "SELECT DISTINCT wfe.wf_contents_id FROM " + wfExecStr +
                        " wfe WHERE wfe.lsid IN (";
                int someLargeNumber = 10000;
                for (int i=0; i< someLargeNumber; i++){
                        wfContentsIdsOfToDelete.concat("?,?,?,?,?,?,?,?,?,?");
                }
                wfContentsIdsOfToDelete.concat(")");
                        _psGetWfContentsIdsToDeleteQuery = _dbType.getPrepStatement(wfContentsIdsOfToDelete);
        }
        if(_psGetWfContentsIdsToNotDeleteQuery == null && ...)
        {
        }
        if(_psGetAssociatedDataDataIdsToDeleteQuery == null && ...)
        {
        }
        if(_psGetAssociatedDataDataIdsToNotDeleteQuery == null && ...)
        {
        }
        if(_psGetPortEventDataIdsToDeleteQuery == null && ...)
        {
        }
        if(_psGetPortEventDataIdsToNotDeleteQuery == null && ...)
        {
        }
        */
        
        if(_psChangeWorkflowName == null && _schema.containsTable("workflow"))
        {
            _psChangeWorkflowName = _dbType.getSQLUpdate("workflow",
                "name = ?", "lsid = ?");
        }
        
        if(_psChangeExecutionLSID == null && _schema.containsTable("workflow_exec"))
        {
                _psChangeExecutionLSID = _dbType.getSQLUpdate("workflow_exec",
                "lsid = ?", "id = ?");
        }

        if(_psChangeExecutionReferralList == null &&
                _schema.containsTable("workflow_exec"))
        {
                _psChangeExecutionReferralList = _dbType.getSQLUpdate("workflow_exec",
                "derived_from = ?", "id = ?");
        }
        
        if(_psChangeExecutionType == null &&
                _schema.containsTable("workflow_exec"))
        {
            _psChangeExecutionType = _dbType.getSQLUpdate("workflow_exec",
                "type = ?", "lsid = ?");
        }

        if(_psAssocDataInsert == null &&
            _schema.containsTable("associated_data"))
        {
            _psAssocDataInsert = _dbType.getSQLInsert("associated_data", "id",
                "data_id, name, val, wf_exec_id", "?, ?, ?, ?");
        }

        if(_psAssocDataInsertNoDataId == null &&
                _schema.containsTable("associated_data"))
        {
            _psAssocDataInsertNoDataId = _dbType.getSQLInsert("associated_data", "id",
                "name, val, wf_exec_id", "?, ?, ?");
        }
    
        if(_psErrorInsert == null &&
            _schema.containsTable("error"))
        {
            _psErrorInsert = _dbType.getSQLInsert("error", "id",
                "entity_id, exec_id, message", "?, ?, ?");
        }
        
        // tagging
        
        if(_psGetTagForExecIdAndURN == null && _schema.containsTable("tag"))
        {
                _psGetTagForExecIdAndURN = _dbType.getSQLSelect("tag", "urn", "wf_exec_id = ? and urn = ?");   
        }
        
        if(_psTagInsert == null && _schema.containsTable("tag"))
        {
            _psTagInsert = _dbType.getSQLInsert("tag", "type, searchstring, urn, wf_exec_id",
                "?,?,?,?");
        }
        
        if(_psDeleteTagForExecIdAndURN == null && _schema.containsTable("tag"))
        {
                _psDeleteTagForExecIdAndURN = _dbType.getSQLDelete("tag",
                "wf_exec_id = ? and urn = ?");
        }
        
        //TODO use or create a getSQLxxxx ?
        /*
        if(_psDeleteExecutions == null && 
                        _schema.containsTable("workflow_exec"))
        {
                String wfExecStr = _dbType.getTableName("workflow_exec");
                String queryStr = "DELETE FROM " +
                        wfExecStr + " wfe " +
                        "WHERE wfe.id IN (?)";
                _psDeleteExecutions = _dbType.getPrepStatement(queryStr);
        }
        if(_psDeleteData == null){
        
        }
        */
        
        if(_psParameterExecInsert == null &&
            _schema.containsTable("parameter_exec"))
        {
            _psParameterExecInsert = _dbType.getSQLInsert("parameter_exec",
                "parameter_id, wf_exec_id", "?, ?");
        }
        
        String wfStr = _dbType.getTableName("workflow");
        String wfExecStr = _dbType.getTableName("workflow_exec");
        
        if(_psDeleteWorkflowsForNoExecutions == null)
        {
            String sqlStr = "DELETE FROM " + wfStr + " wf " +
                "WHERE wf.id NOT IN (SELECT wf_id FROM " + wfExecStr + ")"; 
            _psDeleteWorkflowsForNoExecutions = _dbType.getPrepStatement(sqlStr);
        }
        
        if(_psWorkflowIdsForNoExecutions == null)
        {
            String sqlStr = "SELECT wf.id FROM " + wfStr + " wf " +
                "WHERE wf.id NOT IN (SELECT wf_id FROM " + wfExecStr + ")";
            _psWorkflowIdsForNoExecutions = _dbType.getPrepStatement(sqlStr);
        }
        
        // create the remaining prepared statements.
        super._createPreparedStatements();
    }

    /** Create a Schema to reflect the v8 schema. */
    @Override
    protected Schema _createSchema()
    {
        return Schemas.createSchemaV8();
    }

    /** Get the maximum size of token data that can be stored in the 
     *  database.
     */
    @Override
    protected int _getMaxTokenDataSize() throws SQLException
    {
        return _dbType.getColumnSize("data", "contents");
    }

    /** Get the internal workflow id. */
    @Override
    protected int _getWorkflowId() throws RecordingException
    {
        if(_wfLSID == null) 
        {
            _wfLSID = NamedObjId.getIdFor(_recorderContainer);
        }
        
        //System.out.println("WF LSID is " + _wfLSID);

        // use the lsid string as the string id.
        return _getWorkflowId(_wfLSID.toStringWithoutRevision());
    }   

    /** Find the container id of an entity. Always returns
     *  RegEntity.UNKNOWN_ID since the V8 schema does not use the
     *  container id.
     */
    @Override
    protected int _getContainerId(Nameable namedObj) 
        throws RecordingException
    {
        return RegEntity.UNKNOWN_ID;
    }
    
    /** Set our prepared statements to null. */
    @Override
    protected void _nullPreparedStatements()
    {
        super._nullPreparedStatements();

        _psDataInsert = null;
        _psDataMD5Query = null;
        _psPortEventInsert = null;
        _psPortEventInsertWithChecksum = null;
        _psChangeWorkflowName = null;
        _psChangeExecutionLSID = null;
        _psAssocDataInsert = null;
        _psAssocDataInsertNoDataId = null;
        _psErrorInsert = null;
        _psDeleteExecutions = null;
        _psDeleteData = null;
        _psGetWfContentsIdsToDeleteQuery = null;
        _psGetWfContentsIdsToNotDeleteQuery = null;
        _psGetAssociatedDataDataIdsToDeleteQuery = null;
        _psGetPortEventDataIdsToDeleteQuery = null;
        _psGetTagForExecIdAndURN = null;
        _psTagInsert = null;
        _psDeleteTagForExecIdAndURN = null;
        _psDeleteTags = null;
        _psParameterExecInsert = null;
        _psDeleteWorkflowsForNoExecutions = null;
        _psWorkflowIdsForNoExecutions = null;
        _psChangeExecutionType = null;
    }

    /** Reconnect to the database. */
    @Override
    protected void _reconnectDatabase(boolean resetDB) throws RecordingException
    {
        super._reconnectDatabase(resetDB);

        try
        {
            _queryable = new SQLQueryV8(_dbType);
        }
        catch(QueryException e)
        {
            throw new RecordingException("Error creating queryable: ", e);
        }
    }

    /** Re-acquire the current workflow id. */
    @Override
    protected void _reconnectWorkflow() throws RecordingException
    {
        super._reconnectWorkflow();

        //_debug("reconnectWorkflow()");
        //_debug("    wf name : " + _wfNameStr);
        //_debug("    _containerFullName: " + _containerFullName);


        // see if the workflow name is empty and the container has a name
        // (the latter means the workflow has been saved to a file).
        if(_wfNameStr.length() == 0 &&
            _containerFullName.length() > 1 && !_containerFullName.equals(
            ".configuration.effigyFactory.Graph Editor.blank"))
        {
            // change the workflow name to the top level
            // composite's name.
            _changeWorkflowNameColumn(_recorderContainer.getName());
        }       
     }

    /** Change the workflow.name column for the current workflow. */
    protected void _changeWorkflowNameColumn(String newName)
        throws RecordingException
    {
        if(_wfLSID == null)
        {
            throw new RecordingException("No workflow LSID");
        }

        _changeWorkflowNameColumn(newName, _wfLSID);
    }
    
    /** Change the workflow.name column for a specific LSID. */
    protected void _changeWorkflowNameColumn(String newName, KeplerLSID lsid)
        throws RecordingException
    {
        //_debug("    changing wf name from " + _wfNameStr + " to " + newName);

        // change the name in the workflow table
        try
        {
            _psChangeWorkflowName.setString(1, newName);
            _psChangeWorkflowName.setString(2, lsid.toStringWithoutRevision());
            _psChangeWorkflowName.execute();
        }
        catch(SQLException e)
        {
            throw new RecordingException("Unable to change workflow name:", e);
        }

        if(_debugWriter != null)
        {
            _debugWrite("UPDATE workflow SET name = " + newName + " " +
                "WHERE lsid = " + lsid.toStringWithoutRevision());
        }

        _wfNameStr = newName;
    }
        
        /**
         * Change execution LSID for an execution Id. Does not change and returns
         * false for attempts to change to an older or current revision, or if a
         * QueryException, or if no such execution Id.
         * 
         * @param execId
         * @param newExecLSID
         * @return
         * @throws RecordingException
         */
        @Override
    public boolean changeExecutionLSID(int execId, KeplerLSID newExecLSID,
                        Queryable q) throws RecordingException 
        {

                // make sure we're connected to db
                _checkConnection(false);

                KeplerLSID currentLSID;

                try 
                {
                        currentLSID = q.getExecutionLSIDForExecution(execId);
                } 
                catch (QueryException e1) 
                {
                        // TODO Auto-generated catch block
                        e1.printStackTrace();
                        return false;
                }

                if (currentLSID == null) 
                {
                        return false;
                }

                if (currentLSID.equalsWithoutRevision(newExecLSID))
                {
                        // disallow changing to an older exec LSID
                        if (currentLSID.getRevision() >= newExecLSID.getRevision()) 
                        {
                                // _warn("Current executionLSID:"+ currentLSID +
                                        // " >= "+newExecLSID+" not changing.");
                                return false;
                        }
                }
                else
                {
                        // TODO no need to do anything special for lsid 
                        // notLocalToInstance, is there?
                }

                try 
                {

                        _psChangeExecutionLSID.setString(1, newExecLSID.toString());
                        _psChangeExecutionLSID.setInt(2, execId);
                        _psChangeExecutionLSID.execute();

                        if (_debugWriter != null) 
                        {
                                _debugWrite("UPDATE WORKFLOW_EXEC SET lsid = " + newExecLSID
                                                + " " + "WHERE id = lsid");
                        }
                        return true;

                } 
                catch (SQLException e) 
                {
                        throw new RecordingException("Could not change execution lsid:", e);
                }
        }
        
        /**
         * Change the execution type.
         * @param executionLSID
         * @param type
         * @return
         * @throws RecordingException
         */
        public boolean changeExecutionType(KeplerLSID executionLSID, WorkflowRun.type type) throws RecordingException{
                try 
                {
                        _psChangeExecutionType.setString(1, type.toString());
                        _psChangeExecutionType.setString(2, executionLSID.toString());
                        _psChangeExecutionType.execute();
                        
                        if (_debugWriter != null) 
                        {
                                _debugWrite("UPDATE WORKFLOW_EXEC SET type = " + type.toString()
                                                + " " + "WHERE lsid = lsid");
                        }
                        return true;
                        
                } 
                catch (SQLException e) 
                {
                        throw new RecordingException("Could not change execution type:", e);

                }
        }

    /** Add a port to the port table. */
    @Override
    protected void _regPortReal(TypedIOPort port, RegEntity re)
        throws RecordingException
    {
        boolean multi = port.isMultiport();

        int direction;
        if(port.isInput() && port.isOutput())
        {
            direction = PortDirection.InOut.ordinal();
        }
        else if(port.isInput())
        {
            direction = PortDirection.In.ordinal();
        }
        else
        {
            direction = PortDirection.Out.ordinal();
        }

        try
        {
            synchronized(_psPortInsert)
            {
                _psPortInsert.setInt(1, re.getId());
                _psPortInsert.setInt(2, direction);
                _psPortInsert.setBoolean(3, multi);
                _psPortInsert.executeUpdate();
            }

            if(_debugWriter != null)
            {
                _debugWrite("INSERT INTO PORT (id, " + direction +
                    ", " + multi + ")");
            }

        }
        catch(SQLException e)
        {
            throw new RecordingException(_getExceptionMessage(e));
        }
    }


    /** Record a specific type of firing for an actor. */
    @Override
    protected void _recordFiringEvent(Actor actor, 
        FiringEvent.FiringEventType type, Date timestamp) throws RecordingException
    {
        /*
        System.out.println("firing event: " + type +
                " for " + _getNameableFullName(actor) +
                " at " + timestamp);
        */
        
        // see if it's a start or stop.
        FireState<Integer> fs = _fireStateTable.get(actor);

        // make sure the fire state exists.
        if(fs == null)
        {   
            throw new RecordingException(
                "Received actor fire event for unregistered actor: " +
                _getNameableFullName(actor));
        }

        synchronized(fs)
        {
            // get the last type of firing start
            FiringEvent.FiringEventType lastStartType =
                fs.getLastStartFireType();

            // see if current firing is new iteration:
            // NOTE: PN does not report iterate firings so the iteration
            // may begin with prefire if the last type of firing was not
            // iterate.
            if(type == FiringEvent.BEFORE_ITERATE ||
                (type == FiringEvent.BEFORE_PREFIRE &&
                lastStartType != FiringEvent.BEFORE_ITERATE))
            {
                int fireId;

                //_debug("start firing: " + type + " for " + _getNameableFullName(actor));

                try
                {
                    synchronized(_psActorFireStart)
                    {
                        _psActorFireStart.setInt(1, fs.getActorId());
                        _psActorFireStart.setInt(2, _wfExecId);
                        _psActorFireStart.setTimestamp(3, new Timestamp(timestamp.getTime()));
                        fireId = _dbType.insert(_psActorFireStart,
                            "actor_fire", "id");
                    }

                }
                catch(SQLException e)
                {
                    String message = "Unable to record start of actor " +
                        "firing: " + e.getMessage();
                    _debugWrite(message);
                    throw new RecordingException(message);
                }

                /*
                System.out.println("started firing " + fireId + ": " +
                        " for " + _getNameableFullName(actor) +
                        " at " + timestamp);
                 */
                
                if(_debugWriter != null)
                {
                    _debugWrite("INSERT INTO ACTOR_FIRE(" +
                        fs.getActorId() + ", " + _wfExecId + ", " +
                        "curTime)");
                }

                fs.fireStart(type, fireId);
            }
            // see if current firing is end of iteration:
            else if(type == FiringEvent.AFTER_ITERATE ||
                (type == FiringEvent.AFTER_POSTFIRE &&
                lastStartType == FiringEvent.BEFORE_PREFIRE))
            {
                //_debug("end firing: " + type + " for " + _getNameableFullName(actor));
               
                Integer fireIdInteger;
                // NOTE: if the type is a AFTER_POSTFIRE and last start
                // type is BEFORE_PREFIRE, we are running in PN.
                // in this case, tell the fireState to stop firing using
                // AFTER_PREFIRE instead of AFTER_POSTFIRE, since we never
                // told the fireState about the BEFORE_POSTFIRE.
                if(type == FiringEvent.AFTER_POSTFIRE)
                {
                    fireIdInteger = fs.fireStop(FiringEvent.AFTER_PREFIRE);
                }
                else
                {
                    fireIdInteger = fs.fireStop(type);
                }

                int fireId = RegEntity.UNKNOWN_ID;
                if(fireIdInteger != null)
                {
                    fireId = fireIdInteger;
                }
                
                int tries = 2;
                SQLException exception = null;
                while(tries > 0)
                {
                    synchronized(_psActorFireStop)
                    {
                        try
                        {
                            tries--;
                            exception = null;
                            _recordActorFireStop(fireId, actor, timestamp);
                            break;
                        }
                        catch(SQLException e)
                        {
                            exception = e;  
                        }
                    }
                    
                    if(exception != null && tries > 0)
                    {
                        _reconnectDatabase(false);
                    }
                }
                
                if(exception != null)
                {
                    throw new RecordingException("Unable to record " +
                            "end of actor firing: ", exception);                    
                }
            }
            /*
            else
            {
                _debug("ignoring firing: " + type + " for " +
                    _getNameableFullName(actor));
            }
            */
        }
    }
    
    /** Record the end of an actor fire and attempt to serialize the actor's
     *  state.
     */
    protected void _recordActorFireStop(int fireId, Actor actor, Date timestamp) throws SQLException
    {
        
        /*
        System.out.println("stop firing " + fireId + ": " +
                " for " + _getNameableFullName(actor) +
                " at " + timestamp);
         */
        
        synchronized(_psActorFireStop)
        {
            _psActorFireStop.setTimestamp(1, new Timestamp(timestamp.getTime()));
            _psActorFireStop.setInt(2, fireId);
            _psActorFireStop.executeUpdate();
        }

        if(_debugWriter != null)
        {
            try {
                _debugWrite("UPDATE ACTOR_FIRE set end_time " +
                    "= curTime WHERE id = " + fireId + ")");
            } catch (RecordingException e) {
                System.out.println("Error writing to debug writer: " + e.getMessage());
            }
        }
        
        if(_stateSerializer != null)
        {
            byte[] state = _stateSerializer.serializeActor(actor);
            if(state != null)
            {
                /* XXX todo once state table is created
                PreparedStatement _psActorStateInsert = _dbType.getSQLInsert("actor_state", "id, state", "?, ?");
                synchronized(_psActorStateInsert)
                {
                    _psActorStateInsert.setInt(1, fireId);
                    _psActorStateInsert.setBytes(2, state);
                    _psActorStateInsert.executeUpdate();
                }
                */
                
                if(_debugWriter != null)
                {
                    try {
                        _debugWrite("INSERT INTO ACTOR_STATE(" + fireId + ", " + String.valueOf(state) + ")");
                    } catch (RecordingException e) {
                        System.out.println("Error writing to debug writer: " + e.getMessage());
                    }
                }
            }
        }
    }

    /** Record a specific type of firing for an actor. */
    @Override
    protected void _recordPortEvent(IOPort port, int fireId, int rwfireId, 
        boolean isRead, int channel, Token token, IOPort destPort, Date timestamp)
        throws RecordingException
    {
        //if(isRead) _debug("port event read " + _getNameableFullName(port));
        //else _debug("port event write " + _getNameableFullName(port));
         
        RegEntity.EntityType type;
        
        if(port instanceof ParameterPort)
        {
            type = RegEntity.EntityType.PortParameter;
        }
        else
        {
            type = RegEntity.EntityType.Port;
        }
        
        RegEntity re = _checkEntity(port, type);

        if(re == null)
        {
            throw new RecordingException("Port has not been registered: " +
                _getNameableFullName(port));
        }

        // an actor may read or write a port in a non-fire method, e.g.,
        // SampleDelay. in this case, generate fake before and after fire
        // events since the firing id is a foreign key.
        boolean fakeFired = false;
        if(fireId == RegEntity.UNKNOWN_ID)
        {
            fakeFired = true;
            Actor actor = (Actor)port.getContainer();
            Director director = actor.getDirector();
            actorFire(new FiringEvent(director, actor, FiringEvent.BEFORE_ITERATE),
                    timestamp);
            
            // now get the firing id
            FireState<Integer> fs = _fireStateTable.get(actor);
            fireId = fs.getCurFireId();
          
            // XXX print a warning if the actor is not sampledelay
            if(! (actor instanceof SampleDelay))
            {
                _warn("actor port event in non-fire method: " + actor.getClass());
            }
            
        }
        
        // get the port id
        int portId = re.getId();
       
        String dataMD5Str = null;
        String fileMD5Str = "";
        String typeStr = null;
        String tokenValue = null;

        int writeId = -1;
  
        // if this is a read event, retrieve the write even id that
        // produced this token
        if(isRead)
        {
            TokenInfo tokenInfo = _portConnector.getNextId(port, channel);
            writeId = tokenInfo.writeId;
            // NOTE: these are not necessary for reads since they are
            // in the row for the write event.
            //dataMD5Str = tokenInfo.dataMD5;
            //fileMD5Str = tokenInfo.fileMD5;
            //tokenValue = tokenInfo.tokenValue;
            //typeStr = token.getClass().getName();
        }
        else
        {
            typeStr = token.getClass().getName();

            tokenValue = token.toString();
            
            // only checksum the data is larger than port_event.data column
            if(tokenValue.length() > MAX_PORT_EVENT_DATA_LENGTH) {
                dataMD5Str = _checkData(token.toString().getBytes());
            }
            
            if(_maxFileIncludeSizeKBVal > 0)
            {
                fileMD5Str = _checkForFileData(token);    
            }
        }

        // insert the new row in port_event table.
        try
        {
            if(dataMD5Str == null) {
                synchronized(_psPortEventInsert)
                {
                    _psPortEventInsert.setInt(1, channel);
                    _psPortEventInsert.setString(2, tokenValue);
                    _psPortEventInsert.setInt(3, fireId);
                    _psPortEventInsert.setString(4, fileMD5Str);
                    _psPortEventInsert.setInt(5, portId);
                    _psPortEventInsert.setTimestamp(6, new Timestamp(timestamp.getTime()));
                    _psPortEventInsert.setString(7, typeStr);
                    _psPortEventInsert.setInt(8, writeId);
    
                    // save the value we wrote into the write_event_id column
                    // for debug writer.
                    int writeEventId = writeId;
    
                    writeId = _dbType.insert(_psPortEventInsert, "port_event", "id");
    
                    if(_debugWriter != null)
                    {
                        _debugWrite("INSERT INTO PORT_EVENT(" +
                            channel + ", " +
                            tokenValue + ", " +
                            fireId + ", " +
                            fileMD5Str + ", " +
                            portId + ", " +
                            "curTime, " +
                            typeStr + ", " +
                            writeEventId + ")");
                    }
                }

            } else {
                synchronized(_psPortEventInsertWithChecksum)
                {
                    _psPortEventInsertWithChecksum.setInt(1, channel);
                    _psPortEventInsertWithChecksum.setString(2, dataMD5Str);
                    _psPortEventInsertWithChecksum.setInt(3, fireId);
                    _psPortEventInsertWithChecksum.setString(4, fileMD5Str);
                    _psPortEventInsertWithChecksum.setInt(5, portId);
                    _psPortEventInsertWithChecksum.setTimestamp(6, new Timestamp(timestamp.getTime()));
                    _psPortEventInsertWithChecksum.setString(7, typeStr);
                    _psPortEventInsertWithChecksum.setInt(8, writeId);
    
                    // save the value we wrote into the write_event_id column
                    // for debug writer.
                    int writeEventId = writeId;
    
                    writeId = _dbType.insert(_psPortEventInsertWithChecksum, "port_event",
                        "id");
    
                    if(_debugWriter != null)
                    {
                        _debugWrite("INSERT INTO PORT_EVENT(" +
                            channel + ", " +
                            dataMD5Str + ", " +
                            fireId + ", " +
                            fileMD5Str + ", " +
                            portId + ", " +
                            "curTime, " +
                            typeStr + ", " +
                            writeEventId + ")");
                    }
                }   
            }
            
            // if this is a write event, update the queues for each
            // of the receiving ports
            if(!isRead)
            {
                //_debug("updating any queues connected to " + _getNameableFullName(port));
                            
                TokenInfo tokenInfo = new TokenInfo(writeId, dataMD5Str,
                    fileMD5Str, tokenValue);

                if(destPort == null)
                {
                    _portConnector.sendIdToConnections(port, channel,
                        tokenInfo);
                }
                else
                {
                    _portConnector.receiveId(destPort, channel, tokenInfo);
                }
            }
        }
        catch(SQLException e)
        {
            throw new RecordingException(_getExceptionMessage(e));
        }

        if(fakeFired)
        {
            Actor actor = (Actor)port.getContainer();
            Director director = actor.getDirector();
            actorFire(new FiringEvent(director, actor, FiringEvent.AFTER_ITERATE));
        }
    }

    /** Adds data to the data table if not already there.
     *  @return the md5 of the data
     */
    protected String _checkData(byte[] data) throws RecordingException
    {
        boolean truncated = false;

        // see if data is too large
        if(data.length > _maxDataSize)
        {
            truncated = true;
        }

        // compute checksum
        String retval = _getMD5(data);

        boolean found = false;

        try
        {
            // query table for checksum
            synchronized(_psDataMD5Query)
            {
                _psDataMD5Query.setString(1, retval);
                ResultSet result = _psDataMD5Query.executeQuery();
        
                if(result.next())
                {
                    found = true;
                }
        
                result.close();
            }
        }
        catch(SQLException e)
        {
            throw new RecordingException("Unable to query data table: " +
                e.getMessage());
        }

        // if not found insert
        if(!found)
        {
            try
            {
                synchronized(_psDataInsert)
                {
                    _psDataInsert.setBytes(1, data);
                    _psDataInsert.setString(2, retval);
                    _psDataInsert.setBoolean(3, truncated);
                    _psDataInsert.executeUpdate();
                }
                
                if(_debugWriter != null)
                {
                    // both data and md5 can change depending on time, so don't output.
                    _debugWrite("INSERT INTO DATA(" +
                        "data, " +
                        "md5, " +
                        truncated + ")");
                }
            }
            catch(SQLException e)
            {
                throw new RecordingException("Unable to insert into data " +
                    "table: " + e.getMessage());
            }
        }
        return retval;
    }

    /** Adds data to the data table if not already there.
     *  @return the md5 of the data
     */
    protected String _checkData(File file) throws RecordingException
    {
        String retval = "";
        int fileBytes = Long.valueOf(file.length()).intValue();
        FileInputStream stream = null;

        try
        {
            try
            {
                stream = new FileInputStream(file);                    
            }
            catch(FileNotFoundException e)
            {
                throw new RecordingException("File " + file.getAbsolutePath() +
                " not found: ", e);
            }

            byte[] data = new byte[fileBytes];

            try
            {
                if(stream.read(data) != fileBytes)
                {
                    throw new RecordingException("Unable to read entire " +
                        "file: " + file.getAbsolutePath());
                }
            }
            catch(IOException e)
            {
                throw new RecordingException("Error reading file " +
                    file.getAbsolutePath() + ": ", e);
            }
                
            retval = _checkData(data); 
        }
        finally
        {
            if(stream != null)
            {
                try
                {
                    stream.close();
                }
                catch(IOException e)
                {
                    throw new RecordingException("Error closing file " +
                        file.getAbsolutePath() + ": ", e);
                }
            }
        }
        
        return retval;
    }

    /** See if a token contains a valid filename.
     *  @return If token contains a valid filename, and its size is less than
     *  minFileIncludeKB, then returns the MD5 of the file's contents.
     */
    protected String _checkForFileData(Token token) throws RecordingException
    {
        String retval = "";

        ArrayToken arrayToken = null;
        RecordToken recordToken = null;

        if(token instanceof StringToken)
        {
            String possibleFilePath = ((StringToken)token).stringValue();
            File possibleFile = new File(possibleFilePath);

            //_debug("seeing if file: " + possibleFile.getAbsolutePath());

            int fileBytes = Long.valueOf(possibleFile.length()).intValue();
            float fileKiloBytes = 0;
            if(fileBytes > 1023)
            {
                fileKiloBytes = Float.valueOf(fileBytes) / 1024;
            }

            // see if file is a file (not a directory), is readable, 
            // and is less than the max size.
            if(possibleFile.isFile() && possibleFile.canRead() &&
                fileKiloBytes <= _maxFileIncludeSizeKBVal)
            {
                /*
                _debug("    file is readable and under max (size = " +
                    fileKiloBytes + " max = " + _maxFileIncludeSizeKBVal +
                    "); going to include");
                */

                retval = _checkData(possibleFile);
            }
        }
        //if ArrayToken or RecordToken recurse to check elements
        else if (token instanceof ArrayToken){
                arrayToken = ((ArrayToken)token);
                for (int i=0; i<arrayToken.length(); i++){
                        final Token aToken = arrayToken.getElement(i);
                        if (!aToken.isNil()){
                        _checkForFileData(aToken);
                        }
                }
        }
        else if (token instanceof RecordToken){
                recordToken = ((RecordToken)token);
                Iterator<?> recordTokenItr = recordToken.labelSet().iterator();
                while (recordTokenItr.hasNext()){
                        final String label = (String)recordTokenItr.next();
                        final Token aToken = recordToken.get(label);
                        if (!aToken.isNil()){
                                _checkForFileData(aToken);
                        }
                }
        }

        return retval;
    }

    /** Reset when we use a different db connection. */
    @Override
    protected void _dbReset() throws RecordingException
    {
        _wfLSID = _containerLSID;
        super._dbReset();
    }

    /** Reset when we use a different workflow. */
    @Override
    protected void _wfReset()
    {
        _wfLastExecId = RegEntity.UNKNOWN_ID;
        super._wfReset();
    }

    /** Add all the parameter ids to the parameter_exec table. */
    protected void _updateParameterExecTable() throws RecordingException
    {
        // XXX synchronize access to _entityCacheTable
        for(Map.Entry<Nameable,RegEntity> entry : _entityCacheTable.entrySet())
        {
            Nameable nameable = entry.getKey();
            RegEntity entity = entry.getValue();
            RegEntity.EntityType type = entity.getType();
            
            // add to parameter exec table if a parameter or unconnected
            // port parameter. don't add connected port parameters since
            // their values come from the port.
            if(type == RegEntity.EntityType.Parameter ||
                ((nameable instanceof PortParameter) &&
                ((PortParameter)nameable).getPort().numberOfSources() == 0))
            {
                try
                {                    
                    _psParameterExecInsert.setInt(1, entity.getId());
                    _psParameterExecInsert.setInt(2, _wfExecId);
                    
                    _psParameterExecInsert.executeUpdate();
  
                    if(_debugWriter != null)
                    {
                        // both data and md5 can change depending on time, so don't output.
                        /* XXX
                        _debugWrite("INSERT INTO PARAMETER_EXEC(" +
                            entity.getId() + ", " + _wfExecId + ")");
                        */
                    }
                }
                catch(SQLException e)
                {
                    throw new RecordingException("Error updating parameter_exec table: ", e);
                }
            }
        }
    }
    
    ////////////////////////////////////////////////////////////////////////
    //// protected classes                                              ////

    /** The class used in the port queues to contain the port event write id
     *  and data md5.
     */
    protected static class TokenInfo 
    {
        /** Construct a new PortQueueEntry for a write id and md5 of the
         *  data.
         */
        TokenInfo(int writeId, String dataMD5, String fileMD5, String tokenValue)
        {
            this.writeId = writeId;
            this.dataMD5 = dataMD5;
            this.fileMD5 = fileMD5;
            this.tokenValue = tokenValue;
        }
        
        /** Get a string representation. */
        @Override
        public String toString()
        {
            return "writeId = " + writeId; // + " data md5 = " + _dataMD5; 
        }

        /** The MD5 checksum. */
        public String dataMD5;

        /** The file MD5 checksum. */
        public String fileMD5;

        /** The port event write id. */
        public int writeId;
        
        public String tokenValue;
    }

    /** A subclass of SQLRecordingParameters that adds more parameters. */
    protected static class SQLRecordingParametersV8 
        extends SQLRecordingParameters
    {
        /** Construct a new SQLRecordParametersV8. */
        protected SQLRecordingParametersV8(NamedObj no)
            throws IllegalActionException, NameDuplicationException
        {
            super(no);
            addIntParameter(maxFileIncludeSizeKBStr, _maxFileIncludeSizeKBDefaultVal);
            addStringParameter(execAnnotationStr);
            addStringParameter(nextExecLSIDStr);
            addBooleanParameter(watchForLSIDChangesStr, true);
        }

        /** Name of parameter that specifies maximum size, in KB, of file
         *  to include in provenance.
         */
        protected static final String maxFileIncludeSizeKBStr =
            "maxFileInclusionSizeKB";

        /** Name of execution annotation parameter. */
        protected static final String execAnnotationStr = "execAnnotation";
        
        /** Name of next execution LSID parameter. */
        protected static final String nextExecLSIDStr = "nextExecLSID";
        
        /** Name of watch for LSID changes parameter. */
        protected static final String watchForLSIDChangesStr = "watchForLSIDChanges";
    }

    ////////////////////////////////////////////////////////////////////////
    //// protected variables                                            ////

    /** Class for token dependencies. */
    protected PortConnector<TokenInfo> _portConnector;

    // prepared statements
    protected PreparedStatement _psDataInsert;
    protected PreparedStatement _psDataMD5Query;
    protected PreparedStatement _psPortEventInsert;
    protected PreparedStatement _psPortEventInsertWithChecksum;
    protected PreparedStatement _psChangeWorkflowName;
    protected PreparedStatement _psChangeExecutionLSID;
    protected PreparedStatement _psChangeExecutionReferralList;
    protected PreparedStatement _psChangeExecutionType;
    protected PreparedStatement _psAssocDataInsert;
    protected PreparedStatement _psAssocDataInsertNoDataId;
    protected PreparedStatement _psErrorInsert;
    protected PreparedStatement _psDeleteExecutions;
    protected PreparedStatement _psDeleteData;
    protected PreparedStatement _psGetWfContentsIdsToDeleteQuery;
    protected PreparedStatement _psGetWfContentsIdsToNotDeleteQuery;
    protected PreparedStatement _psGetAssociatedDataDataIdsToDeleteQuery;
    protected PreparedStatement _psGetAssociatedDataDataIdsToNotDeleteQuery;
    protected PreparedStatement _psGetPortEventDataIdsToDeleteQuery;
    protected PreparedStatement _psGetPortEventDataIdsToNotDeleteQuery;
    protected PreparedStatement _psGetTagForExecIdAndURN;
    protected PreparedStatement _psTagInsert;
    protected PreparedStatement _psDeleteTagForExecIdAndURN;
    protected PreparedStatement _psDeleteTags;
    protected PreparedStatement _psParameterExecInsert;
    protected PreparedStatement _psDeleteWorkflowsForNoExecutions;
    protected PreparedStatement _psWorkflowIdsForNoExecutions;
    
    /** Queryable connection. */
    protected Queryable _queryable;
    
    /** The most recently completed workflow execution id. */
    protected int _wfLastExecId;

    /** Value of execution annotation parameter. */
    protected String _execAnnotation;

    /** Value of next execution LSID parameter. */
    protected String _nextExecLSIDStr;
        
    /** The current workflow id. */
    protected KeplerLSID _wfLSID;

    /** True if executionError is called */
    protected boolean _executionHadAnError;

    protected Map<KeplerLSID, Integer> _executionLSIDtoIdMap;
    
    private static final int numMillisecondsInASecond = 1000;

    ////////////////////////////////////////////////////////////////////////
    //// private methods                                                ////
        
    /** Compute an MD5 checksum from a byte array. */
    private String _getMD5(byte[] data) throws RecordingException
    {
        String retval = null;
        try
        {
            MessageDigest digest = MessageDigest.getInstance("MD5");
            digest.update(data);
            byte[] md5sum = digest.digest();
         
            //XXX necessary?
            BigInteger bigInt = new BigInteger(1, md5sum);
            retval = bigInt.toString(16);
                    
            // make sure it's 32 chars
            while(retval.length() < 32)
            {
                retval = "0" + retval;
            }
        }
        catch (NoSuchAlgorithmException e)
        {
            throw new RecordingException("MD5 digest does not exist.");
        }
        return retval;    
    }
  
    ////////////////////////////////////////////////////////////////////////
    //// private variables                                              ////

    /** Maximum size, in kilobytes, of files to include in provenance. */
    private int _maxFileIncludeSizeKBVal;

    /** The default value for maximum size of file for inclusion. */
    private final static int _maxFileIncludeSizeKBDefaultVal = 1024;

    /** A synchronized collection of instances of this class. */
    private static Set<WeakReference<SQLRecordingV8>> _v8Set = 
        Collections.synchronizedSet(new HashSet<WeakReference<SQLRecordingV8>>());

    /** If true, when the workflow LSID changes, reconnect to workflow. */
    private boolean _watchForLSIDChanges;
    
    /** The maximum length of the data stored in the port_event table.
     *  Larger data is stored in the data table.
     */
    public final static int MAX_PORT_EVENT_DATA_LENGTH = 4096;
    
}