Metacat

/**

 *  '$RCSfile$'

 *    Purpose: A Class that searches a relational DB for elements and

 *             attributes that have free text matches a query string,

 *             or structured query matches to a path specified node in the

 *             XML hierarchy.  It returns a result set consisting of the

 *             document ID for each document that satisfies the query

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

 *             National Center for Ecological Analysis and Synthesis

 *    Authors: Matt Jones

 *

 *   '$Author: leinfelder $'

 *     '$Date: 2010-12-27 14:47:01 -0800 (Mon, 27 Dec 2010) $'

 * '$Revision: 5760 $'

 *

 * 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.io.BufferedWriter;

import java.io.File;

import java.io.FileInputStream;

import java.io.FileOutputStream;

import java.io.IOException;

import java.io.InputStream;

import java.io.InputStreamReader;

import java.io.OutputStreamWriter;

import java.io.Reader;

import java.io.StringReader;

import java.io.StringWriter;

import java.io.Writer;

import java.sql.PreparedStatement;

import java.sql.ResultSet;

import java.sql.SQLException;

import java.util.Enumeration;

import java.util.Hashtable;

import java.util.Iterator;

import java.util.StringTokenizer;

import java.util.Vector;

import java.util.zip.ZipEntry;

import java.util.zip.ZipOutputStream;

import javax.servlet.ServletOutputStream;

import javax.servlet.http.HttpServletResponse;

import org.apache.log4j.Logger;

import edu.ucsb.nceas.metacat.accesscontrol.AccessControlInterface;

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.util.AuthUtil;

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

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

import edu.ucsb.nceas.morpho.datapackage.Triple;

import edu.ucsb.nceas.morpho.datapackage.TripleCollection;

import edu.ucsb.nceas.utilities.PropertyNotFoundException;

/**

 * A Class that searches a relational DB for elements and attributes that have

 * free text matches a query string, or structured query matches to a path

 * specified node in the XML hierarchy. It returns a result set consisting of

 * the document ID for each document that satisfies the query

 */

public class DBQuery

{

    static final int ALL = 1;

    static final int WRITE = 2;

    static final int READ = 4;

    private String qformat = "xml";

    //private Connection conn = null;

    private String parserName = null;

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

    /** true if the metacat spatial option is installed **/

    private final boolean METACAT_SPATIAL = true;

    /** useful if you just want to grab a list of docids. Since the docids can be very long,

         it is a vector of vector  **/

    Vector docidOverride = new Vector();

    // a hash table serves as query reuslt cache. Key of hashtable

    // is a query string and value is result xml string

    private static Hashtable queryResultCache = new Hashtable();

    // Capacity of the query result cache

    private static final int QUERYRESULTCACHESIZE;

    static {

    	int qryRsltCacheSize = 0;

    	try {

    		qryRsltCacheSize = Integer.parseInt(PropertyService.getProperty("database.queryresultCacheSize"));

    	} catch (PropertyNotFoundException pnfe) {

    		System.err.println("Could not get QUERYRESULTCACHESIZE property in static block: "

					+ pnfe.getMessage());

    	}

    	QUERYRESULTCACHESIZE = qryRsltCacheSize;

    }

    // Size of page for non paged query

    private static final int NONPAGESIZE = 99999999;

    /**

     * the main routine used to test the DBQuery utility.

     * <p>

     * Usage: java DBQuery <xmlfile>

     * NOTE: encoding should be provided for best results

     * @param xmlfile the filename of the xml file containing the query

     */

    static public void main(String[] args)

