Metacat

/**

 *  '$RCSfile$'

 *  Copyright: 2010 Regents of the University of California and the

 *             National Center for Ecological Analysis and Synthesis

 *

 *   '$Author: jones $'

 *     '$Date: 2010-02-03 17:58:12 -0900 (Wed, 03 Feb 2010) $'

 * '$Revision: 5211 $'

 *

 * This program is free software; you can redistribute it and/or modify

 * it under the terms of the GNU General Public License as published by

 * the Free Software Foundation; either version 2 of the License, or

 * (at your option) any later version.

 *

 * This program is distributed in the hope that it will be useful,

 * but WITHOUT ANY WARRANTY; without even the implied warranty of

 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

 * GNU General Public License for more details.

 *

 * You should have received a copy of the GNU General Public License

 * along with this program; if not, write to the Free Software

 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA

 */

package edu.ucsb.nceas.metacat;

import java.util.*;

import java.sql.PreparedStatement;

import java.sql.ResultSet;

import java.sql.SQLException;

import java.sql.Timestamp;

import org.apache.log4j.Logger;

import org.dataone.service.types.Checksum;

import org.dataone.service.types.ChecksumAlgorithm;

import org.dataone.service.types.ObjectFormat;

import org.dataone.service.types.ObjectInfo;

import org.dataone.service.types.ObjectList;

import org.dataone.service.types.SystemMetadata;

import org.dataone.service.types.Identifier;

import edu.ucsb.nceas.metacat.database.DBConnection;

import edu.ucsb.nceas.metacat.database.DBConnectionPool;

import edu.ucsb.nceas.metacat.properties.PropertyService;

import edu.ucsb.nceas.metacat.shared.ServiceException;

import edu.ucsb.nceas.metacat.util.DocumentUtil;

import edu.ucsb.nceas.utilities.PropertyNotFoundException;

/**

 * Manage the relationship between Metacat local identifiers (LocalIDs) that are

 * codified as the (docid, rev) pair with globally uniqe string identifiers

 * (GUIDs) that are opaque strings.  This class provides methods to manage these

 * identifiers, and to search for and look up LocalIDs based on their GUID and

 * vice versa. IdentifierManager is a singleton.

 * 

 * @author Matthew Jones

 */

public class IdentifierManager {

    public static final String TYPE_SYSTEM_METADATA = "systemmetadata";

    public static final String TYPE_IDENTIFIER = "identifier";

    public static String DATAONE_SM_DOCTYPE;

    /**

     * The single instance of the manager that is always returned.

     */

    private static IdentifierManager self = null;

    private Logger logMetacat = Logger.getLogger(EventLog.class);

    /**

     * A private constructor that initializes the class when getInstance() is

     * called.

     */

    private IdentifierManager()

    {

        try

        {

            DATAONE_SM_DOCTYPE = PropertyService.getProperty("crudService.listObjects.ReturnDoctype");

        }

        catch(Exception e)

        {

            throw new RuntimeException("Error getting System Metadata doctype from " + 

                    "the properties file.  Please make sure crudService.listObjects.ReturnDoctype " +

                    "exists in metacat.properties.");

        }

    }

    /**

     * Return the single instance of the manager after initializing it if it

     * wasn't previously initialized.

     * 

     * @return the single IdentifierManager instance

     */

    public static IdentifierManager getInstance()

    {

        if (self == null) {

            self = new IdentifierManager();

        }

        return self;

    }

    /**

     * return a hash of all of the info that is in the systemmetadata table

     * @param localId

     * @return

     */

    public Hashtable<String, String> getSystemMetadataInfo(String localId)

    throws McdbDocNotFoundException

