//
// See the file LICENSE for redistribution information.
//
// Copyright (c) 2002-2003
//	Sleepycat Software.  All rights reserved.
//

static const char revid[] = "$Id: Container.cpp,v 1.176 2004/01/26 18:45:18 gmf Exp $";

#include "dbxml_config.h"
#include "dbxml/XmlPortability.hpp"
#include "dbxml/XmlContainer.hpp"
#include "dbxml/XmlDocument.hpp"
#include "dbxml/XmlIndexSpecification.hpp"
#include "dbxml/XmlResolver.hpp"
#include "dbxml/XmlQueryExpression.hpp"
#include "dbxml/XmlUpdateContext.hpp"
#include "TypeConversions.hpp"
#include "Container.hpp"
#include "SyntaxManager.hpp"
#include "Cursor.hpp"
#include "OperationContext.hpp"
#include "Results.hpp"
#include "Indexer.hpp"
#include "HighResTimer.hpp"
#include "QueryExpression.hpp"
#include "UpdateContext.hpp"
#include "Name.hpp"
#include "Document.hpp"
#include "Value.hpp"
#include "Modify.hpp"

#include <algorithm>
#include <iostream>
#include <memory>
#include <cassert>

#if defined(DBXML_DOM_XERCES2)
#include <xercesc/util/PlatformUtils.hpp>
#if defined(XERCES_HAS_CPP_NAMESPACE)
  XERCES_CPP_NAMESPACE_USE
#endif
#endif

#include "XPathLexer.hpp"
#include "XPathParser.hpp"
#include "XPathProjectionTreeParser.hpp"
#include "XPathSelectionTreeParser.hpp"

using namespace DbXml;

/*
 * This is the version of the container format
 * supported by this release.  If it does not
 * match that of an opened container, upgrade is required.
 */
int Container::version_ = 2;
#define VERSION_11 10100
#define VERSION_120_COMPAT 10200

Container::Container()
	: open_(false),
	environment_(0),
	databaseCreationFlags_(0),
	pageSize_(0),
	nidId_(0),
	nidName_(0),
	nidContent_(0),
	resolver_(0)
{
	XMLPlatformUtils::Initialize();
}

Container::~Container()
{
	environment_ = 0;
	XMLPlatformUtils::Terminate();
}

void Container::init(DbEnv *environment, const std::string &name, u_int32_t flags)
{
	environment_ = environment;
	name_ = name;
	databaseCreationFlags_ = flags;
}

void Container::constructDatabases()
{
	// Create the databases
	//  Document Primary        - assigns the document ids
	//  Dictionary Primary      - maps names to numbers
	//  Configuration Secondary - container configuration
	//  Document Secondary      - maps document id onto document metadata (including document name and content)
	//  Dictionary Secondary    - maps numbers to names
	//
	databaseDocumentPrimary_.reset(new PrimaryDatabase(environment_, name_, "document", pageSize_, databaseCreationFlags_));
	databaseDictionaryPrimary_.reset(new PrimaryDatabase(environment_, name_, "dictionary", pageSize_, databaseCreationFlags_));
	databaseConfiguration_.reset(new SecondaryDatabase(environment_, name_, "configuration", 0, pageSize_, databaseCreationFlags_));
	databaseDocumentSecondary_.reset(new SecondaryDatabase(environment_, name_, "document", 0, pageSize_, databaseCreationFlags_));
	databaseDictionarySecondary_.reset(new SecondaryDatabase(environment_, name_, "dictionary", 0, pageSize_, databaseCreationFlags_));
	//
	// Iterate over the registered syntax types, creating an index
	// and statistics database for each.
	//
	{
		databaseIndexes_.resize(SyntaxManager::getInstance()->size());
		databaseStatistics_.resize(SyntaxManager::getInstance()->size());
		int i = 0;
		const Syntax *syntax = 0;
		syntax = SyntaxManager::getInstance()->getNextSyntax(i);
		while (syntax != 0)
		{
			string document_index_("document_index_");
			string document_statistics_("document_statistics_");
			SecondaryDatabase::Ptr databaseI(new SecondaryDatabase(environment_, name_, document_index_ + syntax->getName(), syntax, pageSize_, databaseCreationFlags_));
			SecondaryDatabase::Ptr databaseS(new SecondaryDatabase(environment_, name_, document_statistics_ + syntax->getName(), 0, pageSize_, databaseCreationFlags_));
			databaseIndexes_[syntax->getType()] = databaseI;
			databaseStatistics_[syntax->getType()] = databaseS;
			syntax = SyntaxManager::getInstance()->getNextSyntax(i);
		}
	}
	//
	// Build an array of Database pointers for the databases.
	// This is handy for when we need to perform some DB operation
	// against all the databases.
	//
	{
		databases_.resize(5 + (2*SyntaxManager::getInstance()->size()));
		int dn = 0;
		databases_[dn++] = databaseDocumentPrimary_.get();
		databases_[dn++] = databaseDictionaryPrimary_.get();
		databases_[dn++] = databaseConfiguration_.get();
		databases_[dn++] = databaseDocumentSecondary_.get();
		databases_[dn++] = databaseDictionarySecondary_.get();
		SecondaryDatabase::Vector::iterator i;
		for (i = databaseIndexes_.begin();i != databaseIndexes_.end();++i)
		{
			databases_[dn++] = i->get();
		}
		for (i = databaseStatistics_.begin();i != databaseStatistics_.end();++i)
		{
			databases_[dn++] = i->get();
		}
	}
}

void Container::destructDatabases()
{
	databaseDocumentPrimary_.reset(0);
	databaseDictionaryPrimary_.reset(0);
	databaseConfiguration_.reset(0);
	databaseDocumentSecondary_.reset(0);
	databaseDictionarySecondary_.reset(0);
	SecondaryDatabase::Vector::iterator i;
	for (i = databaseIndexes_.begin(); i != databaseIndexes_.end(); ++i)
		i->reset(0);
	for (i = databaseStatistics_.begin(); i != databaseStatistics_.end(); ++i)
		i->reset(0);
	databases_.resize(0);
}

int Container::setPageSize(u_int32_t pagesize)
{
	if (pagesize < 512 || pagesize > 65536)
		throw XmlException(XmlException::INVALID_VALUE,
				   "setPageSize expects a page size between 512 bytes and 64k");
	pageSize_ = pagesize;
	return 0;
}

void Container::setResolver(const XmlResolver *resolver)
{
	resolver_= resolver;
}