    {

        if (args.length < 1) {

            System.err.println("Wrong number of arguments!!!");

            System.err.println("USAGE: java DBQuery [-t] [-index] <xmlfile>");

            return;

        } else {

            try {

                int i = 0;

                boolean showRuntime = false;

                boolean useXMLIndex = false;

                if (args[i].equals("-t")) {

                    showRuntime = true;

                    i++;

                }

                if (args[i].equals("-index")) {

                    useXMLIndex = true;

                    i++;

                }

                String xmlfile = args[i];

                // Time the request if asked for

                double startTime = System.currentTimeMillis();

                // Open a connection to the database

                //Connection dbconn = util.openDBConnection();

                double connTime = System.currentTimeMillis();

                // Execute the query

                DBQuery queryobj = new DBQuery();

                Reader xml = new InputStreamReader(new FileInputStream(new File(xmlfile)));

                Hashtable nodelist = null;

                //nodelist = queryobj.findDocuments(xml, null, null, useXMLIndex);

                // Print the reulting document listing

                StringBuffer result = new StringBuffer();

                String document = null;

                String docid = null;

                result.append("<?xml version=\"1.0\"?>\n");

                result.append("<resultset>\n");

                if (!showRuntime) {

                    Enumeration doclist = nodelist.keys();

                    while (doclist.hasMoreElements()) {

                        docid = (String) doclist.nextElement();

                        document = (String) nodelist.get(docid);

                        result.append("  <document>\n    " + document

                                + "\n  </document>\n");

                    }

                    result.append("</resultset>\n");

                }

                // Time the request if asked for

                double stopTime = System.currentTimeMillis();

                double dbOpenTime = (connTime - startTime) / 1000;

                double readTime = (stopTime - connTime) / 1000;

                double executionTime = (stopTime - startTime) / 1000;

                if (showRuntime) {

                    System.out.print("  " + executionTime);

                    System.out.print("  " + dbOpenTime);

                    System.out.print("  " + readTime);

                    System.out.print("  " + nodelist.size());

                    System.out.println();

                }

                //System.out.println(result);

                //write into a file "result.txt"

                if (!showRuntime) {

                    File f = new File("./result.txt");

                    Writer fw = new OutputStreamWriter(new FileOutputStream(f));

                    BufferedWriter out = new BufferedWriter(fw);

                    out.write(result.toString());

                    out.flush();

                    out.close();

                    fw.close();

                }

            } catch (Exception e) {

                System.err.println("Error in DBQuery.main");

                System.err.println(e.getMessage());

                e.printStackTrace(System.err);

            }

        }

    }

    /**

     * construct an instance of the DBQuery class

     *

     * <p>

     * Generally, one would call the findDocuments() routine after creating an

     * instance to specify the search query

     * </p>

     *

     * @param parserName the fully qualified name of a Java class implementing

     *            the org.xml.sax.XMLReader interface

     */

    public DBQuery() throws PropertyNotFoundException

    {

        String parserName = PropertyService.getProperty("xml.saxparser");

        this.parserName = parserName;

    }

    /**

     * 

     * Construct an instance of DBQuery Class

     * BUT accept a docid Vector that will supersede

     * the query.printSQL() method

     *

     * If a docid Vector is passed in,

     * the docids will be used to create a simple IN query 

     * without the multiple subselects of the printSQL() method

     *

     * Using this constructor, we just check for 

     * a docidOverride Vector in the findResultDoclist() method

     *

     * @param docids List of docids to display in the resultset

     */

    public DBQuery(Vector docids) throws PropertyNotFoundException

    {

    	// since the query will be too long to be handled, so we divided the 

    	// docids vector into couple vectors.

    	int size = (new Integer(PropertyService.getProperty("database.appResultsetSize"))).intValue();

    	logMetacat.info("DBQuery.DBQuery - The size of select doicds is "+docids.size());

    	logMetacat.info("DBQuery.DBQuery - The application result size in metacat.properties is "+size);

    	Vector subset = new Vector();

    	if (docids != null && docids.size() > size)

    	{

    		int index = 0;

    		for (int i=0; i< docids.size(); i++)

    		{

    			if (index < size)

    			{  	

    				subset.add(docids.elementAt(i));

    				index ++;

    			}

    			else

    			{

    				docidOverride.add(subset);

    				subset = new Vector();

    				subset.add(docids.elementAt(i));

    			    index = 1;

    			}

    		}

    		if (!subset.isEmpty())

    		{

    			docidOverride.add(subset);

    		}

    	}

    	else

    	{

    		this.docidOverride.add(docids);

    	}

        String parserName = PropertyService.getProperty("xml.saxparser");

        this.parserName = parserName;

    }

  /**

   * Method put the search result set into out printerwriter

   * @param resoponse the return response

   * @param out the output printer

   * @param params the paratermer hashtable

   * @param user the user name (it maybe different to the one in param)

   * @param groups the group array

   * @param sessionid  the sessionid

   */

  public void findDocuments(HttpServletResponse response,

                                       Writer out, Hashtable params,

                                       String user, String[] groups,

                                       String sessionid) throws PropertyNotFoundException