    {

        try

        {

            AccessionNumber acc = new AccessionNumber(localId, "NONE");

            localId = acc.getDocid();

        }

        catch(Exception e)

        {

            //do nothing. just try the localId as it is

        }

        Hashtable<String, String> h = new Hashtable<String, String>();

        String sql = "select guid, date_uploaded, rights_holder, checksum, checksum_algorithm, " +

          "origin_member_node, authoritative_member_node, date_modified, submitter " +

          "from systemmetadata where docid = ?";

        DBConnection dbConn = null;

        int serialNumber = -1;

        try 

        {

            // Get a database connection from the pool

            dbConn = DBConnectionPool.getDBConnection("IdentifierManager.getDocumentInfo");

            serialNumber = dbConn.getCheckOutSerialNumber();

            // Execute the insert statement

            PreparedStatement stmt = dbConn.prepareStatement(sql);

            stmt.setString(1, localId);

            ResultSet rs = stmt.executeQuery();

            if (rs.next()) 

            {

                String guid = rs.getString(1);

                Timestamp dateUploaded = rs.getTimestamp(2);

                String rightsHolder = rs.getString(3);

                String checksum = rs.getString(4);

                String checksumAlgorithm = rs.getString(5);

                String originMemberNode = rs.getString(6);

                String authoritativeMemberNode = rs.getString(7);

                Timestamp dateModified = rs.getTimestamp(8);

                String submitter = rs.getString(9);

                h.put("guid", guid);

                h.put("date_uploaded", new Long(dateUploaded.getTime()).toString());

                h.put("rights_holder", rightsHolder);

                h.put("checksum", checksum);

                h.put("checksum_algorithm", checksumAlgorithm);

                h.put("origin_member_node", originMemberNode);

                h.put("authoritative_member_node", authoritativeMemberNode);

                h.put("date_modified", new Long(dateModified.getTime()).toString());

                h.put("submitter", submitter);

                stmt.close();

            } 

            else

            {

                stmt.close();

                DBConnectionPool.returnDBConnection(dbConn, serialNumber);

                throw new McdbDocNotFoundException("2Could not find document " + localId);

            }

        } 

        catch (SQLException e) 

        {

            e.printStackTrace();

            logMetacat.error("Error while getting system metadata info for localid " + localId + " : "  

                    + e.getMessage());

        } 

        finally 

        {

            // Return database connection to the pool

            DBConnectionPool.returnDBConnection(dbConn, serialNumber);

        }

        return h;

    }

    /**

     * return information on the document with localId.  These are the fields

     * from the xml_documents table.  They can be used to contstruct metadata 

     * about the object that is stored.

     * @param localId

     * @return

     * @throws McdbDocNotFoundException

     */

    public Hashtable<String, Object> getDocumentInfo(String localId)

        throws McdbDocNotFoundException

    {

        try

        {

            AccessionNumber acc = new AccessionNumber(localId, "NONE");

            localId = acc.getDocid();

        }

        catch(Exception e)

        {

            //do nothing. just try the localId as it is

        }

        Hashtable<String, Object> h = new Hashtable<String, Object>();

        String sql = "select docname, doctype, user_owner, user_updated, " +

            "server_location, rev, date_created, date_updated from " + 

            "xml_documents where docid like '" + localId + "'";

        DBConnection dbConn = null;

        int serialNumber = -1;

        try 

        {

            // Get a database connection from the pool

            dbConn = DBConnectionPool.getDBConnection("IdentifierManager.getDocumentInfo");

            serialNumber = dbConn.getCheckOutSerialNumber();

            // Execute the insert statement

            PreparedStatement stmt = dbConn.prepareStatement(sql);

            ResultSet rs = stmt.executeQuery();

            if (rs.next()) 

            {

                String docname = rs.getString(1);

                String doctype = rs.getString(2);

                String user_owner = rs.getString(3);

                String user_updated = rs.getString(4);

                String server_location = rs.getString(5);

                int rev = rs.getInt(6);

                String date_created = rs.getString(7);

                String date_updated = rs.getString(8);

                h.put("docname", docname);

                h.put("doctype", doctype);

                h.put("user_owner", user_owner);

                h.put("user_updated", user_updated);

                h.put("server_location", server_location);

                h.put("rev", new Integer(rev).toString());

                h.put("date_created", date_created);

                h.put("date_updated", date_updated);

                stmt.close();

            } 

            else

            {

                stmt.close();

                DBConnectionPool.returnDBConnection(dbConn, serialNumber);

                throw new McdbDocNotFoundException("2Could not find document " + localId);

            }

            String sql2 = "select principal_name, permission, perm_type, perm_order from xml_access " +

            "where docid like '" + localId + "'";

            System.out.println("executing sql: " + sql2);

            PreparedStatement stmt2 = dbConn.prepareStatement(sql2);

            rs = stmt2.executeQuery();

            Vector accessVector = new Vector();

            while(rs.next())

            {

                Hashtable accessHash = new Hashtable();

                String principal_name = rs.getString(1);

                String permission = rs.getString(2);

                String permissionType = rs.getString(3);

                String permissionOrder = rs.getString(4);

                accessHash.put("principal_name", principal_name);

                accessHash.put("permission", permission);

                accessHash.put("permission_type", permissionType);

                accessHash.put("permission_order", permissionOrder);

                System.out.println("accessHash: " + accessHash.toString());

                accessVector.add(accessHash);

            }

            h.put("access", accessVector);

        } 

        catch (SQLException e) 

        {

            e.printStackTrace();

            logMetacat.error("Error while getting document info for localid " + localId + " : "  

                    + e.getMessage());

        } 

        finally 

        {

            // Return database connection to the pool

            DBConnectionPool.returnDBConnection(dbConn, serialNumber);

        }

        return h;

    }