int Container::open(DbTxn *txn, u_int32_t flags, int mode, bool doVersionCheck)
{
	int err = 0;
	//
	// Create/Open Databases
	//
	constructDatabases();
	try {
		if (err == 0)
			err = databaseDocumentPrimary_->open(txn, flags, mode);
		if (err == 0)
			err = databaseDictionaryPrimary_->open(txn, flags, mode);
		if (err == 0)
			err = databaseConfiguration_->open(txn, false, flags, mode);
		if (err == 0)
			err = databaseDocumentSecondary_->open(txn, false, flags, mode);
		if (err == 0)
			err = databaseDictionarySecondary_->open(txn, true, flags, mode);
		//
		// Iterate over the registered syntax types, opening each of the
		// index and statistics databases.
		//
		SecondaryDatabase::Vector::iterator i;
		for (i = databaseIndexes_.begin();i != databaseIndexes_.end();++i) {
			if (err == 0)
				err = (*i)->open(txn, true, flags, mode);
		}
		for (i = databaseStatistics_.begin();i != databaseStatistics_.end();++i) {
			if (err == 0)
				err = (*i)->open(txn, false, flags, mode);
		}
		//
		// Lookup/Define the dbxml namespace names
		//
		if (err == 0)
			err = lookupName(txn, Name::dbxml_colon_id, nidId_, true /* define */);
		if (err == 0)
			err = lookupName(txn, Name::dbxml_colon_name, nidName_, true /* define */);
		if (err == 0)
			err = lookupName(txn, Name::dbxml_colon_content, nidContent_, true /* define */);
		if (err == 0 && doVersionCheck)
			err = versionCheck(txn);
	} catch (...) {
		// Something bad happened. Close all the databases and destroy them.
		// JCM - Should there be another layer of exception handling around the close?
		close(0);
		throw; // rethrow the exception
	}

	if (err == 0) {
		open_ = true;
		ostringstream oss;
		oss << "container opened.";
		log(C_CONTAINER, L_DEBUG, oss);
	} else
		close(0);
	return err;
}

int Container::versionCheck(DbTxn *txn)
{
	int err= 0;
	//
	// Get/Put the version number into the configuration database.
	//
	unsigned int saved_version= 0;
	err= getVersion(txn, saved_version);
	if (err == DB_NOTFOUND) {
		err = putVersion(txn, version_);
	} else {
		//
		// In the future, we will need to upgrade,
		// but in the 1.0 series, we don't support this.
		//
		if (version_ != saved_version) {
			// special case: 10200 == 2
			if (version_ == 2 && 
			    saved_version == VERSION_120_COMPAT) {
				// reset to current
				err = putVersion(txn, version_);
				return err;
			}
			ostringstream s;
			s << "Container version '";
			s << saved_version;
			s << "' does not match dbxml library version '";
			s << version_;
			s << "'.";
			if((version_ > saved_version) || 
			   (saved_version == VERSION_11))
			{
				s << " Use XmlContainer::upgrade() to update the container version.";
			}
			else
			{
				s << " Use a more recent release of the dbxml library to open this container.";
			}
			throw XmlException(XmlException::VERSION_MISMATCH, s.str());
		}
	}
	return err;
}

int Container::getVersion(DbTxn *txn, unsigned int &version) const
{
	// Get version number from configuration database
	//
	DbtIn keydbt((void *)"version", strlen("version") + 1);
	DbtOut datadbt;
	int err = databaseConfiguration_->get(txn, &keydbt, &datadbt, 0);
	version = (err!=0 ? 0 : DbXml::stringToInt((char *)datadbt.get_data()));
	return err;
}

int Container::getVersions(DbTxn *txn, unsigned int &current_version, unsigned int &saved_version) const
{
	current_version = version_;
	return getVersion(txn, saved_version);
}

int Container::putVersion(DbTxn *txn, unsigned int version) 
{
	// Put version number into configuration database
	// key = "version\0", data = int (ASCII-encoded int)
	//
	DbtIn keydbt((void *)"version", strlen("version") + 1);
	DbtOut datadbt;
	string s(DbXml::toString(version));
	datadbt.set((void*)s.c_str(), s.length() + 1);
	return databaseConfiguration_->put(txn, &keydbt, &datadbt, 0);
}

bool Container::exists(DbTxn *txn)
{
	if(open_)
		return true;

	try {
		int err = open(txn, DB_RDONLY, 0, /*doVersionCheck=*/false);
		(void)close(0);
		return (err == 0);
	} catch (...) {
		// Ignore the exception - it probably means the file doesn't exist.
		(void)close(0);
	}
	
	return false;
}

int Container::close(u_int32_t flags)
{
	open_ = false;
	//
	// Close the databases
	//
	int err = 0;
	int firstErr = 0;
	std::vector<Database*>::iterator i;
	for (i = databases_.begin();i != databases_.end();++i) {
		// We try to close them all. If one of the close calls
		// fails we keep going until we've closed them all. We
		// return the error from the first close that failed.
		//
		try {
			err = (*i)->close(flags);
			if (err != 0 && firstErr == 0)
				firstErr = err;
		} catch (DbException &e) {
			ostringstream oss;
			oss << "Container::close() Failure closing database '" << (*i)->getDatabaseName() << "' because '" << e.what() << "'";
			log(C_CONTAINER, L_ERROR, oss);
			if (firstErr == 0)
				firstErr = e.get_errno();
		}
	}
	destructDatabases();
	if (err == 0) {
		ostringstream oss;
		oss << "container closed.";
		log(C_CONTAINER, L_DEBUG, oss);
	}
	return firstErr;
}

int Container::setIndexSpecification(DbTxn *txn, const IndexSpecification &newis)
{
	// The current indexing specification
	//
	IndexSpecification oldis;
	int err = oldis.read(*this, txn, /*lock=*/true);
	if (err == 0) {
		// The indexing strategies to add
		//
		IndexSpecification addis(newis);
		addis.disableIndex(oldis);
		//
		// The indexing strategies to delete
		//
		IndexSpecification delis(oldis);
		delis.disableIndex(newis);
		//
		// Mark the indexing strategies to delete
		//
		delis.set(Index::INDEXER_DELETE);
		//
		// Combine the indexing strageies to add and delete
		//
		IndexSpecification combinedis(addis);
		combinedis.enableIndex(delis);
		//
		// Check that there are new indexing strategies.
		//
		if (err == 0 && newis.isIndexed(Index::NONE, Index::NONE)) {
			err = reindex(txn, combinedis);
		}
		if (err == 0) {
			err = newis.write(*this, txn);
		}
		if (err == 0) {
			//
			// Log the indexing changes.
			//
			string uri, name, index;
			IndexSpecificationIterator i1(addis);
			while (i1.next(uri, name, index)) {
				ostringstream oss;
				oss << "Add '" << index << "' index for node '" << uri << "," << name << "'";
				log(C_CONTAINER, L_DEBUG, oss);
			}
			IndexSpecificationIterator i2(delis);
			while (i2.next(uri, name, index)) {
				ostringstream oss;
				oss << "Delete '" << index << "' index for node '" << uri << "," << name << "'";
				log(C_CONTAINER, L_DEBUG, oss);
			}
		}
	}
	return err;
}