  {

    boolean useXMLIndex = (new Boolean(PropertyService.getProperty("database.usexmlindex")))

               .booleanValue();

    findDocuments(response, out, params, user, groups, sessionid, useXMLIndex);

  }

    /**

     * Method put the search result set into out printerwriter

     * @param resoponse the return response

     * @param out the output printer

     * @param params the paratermer hashtable

     * @param user the user name (it maybe different to the one in param)

     * @param groups the group array

     * @param sessionid  the sessionid

     */

    public void findDocuments(HttpServletResponse response,

                                         Writer out, Hashtable params,

                                         String user, String[] groups,

                                         String sessionid, boolean useXMLIndex)

    {

      int pagesize = 0;

      int pagestart = 0;

      long transferWarnLimit = 0; 

      if(params.containsKey("pagesize") && params.containsKey("pagestart"))

      {

        String pagesizeStr = ((String[])params.get("pagesize"))[0];

        String pagestartStr = ((String[])params.get("pagestart"))[0];

        if(pagesizeStr != null && pagestartStr != null)

        {

          pagesize = (new Integer(pagesizeStr)).intValue();

          pagestart = (new Integer(pagestartStr)).intValue();

        }

      }

      String xmlquery = null;

      String qformat = null;

      // get query and qformat

      try {

    	xmlquery = ((String[])params.get("query"))[0];

        logMetacat.info("DBQuery.findDocuments - SESSIONID: " + sessionid);

        logMetacat.info("DBQuery.findDocuments - xmlquery: " + xmlquery);

        qformat = ((String[])params.get("qformat"))[0];

        logMetacat.info("DBQuery.findDocuments - qformat: " + qformat);

      }

      catch (Exception ee)

      {

        logMetacat.error("DBQuery.findDocuments - Couldn't retrieve xmlquery or qformat value from "

                  +"params hashtable in DBQuery.findDocuments: "

                  + ee.getMessage()); 

      }

      // Get the XML query and covert it into a SQL statment

      QuerySpecification qspec = null;

      if ( xmlquery != null)

      {

         xmlquery = transformQuery(xmlquery);

         try

         {

           qspec = new QuerySpecification(xmlquery,

                                          parserName,

                                          PropertyService.getProperty("document.accNumSeparator"));

         }

         catch (Exception ee)

         {

           logMetacat.error("DBQuery.findDocuments - error generating QuerySpecification object: "

                                    + ee.getMessage());

         }

      }

      if (qformat != null && qformat.equals(MetacatUtil.XMLFORMAT))

      {

        //xml format

        if(response != null)

        {

            response.setContentType("text/xml");

        }

        createResultDocument(xmlquery, qspec, out, user, groups, useXMLIndex, 

          pagesize, pagestart, sessionid, qformat);

      }//if

      else

      {

        //knb format, in this case we will get whole result and sent it out

        response.setContentType("text/html");

        Writer nonout = null;

        StringBuffer xml = createResultDocument(xmlquery, qspec, nonout, user,

                                                groups, useXMLIndex, pagesize, 

                                                pagestart, sessionid, qformat);

        //transfer the xml to html

        try

        {

         long startHTMLTransform = System.currentTimeMillis();

         DBTransform trans = new DBTransform();

         response.setContentType("text/html");

         // if the user is a moderator, then pass a param to the 

         // xsl specifying the fact

         if(AuthUtil.isModerator(user, groups)){

        	 params.put("isModerator", new String[] {"true"});

         }

         trans.transformXMLDocument(xml.toString(), "-//NCEAS//resultset//EN",

                                 "-//W3C//HTML//EN", qformat, out, params,

                                 sessionid);

         long transformRunTime = System.currentTimeMillis() - startHTMLTransform;

         transferWarnLimit = Long.parseLong(PropertyService.getProperty("dbquery.transformTimeWarnLimit"));

         if (transformRunTime > transferWarnLimit) {

         	logMetacat.warn("DBQuery.findDocuments - The time to transfrom resultset from xml to html format is "

                  		                             + transformRunTime);

         }

          MetacatUtil.writeDebugToFile("---------------------------------------------------------------------------------------------------------------Transfrom xml to html  "

                             + transformRunTime);

          MetacatUtil.writeDebugToDelimiteredFile(" " + transformRunTime, false);

        }

        catch(Exception e)

        {

         logMetacat.error("DBQuery.findDocuments - Error in MetaCatServlet.transformResultset:"

                                +e.getMessage());

         }

      }//else

  }