    /**

     * return the newest rev for a given localId

     * @param localId

     * @return

     */

    public int getLatestRevForLocalId(String localId)

        throws McdbDocNotFoundException

    {

        try

        {

            AccessionNumber acc = new AccessionNumber(localId, "NONE");

            localId = acc.getDocid();

        }

        catch(Exception e)

        {

            //do nothing. just try the localId as it is

        }

        int rev = 0;

        String sql = "select rev from xml_documents where docid like '" + localId + "'";

        DBConnection dbConn = null;

        int serialNumber = -1;

        try 

        {

            // Get a database connection from the pool

            dbConn = DBConnectionPool.getDBConnection("IdentifierManager.getLatestRevForLocalId");

            serialNumber = dbConn.getCheckOutSerialNumber();

            // Execute the insert statement

            PreparedStatement stmt = dbConn.prepareStatement(sql);

            ResultSet rs = stmt.executeQuery();

            if (rs.next()) 

            {

                rev = rs.getInt(1);

                stmt.close();

            } 

            else

            {

                stmt.close();

                DBConnectionPool.returnDBConnection(dbConn, serialNumber);

                throw new McdbDocNotFoundException("While trying to get the latest rev, could not find document " + localId);

            }

        } 

        catch (SQLException e) 

        {

            logMetacat.error("Error while looking up the guid: " 

                    + e.getMessage());

        } 

        finally 

        {

            // Return database connection to the pool

            DBConnectionPool.returnDBConnection(dbConn, serialNumber);

        }

        return rev;

    }

    /**

     * return all local ids in the object store that do not have associated

     * system metadata and are not themselves system metadata

     */

    public List<String> getLocalIdsWithNoSystemMetadata()

    {

        Vector<String> ids = new Vector<String>();

        //String sql = "select docid from identifier where guid in (select guid from systemmetadata)"; 

        //String sql = "select docid from xml_documents where docid not " +

        //    "in (select docid from identifier where guid in (select guid from systemmetadata)) " +

        //    "and doctype not like '" + DATAONE_SM_DOCTYPE + "'";

        String sql = "select docid, rev from xml_documents where docid not in " +

            "(select docid from systemmetadata) and (docid not in " +

            "(select docid from identifier where guid in (select guid from systemmetadata)))";

        DBConnection dbConn = null;

        int serialNumber = -1;

        try 

        {

            // Get a database connection from the pool

            dbConn = DBConnectionPool.getDBConnection("IdentifierManager.getLocalIdsWithNoSystemMetadata");

            serialNumber = dbConn.getCheckOutSerialNumber();

            // Execute the insert statement

            PreparedStatement stmt = dbConn.prepareStatement(sql);

            ResultSet rs = stmt.executeQuery();

            while (rs.next()) 

            {

                String localid = rs.getString(1);

                String rev = rs.getString(2);

                localid += "." + rev;

                System.out.println("id to add SM for: " + localid);

                ids.add(localid);

            } 

            stmt.close();

        } 

        catch (SQLException e) 

        {

            logMetacat.error("Error while looking up the guid: " 

                    + e.getMessage());

        } 

        finally 

        {

            // Return database connection to the pool

            DBConnectionPool.returnDBConnection(dbConn, serialNumber);

        }

        return ids;

    }

    /**

     * return a listing of all local ids in the object store

     * @return a list of all local ids in metacat

     */