int Container::getIndexSpecification(DbTxn *txn, IndexSpecification &index) const
{
	return index.read(*this, txn, /*lock=*/false);
}

int Container::addIndex(DbTxn *txn, const std::string &uri, const std::string &name, const std::string &index)
{
	int err;
	XmlIndexSpecification is;
	if ((err = getIndexSpecification(txn, is)) == 0) {
		is.addIndex(uri, name, index);
		err = setIndexSpecification(txn, (IndexSpecification&)is);
	}
	return err;
}

int Container::deleteIndex(DbTxn *txn, const std::string &uri, const std::string &name, const std::string &index)
{
	int err;
	XmlIndexSpecification is;
	if ((err = getIndexSpecification(txn, is)) == 0) {
		is.deleteIndex(uri, name, index);
		err = setIndexSpecification(txn, (IndexSpecification&)is);
	}
	return err;
}

int Container::replaceIndex(DbTxn *txn, const std::string &uri, const std::string &name, const std::string &index)
{
	int err;
	XmlIndexSpecification is;
	if ((err = getIndexSpecification(txn, is)) == 0) {
		is.replaceIndex(uri, name, index);
		err = setIndexSpecification(txn, (IndexSpecification&)is);
	}
	return err;
}

int Container::remove(DbTxn *txn)
{
	int err = environment_->dbremove(txn, getName().c_str(), 0, 0);
	if (err == 0) {
		ostringstream oss;
		oss << "container removed.";
		log(C_CONTAINER, L_DEBUG, oss);
	}
	return err;
}

int Container::rename(DbTxn *txn, const std::string &newName)
{
	int err = environment_->dbrename(txn, getName().c_str(), 0, newName.c_str(), 0);
	if (err == 0) {
		ostringstream oss;
		oss << "container renamed from '" << name_ << "' to '" << newName << "'.";
		log(C_CONTAINER, L_DEBUG, oss);
	}
	return err;
}

void Container::getDumpedDatabases(std::vector<Database *> &toDump)
{
	toDump.push_back(databaseConfiguration_.get());
	toDump.push_back(databaseDictionaryPrimary_.get());
	toDump.push_back(databaseDictionarySecondary_.get());
	toDump.push_back(databaseDocumentPrimary_.get());
	toDump.push_back(databaseDocumentSecondary_.get());
}

void Container::writeDumpHeader(std::ostream *out, Database *db)
{
	(*out) << "xml_database=" << db->getDatabaseName() << endl;
}

int Container::dump(std::ostream *out, u_int32_t flags)
{
	int err = 0;
	std::vector<Database*> toDump;

	constructDatabases();
	getDumpedDatabases(toDump);

	for (std::vector<Database*>::iterator i = toDump.begin(); err == 0 && i != toDump.end(); ++i) {
		Database *db = *i;
		writeDumpHeader(out, db);
		err = db->dump(out, flags);
	}

	destructDatabases();
	return (err);
}

int Container::load(std::istream *in, unsigned long *lineno, u_int32_t flags)
{
	int ret = 0, t_ret;
	DbTxn *txn = NULL;

	constructDatabases();

	std::vector<Database*> toLoad;
	getDumpedDatabases(toLoad);

	for (std::vector<Database*>::iterator i = toLoad.begin(); ret == 0 && i != toLoad.end(); ++i) {
		Database *db = *i;
		char keyname[64], dbname[256];

		if (!in->get(keyname, sizeof keyname, '=') ||
		    strcmp(keyname, "xml_database") != 0 ||
		    in->get() != '=' ||
		    !in->get(dbname, sizeof dbname) ||
		    in->get() != '\n') {
			ret = EINVAL;
			break;
		}

		if (db->getDatabaseName() != dbname) {
			ostringstream oss;
			oss << "Container::load() invalid database name (got" << dbname << ", expected " << db->getDatabaseName() << ")";
			log(C_CONTAINER, L_ERROR, oss);
			ret = EINVAL;
			break;
		}

		ret = db->load(in, lineno, flags);
	}

	/* Now destroy databases and reopen so that the configuration is read. */
	destructDatabases();

	if (ret == 0 && (ret = open(txn, DB_CREATE, 0666, /*doVersionCheck=*/true)) == 0) {
		IndexSpecification is;
		ret = getIndexSpecification(0, is);
		ret = reindex(txn, is);
		if ((t_ret = close(0)) != 0 && ret == 0)
			ret = t_ret;
	}

	return (ret);
}

int Container::verify(std::ostream *out, u_int32_t flags)
{
	int ret = 0;

	constructDatabases();
	std::vector<Database*> toSalvage;
	getDumpedDatabases(toSalvage);

	try {
		/* Verify the databases we may be salvaging. */
		int t_ret = 0;
		flags |= DB_NOORDERCHK;
		std::vector<Database*>::iterator i;
		for (i = toSalvage.begin(); i != toSalvage.end(); ++i) {
			Database *db = *i;
			if (flags & DB_SALVAGE)
				writeDumpHeader(out, db);
			if ((t_ret = db->verify(out, flags)) != 0 && ret == 0)
				ret = t_ret;
		}

		/* Now verify the remaining databases without salvage. */
		flags &= ~DB_SALVAGE;
		for (i = databases_.begin(); i != databases_.end(); ++i) {
			Database *db = *i;
			if (find(toSalvage.begin(), toSalvage.end(), db) != toSalvage.end())
				continue;
			if ((t_ret = db->verify(NULL, flags)) != 0 && ret == 0)
				ret = t_ret;
		}
	} catch (...) {
		destructDatabases();
		throw; // rethrow the exception
	}

	destructDatabases();

	return ret;
}