  /**

   * Transforms a hashtable of documents to an xml or html result and sent

   * the content to outputstream. Keep going untill hastable is empty. stop it.

   * add the QuerySpecification as parameter is for ecogrid. But it is duplicate

   * to xmlquery String

   * @param xmlquery

   * @param qspec

   * @param out

   * @param user

   * @param groups

   * @param useXMLIndex

   * @param sessionid

   * @return

   */

    public StringBuffer createResultDocument(String xmlquery,

                                              QuerySpecification qspec,

                                              Writer out,

                                              String user, String[] groups,

                                              boolean useXMLIndex)

    {

    	return createResultDocument(xmlquery,qspec,out, user,groups, useXMLIndex, 0, 0,"", qformat);

    }

  /*

   * Transforms a hashtable of documents to an xml or html result and sent

   * the content to outputstream. Keep going untill hastable is empty. stop it.

   * add the QuerySpecification as parameter is for ecogrid. But it is duplicate

   * to xmlquery String

   */

  public StringBuffer createResultDocument(String xmlquery,

                                            QuerySpecification qspec,

                                            Writer out,

                                            String user, String[] groups,

                                            boolean useXMLIndex, int pagesize,

                                            int pagestart, String sessionid, 

                                            String qformat)

  {

    DBConnection dbconn = null;

    int serialNumber = -1;

    StringBuffer resultset = new StringBuffer();

    //try to get the cached version first    

    // Hashtable sessionHash = MetaCatServlet.getSessionHash();

    // HttpSession sess = (HttpSession)sessionHash.get(sessionid);

    resultset.append("<?xml version=\"1.0\"?>\n");

    resultset.append("<resultset>\n");

    resultset.append("  <pagestart>" + pagestart + "</pagestart>\n");

    resultset.append("  <pagesize>" + pagesize + "</pagesize>\n");

    resultset.append("  <nextpage>" + (pagestart + 1) + "</nextpage>\n");

    resultset.append("  <previouspage>" + (pagestart - 1) + "</previouspage>\n");

    resultset.append("  <query>" + xmlquery + "</query>");

    //send out a new query

    if (out != null)

    {

    	try {

    	  out.write(resultset.toString());

		} catch (IOException e) {

			logMetacat.error(e.getMessage(), e);

		}

    }

    if (qspec != null)

    {

      try

      {

        //checkout the dbconnection

        dbconn = DBConnectionPool.getDBConnection("DBQuery.findDocuments");

        serialNumber = dbconn.getCheckOutSerialNumber();

        //print out the search result

        // search the doc list

        Vector givenDocids = new Vector();

        StringBuffer resultContent = new StringBuffer();

        if (docidOverride == null || docidOverride.size() == 0)

        {

        	logMetacat.debug("DBQuery.createResultDocument - Not in map query");

        	resultContent = findResultDoclist(qspec, out, user, groups,

                    dbconn, useXMLIndex, pagesize, pagestart, 

                    sessionid, givenDocids, qformat);

        }

        else

        {

        	logMetacat.debug("DBQuery.createResultDocument - In map query");

        	// since docid can be too long to be handled. We divide it into several parts

        	for (int i= 0; i<docidOverride.size(); i++)

        	{

        	   logMetacat.debug("DBQuery.createResultDocument - in loop===== "+i);

        		givenDocids = (Vector)docidOverride.elementAt(i);

        		StringBuffer subset = findResultDoclist(qspec, out, user, groups,

                        dbconn, useXMLIndex, pagesize, pagestart, 

                        sessionid, givenDocids, qformat);

        		resultContent.append(subset);

        	}

        }

        resultset.append(resultContent);

      } //try

      catch (IOException ioe)

      {

        logMetacat.error("DBQuery.createResultDocument - IO error: " + ioe.getMessage());

      }

      catch (SQLException e)

      {

        logMetacat.error("DBQuery.createResultDocument - SQL Error: " + e.getMessage());

      }

      catch (Exception ee)

      {

        logMetacat.error("DBQuery.createResultDocument - General exception: "

                                 + ee.getMessage());

        ee.printStackTrace();

      }

      finally

      {

        DBConnectionPool.returnDBConnection(dbconn, serialNumber);

      } //finally

    }//if

    String closeRestultset = "</resultset>";

    resultset.append(closeRestultset);

    if (out != null)

    {

      try {

		out.write(closeRestultset);

		} catch (IOException e) {

			logMetacat.error(e.getMessage(), e);

		}

    }

    //default to returning the whole resultset

    return resultset;

  }//createResultDocuments