    public List<String> getAllLocalIds()

       throws Exception

    {

        Vector<String> ids = new Vector<String>();

        String sql = "select docid from xml_documents";

        DBConnection dbConn = null;

        int serialNumber = -1;

        try 

        {

            // Get a database connection from the pool

            dbConn = DBConnectionPool.getDBConnection("IdentifierManager.getAllLocalIds");

            serialNumber = dbConn.getCheckOutSerialNumber();

            // Execute the insert statement

            PreparedStatement stmt = dbConn.prepareStatement(sql);

            ResultSet rs = stmt.executeQuery();

            while (rs.next()) 

            {

                String localid = rs.getString(1);

                ids.add(localid);

            } 

            stmt.close();

        } 

        catch (SQLException e) 

        {

            logMetacat.error("Error while looking up the guid: " 

                    + e.getMessage());

        } 

        finally 

        {

            // Return database connection to the pool

            DBConnectionPool.returnDBConnection(dbConn, serialNumber);

        }

        return ids;

    }

    /**

     * Lookup a GUID from the database, and return the associated LocalID. If

     * the identifier is not found, throw an exception.

     * 

     * @param guid the global identifier to look up

     * @return String containing the corresponding LocalId

     * @throws McdbDocNotFoundException if the identifier is not found

     */

    public String getLocalId(String guid) throws McdbDocNotFoundException

    {

        return this.getLocalId(guid, TYPE_IDENTIFIER);

    }

    /**

     * Determine if an identifier exists already, returning true if so.

     * 

     * @param guid the global identifier to look up

     * @return boolean true if the identifier exists

     */

    public boolean identifierExists(String guid)

    {

        boolean idExists = false;

        try {

            String id = getLocalId(guid);

            if (id != null) {

                idExists = true;

            }

        } catch (McdbDocNotFoundException e) {

            idExists = false;

        }

        return idExists;

    }

    /**

     * Create a mapping between two identifiers, one representing an 

     * unconstrained, globally unique string (guid), and one a localId 

     * in the Metacat docid format (scope.id.revision). 

     * This mapping allows one to find the global id (guid) from the localId.

     * 

     * @param guid the global string identifier to be mapped

     * @param localId the local identifier to be mapped

     */

    public void createMapping(String guid, String localId)

    {   

        createGenericMapping(guid, localId, TYPE_IDENTIFIER);

    }

    /**

     * 

     * @param guid

     * @param rev

     * @return

     */

    public String generateLocalId(String guid, int rev)

    {

        return generateLocalId(guid, rev, false);

    }

    /**

     * Given a global identifier (guid), create a suitable local identifier that

     * follows Metacat's docid semantics and format (scope.id.rev), and create

     * a mapping between these two identifiers.  This effectively reserves both

     * the global and the local identifier, as they will now be present in the

     * identifier mapping table.  If the incoming guid has the syntax of a

     * Metacat docid (scope.id.rev), then simply use it.

     * 

     * @param guid the global string identifier

     * @param rev the revision number to be used in the localId

     * @return String containing the localId to be used for Metacat operations

     */

    public String generateLocalId(String guid, int rev, boolean isSystemMetadata) 

    {

        String localId = "";

        boolean conformsToDocidFormat = false;

        // Check if the guid passed in is already in docid (scope.id.rev) format

        try {

            AccessionNumber acc = new AccessionNumber(guid, "NONE");

            if (new Integer(acc.getRev()).intValue() > 0) {

                conformsToDocidFormat = true;

            }

        } catch (NumberFormatException e) {

            // No action needed, simply detecting invalid AccessionNumbers

        } catch (AccessionNumberException e) {

            // No action needed, simply detecting invalid AccessionNumbers

        } catch (SQLException e) {

            // No action needed, simply detecting invalid AccessionNumbers

        }

        if (conformsToDocidFormat) {

            // if it conforms, use it for both guid and localId

            localId = guid;

        } else {

            // if not, then generate a new unique localId

            localId = DocumentUtil.generateDocumentId(rev);

        }

        // Register this new pair in the identifier mapping table

        System.out.println("creating mapping in generateLocalId");

        if(!isSystemMetadata)

        { //don't do this if we're generating for system metadata

            createMapping(guid, localId);

        }

        return localId;

    }