int Container::reindex(DbTxn *txn, const IndexSpecification &is)
{
	int ret = 0;
	db_recno_t id_recno;
	DbtIn key((void *)&id_recno, sizeof (id_recno));
	DbtOut value;
	XmlDocument document;
	Indexer indexer(this);
	OperationContext context(txn);

	Cursor cursor(environment_, databaseDocumentPrimary_->getDb(), txn);

	while ((ret = cursor.get(&key, &value, DB_NEXT)) == 0) {
		indexer.reset();
		if ((ret = getDocument(context, id_recno, document, 0)) == 0 &&
		    (ret = indexer.indexDocument(txn, is, (Document&)document)) == 0)
			ret = indexer.addOrDeleteKeys(context, /*add=*/true);
	}

	if (ret == DB_NOTFOUND)
		ret = 0;

	return (ret);
}

int Container::defineName(DbTxn *txn, const Name &name, ID &id) const
{
	// Primary { id -> uri, prefix, name }
	// Secondary { uri:name -> id }
	//
	id.reset();
	DbtOut data;
	name.setDbtFromThis_PrimaryValue(data, databaseDictionaryPrimary_->swap());
	int err = databaseDictionaryPrimary_->putPrimary(txn, id, &data, /*no flags*/0);
	if (err == 0) {
		DbtOut key;
		id.setDbtFromThis(key, databaseDictionaryPrimary_->swap());
		name.setDbtFromThis_SecondaryKey(data, databaseDictionaryPrimary_->swap());
		err = databaseDictionarySecondary_->put(txn, &data, &key, /*no flags*/0);
		if (err == 0) {
			ostringstream oss;
			oss << "Define new name " << id << " -> " << name;
			log(C_DICTIONARY, L_DEBUG, oss);
		}
	}
	return err;
}

int Container::defineName(DbTxn *txn, const char *uriname, ID &id) const // define from uri:name
{
	Name name(uriname);
	return defineName(txn, name, id);
}

int Container::lookupName(DbTxn *txn, const ID &id, Name &name) const // lookup by id
{
	DbtOut key;
	DbtOut data;
	return lookupName(txn, key, data, id, name);
}

int Container::lookupName(DbTxn *txn, DbtOut &key, DbtOut &data, const ID &id, Name &name) const // lookup by id
{
	int err = 0;
	if (id == nidId_) {
		name = Name::dbxml_colon_id;
	} else if (id == nidName_) {
		name = Name::dbxml_colon_name;
	} else if (id == nidContent_) {
		name = Name::dbxml_colon_content;
	} else {
		id.setDbtFromThis(key, databaseDictionaryPrimary_->swap());
		err = databaseDictionaryPrimary_->get(txn, &key, &data, /*no flags*/0);
		if (err == 0) {
			name.setThisFromDbt(data, databaseDictionaryPrimary_->swap());
		} else {
			name.reset();
		}
	}
	return err;
}

int Container::lookupName(DbTxn *txn, const Name &name, ID &id, bool define) const // lookup by uri:name
{
	DbtOut key;
	DbtOut data;
	return lookupName(txn, key, data, name, id, define);
}

int Container::lookupName(DbTxn *txn, DbtOut &key, DbtOut &data, const Name &name, ID &id, bool define) const // lookup by uri:name
{
	int err = 0;
	if (name == Name::dbxml_colon_id) {
		id = nidId_;
	} else if (name == Name::dbxml_colon_name) {
		id = nidName_;
	} else if (name == Name::dbxml_colon_content) {
		id = nidContent_;
	}
	if (id == 0) {
		name.setDbtFromThis_SecondaryKey(key, databaseDictionarySecondary_->swap());
		err = databaseDictionarySecondary_->get(txn, &key, &data, /*no flags*/0);
		if (err == 0) {
			id.setThisFromDbt(data, databaseDictionarySecondary_->swap());
		} else if (err == DB_NOTFOUND && define) {
			err = defineName(txn, name, id); // define from uri:name
		} else {
			id.reset();
		}
	}
	return err;
}

int Container::lookupName(DbTxn *txn, const char *uriname, ID &id, bool define) const // lookup by uri:name
{
	DbtIn key((void*)uriname, strlen(uriname));
	u_int32_t idt = 0;
	DbtIn data(&idt, sizeof(idt));
	int err = databaseDictionarySecondary_->get(txn, &key, &data, /*no flags*/0);
	if (err == 0) {
		id.setThisFromDbt(data, databaseDictionarySecondary_->swap());
	} else if (err == DB_NOTFOUND && define) {
		err = defineName(txn, uriname, id); // define from uri:name
	} else {
		id.reset();
	}
	return err;
}

int Container::lookupName(DbTxn *txn, const std::string &uriname, ID &id) const // lookup by uri:name
{
	return const_cast<Container*>(this)->lookupName(txn, uriname.c_str(), id, /*define=*/false);
}

int Container::getConfigurationItem(DbTxn *txn, const char *key, size_t keyLength, Buffer &b, bool lock) const
{
	// lock means that we acquire a write lock for reading the
	// configuration item. We use this when we're going to change
	// an item and we want to block all the readers and writers
	// until we're done.
	//
	Cursor cursor(environment_, databaseConfiguration_->getDb(), txn);
	DbtIn k((void*)key, keyLength);
	DbtOut v;
	int err = cursor.get(&k, &v, DB_SET | (txn && lock ? DB_RMW : 0));
	if (err == 0) {
		b.write(v.get_data(), v.get_size());
	}
	return err;
}

int Container::putConfigurationItem(DbTxn *txn, const char *key, const Buffer &b)
{
	DbtIn k((void*)key, strlen(key) + 1);
	DbtIn v(b.getBuffer(), b.getOccupancy());
	return databaseConfiguration_->put(txn, &k, &v, 0);
}

// We assume that key has the correct endianness.
int Container::updateStatistics(OperationContext &context, Syntax::Type type, DbtIn &key, const KeyStatistics &statistics)
{
	DbTxn *txn = context.txn();
	bool swap = databaseStatistics_[type]->swap();
	Cursor cursor(environment_, databaseStatistics_[type]->getDb(), txn);
	int err = cursor.get(&key, &context.data(), DB_SET | (txn ? DB_RMW : 0)); // could throw on error
	if (err == 0) {
		KeyStatistics existing;
		existing.setThisFromDbt(context.data(), swap);
		existing.add(statistics);
		existing.setDbtFromThis(context.data(), swap);
		err = cursor.put(&key, &context.data(), DB_CURRENT); // could throw on error
	} else if (err == DB_NOTFOUND) {
		statistics.setDbtFromThis(context.data(), swap);
		err = cursor.put(&key, &context.data(), DB_KEYFIRST); // could throw on error
	}
	return err;
}

