added code to purge obsolete messages when a compound document (esp. mbox) is shortened and a partial update is performed (no general purge). Else the orphan docs remained in the index potentially forever (needed actual reindex of the file by a full pass to go away)
This commit is contained in:
Jean-Francois Dockes
+101
-33
@@ -149,23 +149,34 @@ void *DbUpdWorker(void* vdbp)
|
||||
WorkQueue<DbUpdTask*> *tqp = &(ndbp->m_wqueue);
|
||||
DebugLog::getdbl()->setloglevel(ndbp->m_loglevel);
|
||||
|
||||
DbUpdTask *tsk;
|
||||
DbUpdTask *tsk = 0;
|
||||
for (;;) {
|
||||
size_t qsz;
|
||||
size_t qsz = -1;
|
||||
if (!tqp->take(&tsk, &qsz)) {
|
||||
tqp->workerExit();
|
||||
return (void*)1;
|
||||
}
|
||||
LOGDEB(("DbUpdWorker: got task, ql %d\n", int(qsz)));
|
||||
bool status;
|
||||
if (tsk->txtlen == (size_t)-1) {
|
||||
status = ndbp->m_rcldb->purgeFileWrite(tsk->udi, tsk->uniterm);
|
||||
} else {
|
||||
bool status = false;
|
||||
switch (tsk->op) {
|
||||
case DbUpdTask::AddOrUpdate:
|
||||
LOGDEB(("DbUpdWorker: got add/update task, ql %d\n", int(qsz)));
|
||||
status = ndbp->addOrUpdateWrite(tsk->udi, tsk->uniterm,
|
||||
tsk->doc, tsk->txtlen);
|
||||
break;
|
||||
case DbUpdTask::Delete:
|
||||
LOGDEB(("DbUpdWorker: got delete task, ql %d\n", int(qsz)));
|
||||
status = ndbp->purgeFileWrite(false, tsk->udi, tsk->uniterm);
|
||||
break;
|
||||
case DbUpdTask::PurgeOrphans:
|
||||
LOGDEB(("DbUpdWorker: got orphans purge task, ql %d\n", int(qsz)));
|
||||
status = ndbp->purgeFileWrite(true, tsk->udi, tsk->uniterm);
|
||||
break;
|
||||
default:
|
||||
LOGERR(("DbUpdWorker: unknown op %d !!\n", tsk->op));
|
||||
break;
|
||||
}
|
||||
if (!status) {
|
||||
LOGERR(("DbUpdWorker: addOrUpdateWrite failed\n"));
|
||||
LOGERR(("DbUpdWorker: xxWrite failed\n"));
|
||||
tqp->workerExit();
|
||||
delete tsk;
|
||||
return (void*)0;
|
||||
@@ -1151,8 +1162,8 @@ bool Db::addOrUpdate(const string &udi, const string &parent_udi, Doc &doc)
|
||||
|
||||
#ifdef IDX_THREADS
|
||||
if (m_ndb->m_havewriteq) {
|
||||
DbUpdTask *tp = new DbUpdTask(udi, uniterm, newdocument,
|
||||
doc.text.length());
|
||||
DbUpdTask *tp = new DbUpdTask(DbUpdTask::AddOrUpdate, udi, uniterm,
|
||||
newdocument, doc.text.length());
|
||||
if (!m_ndb->m_wqueue.put(tp)) {
|
||||
LOGERR(("Db::addOrUpdate:Cant queue task\n"));
|
||||
return false;
|
||||
@@ -1292,7 +1303,7 @@ bool Db::doFlush()
|
||||
}
|
||||
|
||||
// Test if doc given by udi has changed since last indexed (test sigs)
|
||||
bool Db::needUpdate(const string &udi, const string& sig)
|
||||
bool Db::needUpdate(const string &udi, const string& sig, bool *existed)
|
||||
{
|
||||
if (m_ndb == 0)
|
||||
return false;
|
||||
@@ -1300,8 +1311,12 @@ bool Db::needUpdate(const string &udi, const string& sig)
|
||||
// If we are doing an in place or full reset, no need to
|
||||
// test. Note that there is no need to update the existence map
|
||||
// either, it will be done when updating the index
|
||||
if (o_inPlaceReset || m_mode == DbTrunc)
|
||||
if (o_inPlaceReset || m_mode == DbTrunc) {
|
||||
// For in place reset, pretend the doc existed, to enable subdoc purge
|
||||
if (existed)
|
||||
*existed = o_inPlaceReset;
|
||||
return true;
|
||||
}
|
||||
|
||||
string uniterm = make_uniterm(udi);
|
||||
string ermsg;
|
||||
@@ -1325,9 +1340,13 @@ bool Db::needUpdate(const string &udi, const string& sig)
|
||||
if (docid == m_ndb->xrdb.postlist_end(uniterm)) {
|
||||
// If no document exist with this path, we do need update
|
||||
LOGDEB(("Db::needUpdate:yes (new): [%s]\n", uniterm.c_str()));
|
||||
if (existed)
|
||||
*existed = false;
|
||||
return true;
|
||||
}
|
||||
Xapian::Document doc = m_ndb->xrdb.get_document(*docid);
|
||||
if (existed)
|
||||
*existed = true;
|
||||
|
||||
// Retrieve old file/doc signature from value
|
||||
string osig = doc.get_value(VALUE_SIG);
|
||||
@@ -1542,8 +1561,8 @@ bool Db::purgeFile(const string &udi, bool *existed)
|
||||
|
||||
#ifdef IDX_THREADS
|
||||
if (m_ndb->m_havewriteq) {
|
||||
Xapian::Document xdoc;
|
||||
DbUpdTask *tp = new DbUpdTask(udi, uniterm, xdoc, (size_t)-1);
|
||||
DbUpdTask *tp = new DbUpdTask(DbUpdTask::Delete, udi, uniterm,
|
||||
Xapian::Document(), (size_t)-1);
|
||||
if (!m_ndb->m_wqueue.put(tp)) {
|
||||
LOGERR(("Db::purgeFile:Cant queue task\n"));
|
||||
return false;
|
||||
@@ -1552,49 +1571,98 @@ bool Db::purgeFile(const string &udi, bool *existed)
|
||||
}
|
||||
}
|
||||
#endif
|
||||
|
||||
return purgeFileWrite(udi, uniterm);
|
||||
/* We get there is IDX_THREADS is not defined or there is no queue */
|
||||
return m_ndb->purgeFileWrite(false, udi, uniterm);
|
||||
}
|
||||
|
||||
bool Db::purgeFileWrite(const string& udi, const string& uniterm)
|
||||
/* Delete subdocs with an out of date sig. We do this to purge
|
||||
obsolete subdocs during a partial update where no general purge
|
||||
will be done */
|
||||
bool Db::purgeOrphans(const string &udi)
|
||||
{
|
||||
LOGDEB(("Db:purgeOrphans: [%s]\n", udi.c_str()));
|
||||
if (m_ndb == 0 || !m_ndb->m_iswritable)
|
||||
return false;
|
||||
|
||||
string uniterm = make_uniterm(udi);
|
||||
|
||||
#ifdef IDX_THREADS
|
||||
if (m_ndb->m_havewriteq) {
|
||||
DbUpdTask *tp = new DbUpdTask(DbUpdTask::PurgeOrphans, udi, uniterm,
|
||||
Xapian::Document(), (size_t)-1);
|
||||
if (!m_ndb->m_wqueue.put(tp)) {
|
||||
LOGERR(("Db::purgeFile:Cant queue task\n"));
|
||||
return false;
|
||||
} else {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
#endif
|
||||
/* We get there is IDX_THREADS is not defined or there is no queue */
|
||||
return m_ndb->purgeFileWrite(true, udi, uniterm);
|
||||
}
|
||||
|
||||
bool Db::Native::purgeFileWrite(bool orphansOnly, const string& udi,
|
||||
const string& uniterm)
|
||||
{
|
||||
#if defined(IDX_THREADS)
|
||||
// We need a mutex even if we have a write queue (so we can only
|
||||
// be called by a single thread) to protect about multiple acces
|
||||
// to xrdb from subDocs() which is also called from needupdate()
|
||||
// (called from outside the write thread !
|
||||
PTMutexLocker lock(m_ndb->m_mutex);
|
||||
PTMutexLocker lock(m_mutex);
|
||||
#endif // IDX_THREADS
|
||||
|
||||
Xapian::WritableDatabase db = m_ndb->xwdb;
|
||||
string ermsg;
|
||||
try {
|
||||
Xapian::PostingIterator docid = db.postlist_begin(uniterm);
|
||||
if (docid == db.postlist_end(uniterm)) {
|
||||
Xapian::PostingIterator docid = xwdb.postlist_begin(uniterm);
|
||||
if (docid == xwdb.postlist_end(uniterm)) {
|
||||
return true;
|
||||
}
|
||||
LOGDEB(("purgeFile: delete docid %d\n", *docid));
|
||||
if (m_flushMb > 0) {
|
||||
Xapian::termcount trms = m_ndb->xwdb.get_doclength(*docid);
|
||||
maybeflush(trms * 5);
|
||||
if (m_rcldb->m_flushMb > 0) {
|
||||
Xapian::termcount trms = xwdb.get_doclength(*docid);
|
||||
m_rcldb->maybeflush(trms * 5);
|
||||
}
|
||||
string sig;
|
||||
if (orphansOnly) {
|
||||
Xapian::Document doc = xwdb.get_document(*docid);
|
||||
sig = doc.get_value(VALUE_SIG);
|
||||
if (sig.empty()) {
|
||||
LOGINFO(("purgeFileWrite: got empty sig\n"));
|
||||
return false;
|
||||
}
|
||||
} else {
|
||||
LOGDEB(("purgeFile: delete docid %d\n", *docid));
|
||||
xwdb.delete_document(*docid);
|
||||
}
|
||||
db.delete_document(*docid);
|
||||
vector<Xapian::docid> docids;
|
||||
m_ndb->subDocs(udi, docids);
|
||||
subDocs(udi, docids);
|
||||
LOGDEB(("purgeFile: subdocs cnt %d\n", docids.size()));
|
||||
for (vector<Xapian::docid>::iterator it = docids.begin();
|
||||
it != docids.end(); it++) {
|
||||
LOGDEB(("Db::purgeFile: delete subdoc %d\n", *it));
|
||||
if (m_flushMb > 0) {
|
||||
Xapian::termcount trms = m_ndb->xwdb.get_doclength(*it);
|
||||
maybeflush(trms * 5);
|
||||
if (m_rcldb->m_flushMb > 0) {
|
||||
Xapian::termcount trms = xwdb.get_doclength(*it);
|
||||
m_rcldb->maybeflush(trms * 5);
|
||||
}
|
||||
string subdocsig;
|
||||
if (orphansOnly) {
|
||||
Xapian::Document doc = xwdb.get_document(*it);
|
||||
subdocsig = doc.get_value(VALUE_SIG);
|
||||
if (subdocsig.empty()) {
|
||||
LOGINFO(("purgeFileWrite: got empty sig for subdoc??\n"));
|
||||
continue;
|
||||
}
|
||||
}
|
||||
|
||||
if (!orphansOnly || sig != subdocsig) {
|
||||
LOGDEB(("Db::purgeFile: delete subdoc %d\n", *it));
|
||||
xwdb.delete_document(*it);
|
||||
}
|
||||
db.delete_document(*it);
|
||||
}
|
||||
return true;
|
||||
} XCATCHERROR(ermsg);
|
||||
if (!ermsg.empty()) {
|
||||
LOGERR(("Db::purgeFile: %s\n", ermsg.c_str()));
|
||||
LOGERR(("Db::purgeFileWrite: %s\n", ermsg.c_str()));
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
+5
-2
@@ -232,7 +232,7 @@ class Db {
|
||||
* Side-effect: set the existence flag for the file document
|
||||
* and all subdocs if any (for later use by 'purge()')
|
||||
*/
|
||||
bool needUpdate(const string &udi, const string& sig);
|
||||
bool needUpdate(const string &udi, const string& sig, bool *existed=0);
|
||||
|
||||
/** Add or update document identified by unique identifier.
|
||||
* @param config Config object to use. Can be the same as the member config
|
||||
@@ -260,6 +260,10 @@ class Db {
|
||||
|
||||
/** Delete document(s) for given UDI, including subdocs */
|
||||
bool purgeFile(const string &udi, bool *existed = 0);
|
||||
/** Delete subdocs with an out of date sig. We do this to purge
|
||||
obsolete subdocs during a partial update where no general purge
|
||||
will be done */
|
||||
bool purgeOrphans(const string &udi);
|
||||
|
||||
/** Remove documents that no longer exist in the file system. This
|
||||
* depends on the update map, which is built during
|
||||
@@ -442,7 +446,6 @@ private:
|
||||
#ifdef IDX_THREADS
|
||||
friend void *DbUpdWorker(void*);
|
||||
#endif // IDX_THREADS
|
||||
bool purgeFileWrite(const string& udi, const string& uniterm);
|
||||
|
||||
// Internal form of close, can be called during destruction
|
||||
bool i_close(bool final);
|
||||
|
||||
+22
-5
@@ -36,17 +36,32 @@ namespace Rcl {
|
||||
class Query;
|
||||
|
||||
#ifdef IDX_THREADS
|
||||
// Task for the index update thread. This can be either and add/update
|
||||
// or a purge op, in which case the txtlen is (size_t)-1
|
||||
// Task for the index update thread. This can be
|
||||
// - add/update for a new / update documment
|
||||
// - delete for a deleted document
|
||||
// - purgeOrphans when a multidoc file is updated during a partial pass (no
|
||||
// general purge). We want to remove subDocs that possibly don't
|
||||
// exist anymore. We find them by their different sig
|
||||
// txtlen and doc are only valid for add/update else, len is (size_t)-1 and doc
|
||||
// is empty
|
||||
class DbUpdTask {
|
||||
public:
|
||||
DbUpdTask(const string& ud, const string& un, const Xapian::Document &d,
|
||||
size_t tl)
|
||||
: udi(ud), uniterm(un), doc(d), txtlen(tl)
|
||||
enum Op {AddOrUpdate, Delete, PurgeOrphans};
|
||||
// Note that udi and uniterm are strictly equivalent and are
|
||||
// passed both just to avoid recomputing uniterm which is
|
||||
// available on the caller site.
|
||||
DbUpdTask(Op _op, const string& ud, const string& un,
|
||||
const Xapian::Document &d, size_t tl)
|
||||
: op(_op), udi(ud), uniterm(un), doc(d), txtlen(tl)
|
||||
{}
|
||||
// Udi and uniterm equivalently designate the doc
|
||||
Op op;
|
||||
string udi;
|
||||
string uniterm;
|
||||
Xapian::Document doc;
|
||||
// txtlen is used to update the flush interval. It's -1 for a
|
||||
// purge because we actually don't know it, and the code fakes a
|
||||
// text length based on the term count.
|
||||
size_t txtlen;
|
||||
};
|
||||
#endif // IDX_THREADS
|
||||
@@ -86,6 +101,8 @@ class Db::Native {
|
||||
bool addOrUpdateWrite(const string& udi, const string& uniterm,
|
||||
Xapian::Document& doc, size_t txtlen);
|
||||
|
||||
bool purgeFileWrite(bool onlyOrphans, const string& udi,
|
||||
const string& uniterm);
|
||||
bool getPagePositions(Xapian::docid docid, vector<int>& vpos);
|
||||
int getPageNumberForPosition(const vector<int>& pbreaks, unsigned int pos);
|
||||
|
||||
|
||||
Reference in New Issue
Block a user