    /*

     * Find the doc list which match the query

     */

    private StringBuffer findResultDoclist(QuerySpecification qspec,

                                      Writer out,

                                      String user, String[]groups,

                                      DBConnection dbconn, boolean useXMLIndex,

                                      int pagesize, int pagestart, String sessionid, 

                                      Vector givenDocids, String qformat)

                                      throws Exception

    {

      StringBuffer resultsetBuffer = new StringBuffer();

      String query = null;

      int count = 0;

      int index = 0;

      ResultDocumentSet docListResult = new ResultDocumentSet();

      PreparedStatement pstmt = null;

      String docid = null;

      String docname = null;

      String doctype = null;

      String createDate = null;

      String updateDate = null;

      StringBuffer document = null;

      boolean lastpage = false;

      int rev = 0;

      double startTime = 0;

      int offset = 1;

      long startSelectionTime = System.currentTimeMillis();

      ResultSet rs = null;

      // this is a hack for offset. in postgresql 7, if the returned docid list is too long,

      //the extend query which base on the docid will be too long to be run. So we 

      // have to cut them into different parts. Page query don't need it somehow.

      if (out == null)

      {

        // for html page, we put everything into one page

        offset =

            (new Integer(PropertyService.getProperty("database.webResultsetSize"))).intValue();

      }

      else

      {

          offset =

              (new Integer(PropertyService.getProperty("database.appResultsetSize"))).intValue();

      }

      /*

       * Check the docidOverride Vector

       * if defined, we bypass the qspec.printSQL() method

       * and contruct a simpler query based on a 

       * list of docids rather than a bunch of subselects

       */

      if ( givenDocids == null || givenDocids.size() == 0 ) {

          query = qspec.printSQL(useXMLIndex);

      } else {

          logMetacat.info("DBQuery.findResultDoclist - docid override " + givenDocids.size());

          StringBuffer queryBuffer = new StringBuffer( "SELECT docid,docname,doctype,date_created, date_updated, rev " );

          queryBuffer.append( " FROM xml_documents WHERE docid IN (" );

          for (int i = 0; i < givenDocids.size(); i++) {  

              queryBuffer.append("'");

              queryBuffer.append( (String)givenDocids.elementAt(i) );

              queryBuffer.append("',");

          }

          // empty string hack 

          queryBuffer.append( "'') " );

          query = queryBuffer.toString();

      } 

      String ownerQuery = getOwnerQuery(user);

      //logMetacat.debug("query: " + query);

      logMetacat.debug("DBQuery.findResultDoclist - owner query: " + ownerQuery);

      // if query is not the owner query, we need to check the permission

      // otherwise we don't need (owner has all permission by default)

      if (!query.equals(ownerQuery))

      {

        // set user name and group

        qspec.setUserName(user);

        qspec.setGroup(groups);

        // Get access query

        String accessQuery = qspec.getAccessQuery();

        if(!query.endsWith("WHERE")){

            query = query + accessQuery;

        } else {

            query = query + accessQuery.substring(4, accessQuery.length());

        }

      }

      logMetacat.debug("DBQuery.findResultDoclist - final selection query: " + query);

      String selectionAndExtendedQuery = null;

      // we only get cache for public

      if (user != null && user.equalsIgnoreCase("public") 

     		 && pagesize == 0 && PropertyService.getProperty("database.queryCacheOn").equals("true"))

      {

    	  selectionAndExtendedQuery = query +qspec.getReturnDocList()+qspec.getReturnFieldList();

   	      String cachedResult = getResultXMLFromCache(selectionAndExtendedQuery);

   	      logMetacat.debug("DBQuery.findResultDoclist - The key of query cache is " + selectionAndExtendedQuery);

   	      //System.out.println("==========the string from cache is "+cachedResult);

   	      if (cachedResult != null)

   	      {

   	    	logMetacat.info("DBQuery.findResultDoclist - result from cache !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!");

   	    	 if (out != null)

   	         {

   	             out.write(cachedResult);

   	         }

   	    	 resultsetBuffer.append(cachedResult);

   	    	 return resultsetBuffer;

   	      }

      }

      startTime = System.currentTimeMillis() / 1000;

      pstmt = dbconn.prepareStatement(query);

      rs = pstmt.executeQuery();

      double queryExecuteTime = System.currentTimeMillis() / 1000;

      logMetacat.debug("DBQuery.findResultDoclist - Time to execute select docid query is "

                    + (queryExecuteTime - startTime));

      MetacatUtil.writeDebugToFile("\n\n\n\n\n\nExecute selection query  "

              + (queryExecuteTime - startTime));

      MetacatUtil.writeDebugToDelimiteredFile(""+(queryExecuteTime - startTime), false);

      boolean tableHasRows = rs.next();

      if(pagesize == 0)

      { //this makes sure we get all results if there is no paging

        pagesize = NONPAGESIZE;

        pagestart = NONPAGESIZE;

      } 

      int currentIndex = 0;

      while (tableHasRows)

      {

        logMetacat.debug("DBQuery.findResultDoclist - getting result: " + currentIndex);

        docid = rs.getString(1).trim();

        logMetacat.debug("DBQuery.findResultDoclist -  processing: " + docid);

        docname = rs.getString(2);

        doctype = rs.getString(3);

        logMetacat.debug("DBQuery.findResultDoclist - processing: " + doctype);

        createDate = rs.getString(4);

        updateDate = rs.getString(5);

        rev = rs.getInt(6);

         Vector returndocVec = qspec.getReturnDocList();

       if (returndocVec.size() == 0 || returndocVec.contains(doctype))

        {

          logMetacat.debug("DBQuery.findResultDoclist - NOT Back tracing now...");

           document = new StringBuffer();

           String completeDocid = docid

                            + PropertyService.getProperty("document.accNumSeparator");

           completeDocid += rev;

           document.append("<docid>").append(completeDocid).append("</docid>");

           if (docname != null)

           {

               document.append("<docname>" + docname + "</docname>");

           }

           if (doctype != null)

           {

              document.append("<doctype>" + doctype + "</doctype>");

           }

           if (createDate != null)

           {

               document.append("<createdate>" + createDate + "</createdate>");

           }

           if (updateDate != null)

           {

             document.append("<updatedate>" + updateDate + "</updatedate>");

           }

           // Store the document id and the root node id

           docListResult.addResultDocument(

             new ResultDocument(docid, (String) document.toString()));

           logMetacat.info("DBQuery.findResultDoclist - real result: " + docid);

           currentIndex++;

           count++;

        }//else

        // when doclist reached the offset number, send out doc list and empty

        // the hash table

        if (count == offset && pagesize == NONPAGESIZE)

        { //if pagesize is not 0, do this later.

          //reset count

          //logMetacat.warn("############doing subset cache");

          count = 0;

          handleSubsetResult(qspec, resultsetBuffer, out, docListResult,

                              user, groups,dbconn, useXMLIndex, qformat);

          //reset docListResult

          docListResult = new ResultDocumentSet();

        }

       logMetacat.debug("DBQuery.findResultDoclist - currentIndex: " + currentIndex);

       logMetacat.debug("DBQuery.findResultDoclist - page comparator: " + (pagesize * pagestart) + pagesize);

       if(currentIndex >= ((pagesize * pagestart) + pagesize))

       {

         ResultDocumentSet pagedResultsHash = new ResultDocumentSet();

         for(int i=pagesize*pagestart; i<docListResult.size(); i++)

         {

           pagedResultsHash.put(docListResult.get(i));

         }

         docListResult = pagedResultsHash;

         break;

       }

       // Advance to the next record in the cursor

       tableHasRows = rs.next();

       if(!tableHasRows)

       {

         ResultDocumentSet pagedResultsHash = new ResultDocumentSet();

         //get the last page of information then break

         if(pagesize != NONPAGESIZE)

         {

           for(int i=pagesize*pagestart; i<docListResult.size(); i++)

           {

             pagedResultsHash.put(docListResult.get(i));

           }

           docListResult = pagedResultsHash;

         }

         lastpage = true;

         break;

       }

     }//while

     rs.close();

     pstmt.close();

     long docListTime = System.currentTimeMillis() - startSelectionTime;

     long docListWarnLimit = Long.parseLong(PropertyService.getProperty("dbquery.findDocListTimeWarnLimit"));

     if (docListTime > docListWarnLimit) {

    	 logMetacat.warn("DBQuery.findResultDoclist - Total time to get docid list is: "

                          + docListTime);

     }

     MetacatUtil.writeDebugToFile("---------------------------------------------------------------------------------------------------------------Total selection: "

             + docListTime);

     MetacatUtil.writeDebugToDelimiteredFile(" "+ docListTime, false);

     //if docListResult is not empty, it need to be sent.

     if (docListResult.size() != 0)

     {

       handleSubsetResult(qspec,resultsetBuffer, out, docListResult,

                              user, groups,dbconn, useXMLIndex, qformat);

     }

     resultsetBuffer.append("\n<lastpage>" + lastpage + "</lastpage>\n");

     if (out != null)

     {

         out.write("\n<lastpage>" + lastpage + "</lastpage>\n");

     }

     // now we only cached none-paged query and user is public

     if (user != null && user.equalsIgnoreCase("public") 

    		 && pagesize == NONPAGESIZE && PropertyService.getProperty("database.queryCacheOn").equals("true"))

     {

       //System.out.println("the string stored into cache is "+ resultsetBuffer.toString());

  	   storeQueryResultIntoCache(selectionAndExtendedQuery, resultsetBuffer.toString());

     }

     return resultsetBuffer;

    }//findReturnDoclist