void Container::getStatistics(OperationContext &context, const Key &key, KeyStatistics &es) const
{
	Syntax::Type type = key.getSyntaxType();
	const Syntax *syntax = key.getSyntax();
	key.setDbtFromThis(context.key(), databaseStatistics_[type]->swap());
 	context.key().set_size(syntax->structureKeyLength(key.getIndex(), context.key().get_size())); // trim the value off
	int err = databaseStatistics_[type]->get(context.txn(), &context.key(), &context.data(), /*no flags*/0); // could throw on error
	if (err == 0) {
		es.setThisFromDbt(context.data(), databaseStatistics_[type]->swap());
	} else {
		es.zero();
	}
}

int Container::addDocument(DbTxn *txn, Document &document, ID &id, UpdateContext &context, u_int32_t flags)
{
	OperationContext &oc = context.getOperationContext(txn);
	Indexer &indexer = context.getIndexer();
	IndexSpecification &index = context.getIndexSpecification(txn);
	int err = indexer.indexDocument(txn, index, document);
	if (err == 0) {
		if (id == 0) {
			// putPrimary assigns the document ID
			err = databaseDocumentPrimary_->putPrimary(txn, id, &oc.data(), flags); // could throw on error
		} else {
			// reusing an existing document ID
			id.setDbtFromThis(oc.key(), swap());
			err = databaseDocumentPrimary_->put(txn, &oc.key(), &oc.data(), flags); // could throw on error
		}
		if (err == 0) {
			document.setID(id);
			if(resolver_==0) {
				//
				// Write each of the meta-data items to the document secondary database.
				// (Note that the content is also a meta-data item.)
				//
				MetaData::const_iterator i;
				for (i = document.metaDataBegin();err == 0 && i != document.metaDataEnd();++i) {
					ID nid;
					err = lookupName(txn, oc.key(), oc.data(), (*i)->getName(), nid, /*define=*/true);
					DbtIn value;
					MetaDatum::setKeyDbt(id, nid, (*i)->getType(), oc.key(), swap());
					(*i)->setValueDbtFromThis(value, swap());
					err = databaseDocumentSecondary_->put(txn, &oc.key(), &value, flags); // could throw on error
				}
			}
			if (err == 0) {
				indexer.addOrDeleteKeys(oc, /*add=*/true);
			}
		}
	}
	return err;
}

int Container::getDocument(DbTxn *txn, const ID &id, XmlDocument &document, u_int32_t flags) const
{
	OperationContext context(txn);
	return getDocument(context, id, document, flags);
}

int Container::getDocument(OperationContext &context, const ID &did, XmlDocument &document, u_int32_t flags) const
{
	int err = 0;
	if (did == 0) {
		err = getMetaDocument(context.txn(), document);
	} else {
		if(resolver_) {
			err= resolver_->getDocument(context.txn(), did.raw(), document);
			((Document&)document).setID(did);
		} else {
			//
			// Read each of the meta-data items from the document secondary database.
			// (Note that the content is also a meta-data item.)
			//
			Cursor cursor(environment_, databaseDocumentSecondary_->getDb(), context.txn());
			flags = DB_SET_RANGE;
			bool done = false;
			while (!done) {
				did.setDbtFromThis(context.key(), databaseDocumentSecondary_->swap());
				DbtIn none;
				none.set_flags(DB_DBT_PARTIAL); // Don't pull back the data.
				err = cursor.get(&context.key(), &none, flags);
				if (err == 0) {
					ID id1, id2;
					XmlValue::Type type;
					MetaDatum::decodeKeyDbt(context.key(), databaseDocumentSecondary_->swap(), id1, id2, type);
					if (did == id1) {
						DbtOut *data = new DbtOut();
						err = cursor.get(&context.key(), data, DB_SET);
						err = ((Document&)document).setThisFromDbt(context, *this, context.key(), &data, databaseDocumentSecondary_->swap());
						delete data; // If not consumed by setThisFromDbt
						flags = DB_NEXT;
					} else {
						err = 0;
						done = true;
					}
				} else if (err == DB_NOTFOUND) {
					err = 0;
					done = true;
				} else {
					done = true;
				}
			}
			if (document.getID() == 0) {
				err = DB_NOTFOUND;
			}
		}
	}
	return err;
}

int Container::getMetaDataItem(OperationContext &context, ID did, ID nid, XmlValue &value) const
{
	Cursor cursor(environment_, databaseDocumentSecondary_->getDb(), context.txn());
	MetaDatum::setKeyDbt(did, nid, XmlValue::NONE, context.key(), databaseDocumentSecondary_->swap());
	int err = cursor.get(&context.key(), &context.data(), DB_SET_RANGE);
	if (err == 0) {
		ID id1, id2;
		XmlValue::Type type;
		MetaDatum::decodeKeyDbt(context.key(), databaseDocumentSecondary_->swap(), id1, id2, type);
		value = XmlValue(type, context.data());
	} else if (err == DB_NOTFOUND) {
		err = 0;
	}
	return err;
}

int Container::deleteDocument(DbTxn *txn, ID id, UpdateContext &context, u_int32_t flags)
{
	XmlDocument document;
	int err = Container::getDocument(txn, id, document, flags);
	if (err == 0) {
		err = deleteDocument(txn, document, context, flags);
	}
	return err;
}

int Container::deleteDocument(DbTxn *txn, Document &document, UpdateContext &context, u_int32_t flags)
{
	int err = 0;
	//
	// We may have been given an empty document, or a document that has
	// already been deleted, or a document with a mangled ID. Db::del
	// returns EINVAL if it can't delete a key/data pair. We change this
	// to DB_NOTFOUND, which makes more sense.
	//
	OperationContext &oc = context.getOperationContext(txn);
	ID id = document.getID();
	try {
		err = databaseDocumentPrimary_->deletePrimary(oc.txn(), id, flags); // throw on error
	} catch (DbException &e) {
		err = e.get_errno();
		if (err == EINVAL) {
			err = DB_NOTFOUND;
		}
	}
	if (err == 0) {
		if(resolver_==0)
		{
			// Delete each of the meta-data items from the document secondary database.
			// (Note that the content is also a meta-data item.)
			//
			oc.data().set_flags(DB_DBT_PARTIAL); // Don't pull back the data.
			int err = 0;
			Cursor cursor(environment_, databaseDocumentSecondary_->getDb(), oc.txn());
			bool done = false;
			while (!done) {
				id.setDbtFromThis(oc.key(), databaseDocumentSecondary_->swap());
				err = cursor.get(&oc.key(), &oc.data(), DB_SET_RANGE);
				if (err == 0) {
					ID id2;
					id2.setThisFromDbt(oc.key(), databaseDocumentSecondary_->swap());
					if (id == id2) {
						cursor.del( /*flags=*/0);
					} else {
						err = 0;
						done = true;
					}
				} else if (err == DB_NOTFOUND) {
					err = 0;
					done = true;
				} else {
					done = true;
				}
			}
			oc.data().set_flags(0); // Reset the context dbt
		}
		if (err == 0) {
			Indexer &indexer = context.getIndexer();
			IndexSpecification &index = context.getIndexSpecification(oc.txn());
			err = indexer.indexDocument(oc.txn(), index, document);
			if (err == 0) {
				err = indexer.addOrDeleteKeys(oc, /*add=*/false);
			}
		}
	}
	return err;
}