    /**

     * given a local identifer, look up the guid.  Throw McdbDocNotFoundException

     * if the docid, rev is not found in the identifiers or systemmetadata tables

     *

     * @param docid the docid to look up

     * @param rev the revision of the docid to look up

     * @return String containing the mapped guid

     * @throws McdbDocNotFoundException if the docid, rev is not found

     */

    public String getGUID(String docid, int rev)

      throws McdbDocNotFoundException

    {

        System.out.println("getting guid for " + docid);

        String query = "select guid from identifier where docid = ? and rev = ?";

        String guid = null;

        boolean found = false;

        DBConnection dbConn = null;

        int serialNumber = -1;

        try {

            // Get a database connection from the pool

            dbConn = DBConnectionPool.getDBConnection("IdentifierManager.getGUID");

            serialNumber = dbConn.getCheckOutSerialNumber();

            // Execute the insert statement

            PreparedStatement stmt = dbConn.prepareStatement(query);

            stmt.setString(1, docid);

            stmt.setInt(2, rev);

            ResultSet rs = stmt.executeQuery();

            if (rs.next()) 

            {

                guid = rs.getString(1);

            } 

            else

            {

                query = "select guid from systemmetadata where docid = ? and rev = ?";

                stmt = dbConn.prepareStatement(query);

                stmt.setString(1, docid);

                stmt.setInt(2, rev);

                rs = stmt.executeQuery();

                if(rs.next())

                {

                    guid = rs.getString(1);

                }

                else

                {

                    throw new McdbDocNotFoundException("No guid registered for docid " + docid + "." + rev);

                }

            }

        } catch (SQLException e) {

            logMetacat.error("Error while looking up the guid: " 

                    + e.getMessage());

        } finally {

            // Return database connection to the pool

            DBConnectionPool.returnDBConnection(dbConn, serialNumber);

        }

        return guid;

    }

    /**

     * creates a mapping for a system metadata document, but DOES NOT add any

     * of the extra system metadata information to the table.  You must call 

     * insertAdditionalSystemMetadataFields to complete the entry

     * 

     * @param guid the id to insert

     * @param localId the systemMetadata object to get the local id for

     */

    public void createSystemMetadataMapping(String guid, String localId)

    {

        createGenericMapping(guid, localId, TYPE_SYSTEM_METADATA);

    }

    /**

     * creates a system metadata mapping and adds additional fields from sysmeta

     * to the table for quick searching.

     * 

     * @param guid the id to insert

     * @param localId the systemMetadata object to get the local id for

     */

    public void createSystemMetadataMapping(SystemMetadata sysmeta, String localId)

    {

        createGenericMapping(sysmeta.getIdentifier().getValue(), localId, TYPE_SYSTEM_METADATA);

        insertAdditionalSystemMetadataFields(sysmeta);

    }

    /**

     * update a mapping

     * @param guid

     * @param localId

     */

    public void updateSystemMetadataMapping(String guid, String localId)

    {

        int serialNumber = -1;

        DBConnection dbConn = null;

        try {

            // Parse the localId into scope and rev parts

            AccessionNumber acc = new AccessionNumber(localId, "NOACTION");

            String docid = acc.getDocid();

            int rev = 1;

            if(acc.getRev() != null)

            {

              rev = (new Integer(acc.getRev()).intValue());

            }

            // Get a database connection from the pool

            dbConn = 

                DBConnectionPool.getDBConnection("IdentifierManager.createMapping");

            serialNumber = dbConn.getCheckOutSerialNumber();

            // Execute the insert statement

            String query = "update systemmetadata set (docid, rev) = (?, ?) where guid='" + guid + "'";

            //System.out.println("query: " + query + " for params: (guid:" + guid + ", docid=" + docid + ", rev=" + rev + ")");

            PreparedStatement stmt = dbConn.prepareStatement(query);

            stmt.setString(1, docid);

            stmt.setInt(2, rev);

            int rows = stmt.executeUpdate();

            stmt.close();

        } catch (SQLException e) {

            e.printStackTrace();

            logMetacat.error("SQL error while updating a mapping to the system metadata identifier: " 

                    + e.getMessage());

        } catch (NumberFormatException e) {

            e.printStackTrace();

            logMetacat.error("NumberFormat error while updating a mapping to the system metadata identifier: " 

                    + e.getMessage());

        } catch (AccessionNumberException e) {

            e.printStackTrace();

            logMetacat.error("AccessionNumber error while updating a mapping to the system metadata identifier: " 

                    + e.getMessage());

        } finally {

            // Return database connection to the pool

            DBConnectionPool.returnDBConnection(dbConn, serialNumber);

        }

    }