    /*

     * Send completed search hashtable(part of reulst)to output stream

     * and buffer into a buffer stream

     */

    private StringBuffer handleSubsetResult(QuerySpecification qspec,

                                           StringBuffer resultset,

                                           Writer out, ResultDocumentSet partOfDoclist,

                                           String user, String[]groups,

                                       DBConnection dbconn, boolean useXMLIndex,

                                       String qformat)

                                       throws Exception

   {

     double startReturnFieldTime = System.currentTimeMillis();

     // check if there is a record in xml_returnfield

     // and get the returnfield_id and usage count

     int usage_count = getXmlReturnfieldsTableId(qspec, dbconn);

     boolean enterRecords = false;

     // get value of database.xmlReturnfieldCount

     int count = (new Integer(PropertyService

                            .getProperty("database.xmlReturnfieldCount")))

                            .intValue();

     // set enterRecords to true if usage_count is more than the offset

     // specified in metacat.properties

     if(usage_count > count){

         enterRecords = true;

     }

     if(returnfield_id < 0){

         logMetacat.warn("DBQuery.handleSubsetResult - Error in getting returnfield id from"

                                  + "xml_returnfield table");

         enterRecords = false;

     }

     // get the hashtable containing the docids that already in the

     // xml_queryresult table

     logMetacat.info("DBQuery.handleSubsetResult - size of partOfDoclist before"

                             + " docidsInQueryresultTable(): "

                             + partOfDoclist.size());

     long startGetReturnValueFromQueryresultable = System.currentTimeMillis();

     Hashtable queryresultDocList = docidsInQueryresultTable(returnfield_id,

                                                        partOfDoclist, dbconn);

     // remove the keys in queryresultDocList from partOfDoclist

     Enumeration _keys = queryresultDocList.keys();

     while (_keys.hasMoreElements()){

         partOfDoclist.remove((String)_keys.nextElement());

     }

     long queryResultReturnValuetime = System.currentTimeMillis() - startGetReturnValueFromQueryresultable;

     long queryResultWarnLimit = 

    	 Long.parseLong(PropertyService.getProperty("dbquery.findQueryResultsTimeWarnLimit"));

     if (queryResultReturnValuetime > queryResultWarnLimit) {

    	 logMetacat.warn("DBQuery.handleSubsetResult - Time to get return fields from xml_queryresult table is (Part1 in return fields) " +

    		 queryResultReturnValuetime);

     }

     MetacatUtil.writeDebugToFile("-----------------------------------------Get fields from xml_queryresult(Part1 in return fields) " +

    		 queryResultReturnValuetime);

     MetacatUtil.writeDebugToDelimiteredFile(" " + queryResultReturnValuetime,false);

     long startExtendedQuery = System.currentTimeMillis();