int Container::updateDocument(DbTxn *txn, Document &document, UpdateContext &context)
{
	int err = deleteDocument(txn, document.getID(), context, /*flags=*/0);
	if (err == 0) {
		ID id(document.getID());
		err = addDocument(txn, document, id, context, /*flags=*/0);
	}
	return err;
}

PrimaryDatabase *Container::getPrimaryDatabase() const
{
	return databaseDocumentPrimary_.get();
}

SecondaryDatabase *Container::getIndex(Syntax::Type type) const
{
	return databaseIndexes_[type].get();
}

int Container::parseXPathExpression(DbTxn *txn, const std::string &xpath, XmlQueryContext &context, QueryExpression &expression) const
{
	QueryPlan::SharedPtr qp;
	try {
		//
		// Parse the XPath query.
		//
		HighResTimer t;
		t.start();
		istringstream oss(xpath);
		XPathLexer lexer(oss);
		XPathParser parser(lexer);
		ASTFactory ast_factory;
		parser.initializeASTFactory(ast_factory);
		parser.setASTFactory(&ast_factory);
		parser.xpath(context);
		t.stop();
		if (parser.getAST() != NULL) {
			if (isLogEnabled(C_QUERY, L_DEBUG)) {
				logProgress("AST", t, parser.getAST()->toStringList());
			}
			//
			// Generate a Query Plan
			//
			t.start();
			createQueryPlan(parser, xpath, context, qp); // throws XmlException
			t.stop();
			logProgressQP("RQP", t, qp.get());
			if (qp) {
				bool debugging = false;
				string debugVariable;
				XmlQueryContext::EvaluationType et = XmlQueryContext::Eager;
				XmlQueryContext::ReturnType rt = XmlQueryContext::ResultDocuments;
				QueryContext &qc = context;
				et = qc.getEvaluationType();
				rt = qc.getReturnType();
				XmlValue v;
				if (context.getVariableValue(metaDataNamespace_prefix_debug, v)) {
					debugging = true;
					debugVariable = v.asString(&context);
					qc.setVariableValue(debugVariable + "._rqp", qp->toString());
				}
				//
				// Optimize the Query Plan
				//
				t.start();
				IndexSpecification is;
				getIndexSpecification(txn, is);
				qp->optimize(txn, *this, context, is, 0);
				t.stop();
				logProgressQP("OQP", t, qp.get());
				if (debugging) {
					context.setVariableValue(debugVariable + "._oqp", qp->toString());
				}
			} else {
				throw XmlException(XmlException::INTERNAL_ERROR, "Query Plan Generator was unable to generate a query plan for the XPath expression. XPath= '" + xpath + "'");
			}
		} else {
			throw XmlException(XmlException::XPATH_PARSER_ERROR, "XPath parser was unable to parse XPath expression '" + xpath + "'.");
		}
	} catch (ANTLRException &e) {
		throw XmlException(XmlException::XPATH_PARSER_ERROR, "XPath parser was unable to parse XPath expression '" + xpath + "'. " + e.toString());
	}
	expression.set(xpath, qp);
	return 0;
}

void Container::createQueryPlan(XPathParser &parser, const string &xpath, XmlQueryContext &context, QueryPlan::SharedPtr &qp) const
{
	try {
		QueryContext &qc = context;
		qp = qc.getXPathSelectionTreeParser().xpath(parser.getAST(), context);
		if (qp) {
			if (!QueryPlan::is(qp, QueryPlan::CONSTANT)) {
				// By default we search for result documents and return those documents.
				//
				XmlQueryContext::ReturnType rt = XmlQueryContext::ResultDocuments;
				XmlQueryContext::EvaluationType et = XmlQueryContext::Eager;
				QueryContext &qc = context;
				rt = qc.getReturnType();
				et = qc.getEvaluationType();
				switch (rt) {
				case XmlQueryContext::CandidateDocuments:
					// No RF needed for fetching Candidate Documents.
					// RA fetches the Candidate Documents.
					qp.reset(new RecordAccessQP(qp));
					break;
				case XmlQueryContext::ResultDocuments:
					// RA fetches the Candidate Documents.
					qp.reset(new RecordAccessQP(qp));
					// RF filters the Candidate Documents.
					qp.reset(new RestrictionFilterAndOrProjectQP(xpath, qp, /*project=*/false));
					// RA fetches the Result Documents.
					qp.reset(new RecordAccessQP(qp));
					break;
				case XmlQueryContext::ResultDocumentsAndValues:
				case XmlQueryContext::ResultValues: {
						//
						// jcm - This code is a small foray into us doing the
						// projection instead of Pathan...
						//
						XPathProjectionTreeParser xptp;
						QueryPlan::SharedPtr pqp = xptp.xpath(parser.getAST());
						if (pqp) {
							// RA fetches the Candidate Documents.
							qp.reset(new RecordAccessQP(qp));
							// RF filters the Candidate Documents.
							qp.reset(new RestrictionFilterAndOrProjectQP(xpath, qp, /*project=*/false));
							// MDA fetches the Metadata Attributes.
							((QueryPlanNode*)pqp.get())->setLeft(qp);
							qp = pqp;
						} else {
							// RA fetches the Candidate Documents.
							qp.reset(new RecordAccessQP(qp));
							// RFP filters the Candidate Documents, and performs a projection.
							qp.reset(new RestrictionFilterAndOrProjectQP(xpath, qp, /*project=*/true));
						}
					}
					break;
				}
			} else {
				// The result is a value...
			}
			qp.reset(new RootQP(qp));
		} else {
			throw XmlException(XmlException::INTERNAL_ERROR, "Query Plan Generator couldn't transform XPath expression into a Raw Query Plan. XPath= '" + xpath + "'.", __FILE__, __LINE__);
		}
	} catch (ANTLRException &e) {
		throw XmlException(XmlException::XPATH_PARSER_ERROR, "Query Plan Generator couldn't transform XPath expression into a Raw Query Plan. XPath= '" + xpath + "'. " + e.toString());
	}
}