    /**

     * return the localId of a system metadata document based on its guid

     */

    public String getSystemMetadataLocalId(String guid)

      throws McdbDocNotFoundException

    {

      return this.getLocalId(guid, TYPE_SYSTEM_METADATA);

    }

    public void insertAdditionalSystemMetadataFields(long dateUploaded, String rightsHolder,

            String checksum, String checksumAlgorithm, String originMemberNode,

            String authoritativeMemberNode, long modifiedDate, String submitter, String guid)

    {

        DBConnection dbConn = null;

        int serialNumber = -1;

        try {

            // Get a database connection from the pool

            dbConn = 

                DBConnectionPool.getDBConnection("IdentifierManager.createMapping");

            serialNumber = dbConn.getCheckOutSerialNumber();

            // Execute the insert statement

            String query = "update " + TYPE_SYSTEM_METADATA + 

                " set (date_uploaded, rights_holder, checksum, checksum_algorithm, " +

                "origin_member_node, authoritive_member_node, date_modified, submitter) " +

                "= (?, ?, ?, ?, ?, ?, ?, ?) where guid = ?";

            PreparedStatement stmt = dbConn.prepareStatement(query);

            //data values

            stmt.setTimestamp(1, new java.sql.Timestamp(dateUploaded));

            stmt.setString(2, rightsHolder);

            stmt.setString(3, checksum);

            stmt.setString(4, checksumAlgorithm);

            stmt.setString(5, originMemberNode);

            stmt.setString(6, authoritativeMemberNode);

            stmt.setTimestamp(7, new java.sql.Timestamp(modifiedDate));

            stmt.setString(8, submitter);

            //where clause

            stmt.setString(9, guid);

            System.out.println("stmt: " + stmt.toString());

            //execute

            int rows = stmt.executeUpdate();

            stmt.close();

        } catch (SQLException e) {

            e.printStackTrace();

            logMetacat.error("SQL error while creating a mapping to the system metadata identifier: " 

                    + e.getMessage());

        } catch (NumberFormatException e) {

            e.printStackTrace();

            logMetacat.error("NumberFormat error while creating a mapping to the system metadata identifier: " 

                    + e.getMessage());

        } finally {

            // Return database connection to the pool

            DBConnectionPool.returnDBConnection(dbConn, serialNumber);

        }

    }

    /**

     * Insert the system metadata fields into the db

     * @param sm

     */

    public void insertAdditionalSystemMetadataFields(SystemMetadata sm)

    {

        insertAdditionalSystemMetadataFields(

                sm.getDateUploaded().getTime(),

                sm.getRightsHolder().getValue(), 

                sm.getChecksum().getValue(), 

                sm.getChecksum().getAlgorithm().name(), 

                sm.getOriginMemberNode().getValue(),

                sm.getAuthoritativeMemberNode().getValue(), 

                sm.getDateSysMetadataModified().getTime(),

                sm.getSubmitter().getValue(), 

                sm.getIdentifier().getValue());

    }

    /**

     * return a localId based on a guid.  The type can either be 'identifier' 

     * to get an id from the identifier table or 'systemmetadata' to get

     * the identifier from the systemidentifier table

     */

    private String getLocalId(String guid, String type)

      throws McdbDocNotFoundException