int Container::queryWithXPath(DbTxn *txn, const string &xpath, Results *result, XmlQueryContext &context, u_int32_t flags) const // throws XmlException
{
	QueryExpression expression(context);
	int err = parseXPathExpression(txn, xpath, context, expression);
	if (err == 0) {
		err = queryWithXPath(txn, expression, result, flags);
	}
	return err;
}

int Container::queryWithXPath(DbTxn *txn, QueryExpression &expression, Results *result, u_int32_t flags) const
{
	UNUSED(flags);

	int err = 0;
	QueryContext &context = expression.getContext();
	QueryPlan::SharedPtr qp = expression.getQueryPlan();
	bool debugging = false;
	string debugVariable;
	XmlQueryContext::EvaluationType et = XmlQueryContext::Eager;
	XmlQueryContext::ReturnType rt = XmlQueryContext::ResultDocuments;
	et = context.getEvaluationType();
	rt = context.getReturnType();
	XmlValue v;
	static std::string s(metaDataNamespace_prefix_debug);
	if (context.getVariableValue(s, v)) {
		debugVariable = ((Value*)v)->asString(&context);
		debugging = true;
	}
	//
	// Execute the Query Plan
	//
	context.setTransaction(txn);
	OperationContext &oc = context.getOperationContext();
	QueryExecutionContext qec(*this, context, debugging);
	IDS::SharedPtr pids;
	err = qp->execute(oc, qec, pids);
	if (debugging) {
		context.setVariableValue(debugVariable + "._eqp", qec.getExecutionPath());
	}
	//
	// Create the results
	//
	err = result->set(const_cast<Container&>(*this), qp);
	if (err == 0) {
		if (debugging && et == XmlQueryContext::Eager) {
			//
			// The query exection context is asking for the debugging variables to be set.
			//
			if (rt == XmlQueryContext::ResultDocuments) {
				XmlValue value;
				IDS ids;
				while ((err = result->next(value)) == 0 && !value.isNull()) {
					if (value.isDocument(0)) {
						XmlDocument d(value.asDocument(0));
						ids.push_back(d.getID());
					}
				}
				context.setVariableValue(debugVariable, "");
				context.setVariableValue(debugVariable + "._rids", ids.toString());
				context.setVariableValue(debugVariable + "._nrids", (double)ids.size());
				result->reset();
			}
			if (rt == XmlQueryContext::ResultValues) {
				int nestederr;
				int n = 0;
				string r("");
				XmlValue value;
				while ((nestederr = result->next(value)) == 0 && !value.isNull()) {
					r += value.asString(0);
					r += "\n"; // separate each result.
					n++;
				}
				context.setVariableValue(debugVariable, r);
				context.setVariableValue(debugVariable + "._rids", "");
				context.setVariableValue(debugVariable + "._nrids", (double)n);
				result->reset();
			}
		}
	}
	return err;
}

// Algorithm *requires* that results are grouped by document, so that only one
// re-write is performed for each document involved.
int Container::modifyDocument(DbTxn *txn, const Modify &modify, XmlUpdateContext *uContext, u_int32_t flags)
{
	int err= 0;
	bool newExpression = false;
	bool newContext = false;
	QueryExpression *expr;   // must get initialized
	XmlQueryContext *qContext;  // must get initialized
	XmlQueryContext::ReturnType origRetType;
	XmlQueryContext::EvaluationType origEvalType;
	modify.clearOps();
	// need to get/create query expression, if not present
	XmlQueryExpression *expression = modify.getExpression();
	if (!expression) {
		qContext = modify.getContext();
		if (!qContext) {
			newContext = true;
			qContext = new XmlQueryContext(XmlQueryContext::ResultDocumentsAndValues,
						       XmlQueryContext::Lazy);
		}
		newExpression = true;
		expr = new QueryExpression(*qContext);
		err = parseXPathExpression(txn, *modify.getXPath(), *qContext, *expr);
	} else {
		expr = &((QueryExpression&)*expression);
		qContext = &expression->getQueryContext();
	}
	// save/restore return type in QueryContext if user provided QueryContext
	if (!err && !newContext) {
		origRetType = qContext->getReturnType();
		origEvalType = qContext->getEvaluationType();
		qContext->setReturnType(XmlQueryContext::ResultDocumentsAndValues);
		qContext->setEvaluationType(XmlQueryContext::Lazy);
	}

	// now the real work can proceed
	try {
		Results results(*qContext, txn);
		err = queryWithXPath(txn, *expr, &results, 0);
		if(!err) {
			// gmf: this code looks a lot like Document::xupdate, but
			// new content is not pre-created and reused, as it is
			// in the Document method; it's done for
			// each result.  Could probably do better than this.
			XmlDocument curXmlDoc;
			XmlValue curValue;
			results.next(curValue);
			if (!curValue.isNull())
				curXmlDoc = curValue.asDocument();
			while (!err && !curValue.isNull()) {
				((Document&)curXmlDoc).updateDocNode(curValue, modify, NULL);
				results.next(curValue);
				if (curValue.isNull() || (curXmlDoc != curValue.asDocument())) {
					// moving on to a new document or done; rewrite current doc
					if (qContext->getWithMetaData())
						((Document&)curXmlDoc).domMetaData(true);
					((Document&)curXmlDoc).setContentFromDOM(modify.getNewEncoding());
					// replace rewritten doc in container
					if (uContext) {
						err = updateDocument(txn, (Document&)curXmlDoc,*uContext);
					} else {
						UpdateContext uc(this); // temp
						err = updateDocument(txn, (Document&)curXmlDoc, uc);
					}
					if (!err && !curValue.isNull())
						curXmlDoc = curValue.asDocument();
				}
			}
		}
	} catch (...) {
		if (!newContext) {
			qContext->setReturnType(origRetType);
			qContext->setEvaluationType(origEvalType);
		} else {
			delete qContext;
		}
		if (newExpression)
			delete expr;
		throw; // simple re-throw
	}
	if (!newContext) {
		qContext->setReturnType(origRetType);
		qContext->setEvaluationType(origEvalType);
	} else {
		delete qContext;
	}
	if (newExpression)
		delete expr;
	return err;
}

void Container::logProgressQP(const char *tag, const HighResTimer &t, const QueryPlan *qp) const
{
	if (isLogEnabled(C_QUERY, L_DEBUG)) {
		static bool outputLegend = true;
		if (outputLegend) {
			outputLegend = false;
			ostringstream oss;
			oss
			<< "\n"
			<< "\t Legend for the Query Plan debug output\n"
			<< "\n"
			<< "\t AST  - Abstract Syntax Tree of the query\n"
			<< "\t RQP  - Raw Query Plan before any optimizations\n"
			<< "\t PnRm - Optimization, Phase n Rule m\n"
			<< "\t OQP  - Optimized Query Plan after optimizations\n"
			<< "\n"
			<< "\t SS   - Sequential Scan\n"
			<< "\t RA   - Record Access\n"
			<< "\t RI   - Restrict Index\n"
			<< "\t RF   - Record Filter\n"
			<< "\t n    - Intersection\n"
			<< "\t u    - Union\n"
			<< "\n"
			<< "\t ef   - estimate function cost (in pages read)\n"
			<< "\t efd  - estimate function and decendant cost (in pages read)\n"
			<< "\t af   - actual function time (in ms)\n"
			<< "\t afd  - actual function and decendants time (in ms)\n"
			<< "\n";
			log(C_QUERY, L_DEBUG, oss);
		}
		logProgress(tag, t, (qp == 0 ? string("null") : qp->toString()));
	}
}

void Container::logProgress(const char *tag, const HighResTimer &t, const string &description) const
{
	long seconds, microseconds;
	t.duration(&seconds, &microseconds);
	double duration = ((double)seconds * 1000) + ((double)microseconds / 1000.0);
	ostringstream oss;
	oss << tag << ": afd=" << duration << " : " << description.c_str();
	log(C_QUERY, L_DEBUG, oss);
}

void Container::log(ImplLogCategory c, ImplLogLevel l, const ostringstream &s) const
{
	DbXml::log(environment_, c, l, getName().c_str(), s.str().c_str());
}

int Container::getMetaDocument(DbTxn *txn, XmlDocument &document) const
{
	int err = 0;
	std::string s;
	s += "<statistics>\n";
	unsigned char prefix = 0;
	for (unsigned int i = 0;i < databaseStatistics_.size();++i) {
		SecondaryDatabase *database = databaseStatistics_[i].get();
		Cursor cursor(environment_, database->getDb(), txn);
		u_int32_t flags = DB_FIRST;
		bool done = false;
		while (!done) {
			DbtOut kDbt;
			DbtOut ksDbt;
			err = cursor.get(&kDbt, &ksDbt, flags);
			if (err == 0) {
				Key k;
				KeyStatistics ks;
				k.setThisFromDbt(kDbt, database->swap());
				ks.setThisFromDbt(ksDbt, database->swap());
				//
				unsigned char newprefix = *((const unsigned char *)kDbt.get_data());
				if (newprefix != prefix) {
					if (prefix != 0) {
						s += "  </index>\n";
					}
					Index index(k.getIndex());
					index.set(i); // set the syntax... as the key prefix doesn't contain this.
					s += "  <index name='";
					s += index.asString();
					s += "'>\n";
					prefix = newprefix;
				}
				//
				s += "    <ks ";
				s += k.asString_XML(txn, *this);
				s += ks.asString();
				s += "/>\n";
				//
				flags = DB_NEXT;
			} else if (err == DB_NOTFOUND) {
				err = 0;
				done = true;
			} else {
				done = true;
			}
		}
	}
	if (prefix != 0) {
		s += "  </index>\n";
	}
	s += "</statistics>\n";
	document.setContent(s);
	return err;
}

int Container::upgrade(u_int32_t flags)
{
	int err= 0;
	//
	// Upgrade Berkeley DB
	//
	{
		Db db(environment_, 0);
		err= db.upgrade(name_.c_str(), flags);
	}
	//
	// Upgrade Berkeley DB XML
	//
	if(err==0) {
		err= open(0, 0, 0, /*doVersionCheck=*/false);
		if(err==0) {
			unsigned int current_version, saved_version;
			getVersions(0, current_version, saved_version);
			//
			// 1.1.0 => 1.2.0  Substring indexing changed.
			// 1.2.1 changed version from 10200 to 2.  This
			// is handled in versionCheck(), mostly.
			//
			if(saved_version==10100 && current_version==2) {
				err= upgrade_010100_010200();
				if(err==0) {
					err= putVersion(0, current_version);
				}
			}
			close(0);
		}
	}
	return err;
}

int Container::upgrade_010100_010200()
{
	// Check if there are any substring indexes enabled.
	IndexSpecification is;
	int err = is.read(*this, 0, /*lock=*/false);
	if(err==0 && is.isIndexed(Index::KEY_SUBSTRING, Index::KEY_MASK)) {
		err= upgrade_010100_010200_delete_substring_keys(databaseIndexes_[Syntax::STRING]);
		if (err == 0)
		{
			err= upgrade_010100_010200_delete_substring_keys(databaseStatistics_[Syntax::STRING]);
			if (err == 0) {
				// Walk over all the documents reindexing just the substring indexes.
				IndexSpecification is2(is);
				is.disableIndex(Index::KEY_SUBSTRING, Index::KEY_MASK);
				is2.disableIndex(is);
				err = reindex(0, is2);
			}
		}
	}
	return err;
}

int Container::upgrade_010100_010200_delete_substring_keys(SecondaryDatabase::Ptr &database)
{
	//
	// Walk over all the keys deleting the substring ones.
	//
	int err= 0;
	Cursor cursor(environment_, database->getDb(), 0);
	Index index;
	u_int32_t flags = DB_FIRST;
	DbtOut k, d;
	d.set_flags(DB_DBT_PARTIAL); // Don't pull back the data.
	bool done = false;
	while (!done) {
		err = cursor.get(&k, &d, flags);
		if (err == 0) {
			unsigned char *p= (unsigned char *)k.get_data();
			index.setFromPrefix(*p);
			if(index.equalsMask(Index::KEY_SUBSTRING, Index::KEY_MASK))
			{
				cursor.del(/*flags=*/0);
			}
			flags = DB_NEXT;
		} else if (err == DB_NOTFOUND) {
			err = 0;
			done = true;
		} else {
			done = true;
		}
	}
	return err;
}