    {

      if(!type.equals(TYPE_IDENTIFIER) &&

        !type.equals(TYPE_SYSTEM_METADATA))

      {

        throw new RuntimeException("cannot lookup the identifier for type " + type +

          ".  Please choose 'identifier' or 'systemmetadata'.");

      }

      String db_guid = "";

      String docid = "";

      int rev = 0;

      String query = "select guid, docid, rev from " + type + " where guid = ?";

      //System.out.println("query: " + query + " for params: (" + guid + ")");

      DBConnection dbConn = null;

      int serialNumber = -1;

      try {

          // Get a database connection from the pool

          dbConn = DBConnectionPool.getDBConnection("Identifier.getLocalId");

          serialNumber = dbConn.getCheckOutSerialNumber();

          // Execute the insert statement

          PreparedStatement stmt = dbConn.prepareStatement(query);

          stmt.setString(1, guid);

          ResultSet rs = stmt.executeQuery();

          if (rs.next()) {

              db_guid = rs.getString(1);

              docid = rs.getString(2);

              rev = rs.getInt(3);

              assert(db_guid.equals(guid));

          } else {

              throw new McdbDocNotFoundException("Document not found:" + guid);

          }

          stmt.close();

      } catch (SQLException e) {

          logMetacat.error("Error while looking up the local identifier: " 

                  + e.getMessage());

      } finally {

          // Return database connection to the pool

          DBConnectionPool.returnDBConnection(dbConn, serialNumber);

      }

      return docid + "." + rev;

    }

    /**

     * query the systemmetadata table based on the given parameters

     * @param startTime

     * @param endTime

     * @param objectFormat

     * @param replicaStatus

     * @param start

     * @param count

     * @return ObjectList

     */

    public ObjectList querySystemMetadata(Date startTime, Date endTime, 

            ObjectFormat objectFormat, boolean replicaStatus, int start, int count)

    {

        ObjectList ol = new ObjectList();

        int total = 0;

        try

        {

            String sql = "select guid, date_uploaded, rights_holder, checksum, " +

                "checksum_algorithm, origin_member_node, authoritive_member_node, " +

                "date_modified, submitter, xml_documents.doctype from systemmetadata, " +

                "xml_documents where xml_documents.docid in " +

                "(select docid from identifier where guid = systemmetadata.guid)";

            boolean f1 = false;

            boolean f2 = false;

            boolean f3 = false;

            if(startTime != null)

            {

                sql += " and systemmetadata.date_modified > ?";

                f1 = true;

            }

            if(endTime != null)

            {

                sql += " and systemmetadata.date_modified < ?";

                f2 = true;

            }

            if(objectFormat != null)

            {

                sql += " and xml_documents.doctype = ?";

                f3 = true;

            }

            if(replicaStatus)

            {

                String currentNodeId = PropertyService.getInstance().getProperty("dataone.memberNodeId");

                sql += " and authoritive_member_node != '" + currentNodeId.trim() + "'";

            }

            DBConnection dbConn = DBConnectionPool.getDBConnection("IdentifierManager.createMapping");

            int serialNumber = dbConn.getCheckOutSerialNumber();

            PreparedStatement stmt = dbConn.prepareStatement(sql);

            if(f1 && f2 && f3)

            {

                stmt.setTimestamp(1, new Timestamp(startTime.getTime()));

                stmt.setTimestamp(2, new Timestamp(endTime.getTime()));

                stmt.setString(3, objectFormat.toString());

            }

            else if(f1 && f2 && !f3)

            {

                stmt.setTimestamp(1, new Timestamp(startTime.getTime()));

                stmt.setTimestamp(2, new Timestamp(endTime.getTime()));

            }

            else if(f1 && !f2 && f3)

            {

                stmt.setTimestamp(1, new Timestamp(startTime.getTime()));

                stmt.setString(2, objectFormat.toString());

            }

            else if(f1 && !f2 && !f3)

            {

                stmt.setTimestamp(1, new Timestamp(startTime.getTime()));

            }

            else if(!f1 && f2 && f3)

            {

                stmt.setTimestamp(1, new Timestamp(endTime.getTime()));

                stmt.setString(2, objectFormat.toString());

            }

            else if(!f1 && !f2 && f3)

            {

                stmt.setString(1, objectFormat.toString());

            }

            else if(!f1 && f2 && !f3)

            {

                stmt.setTimestamp(1, new Timestamp(endTime.getTime()));

            }

            //System.out.println("LISTOBJECTS QUERY: " + stmt.toString());

            ResultSet rs = stmt.executeQuery();

            for(int i=0; i<start - 1; i++)

            {

                rs.next();

            }

            int countIndex = 0;

            while(rs.next()) 

            {

                total++;

                if(countIndex >= count)

                {

                    break;

                }

                String guid = rs.getString(1);

                //System.out.println("query found doc with guid " + guid);

                //Timestamp dateUploaded = rs.getTimestamp(2);