Changeset 293

Timestamp:

02/22/08 04:06:12 (9 months ago)

Author:

jake

Message:

Ignore on disk docs that have been updated

Location:

trunk/thrudex/src

Files:

• CLuceneIndex.cpp (modified) (19 diffs)
• CLuceneIndex.h (modified) (2 diffs)
• Makefile.am (modified) (1 diff)
• UpdateFilter.cpp (added)
• UpdateFilter.h (added)
• thrudex.conf (modified) (1 diff)

trunk/thrudex/src/CLuceneIndex.cpp

@@ -11 +11 @@
 
 #include "bloom_filter.hpp"
+#include "UpdateFilter.h"
 
 using namespace thrudex;
@@ -73 +74 @@
 
         //build up bloom filter
-
         disk_bloom    = shared_ptr<bloom_filter>(new bloom_filter(filter_space,1.0/(1.0 * filter_space), random_seed));
+        disk_reader   = shared_ptr<IndexReader>(IndexReader::open(idx_path.c_str()));
+        disk_filter   = shared_ptr<UpdateFilter>(new UpdateFilter(disk_reader));
 
         if(!new_index){
 
-            boost::shared_ptr<IndexReader>  reader(IndexReader::open(idx_path.c_str()));
-
-            int max = reader->maxDoc();
+            int max = disk_reader->maxDoc();
             char buf[1024];
 
             for(int i=0; i<max; i++){
-                const wchar_t *id   = reader->document(i)->get( DOC_KEY );
+                const wchar_t *id   = disk_reader->document(i)->get( DOC_KEY );
 
                 STRCPY_TtoA(buf,id,1024);
@@ -93 +93 @@
             }
 
-            reader->close();
+
         }
 
@@ -105 +105 @@
         ram_searcher  = shared_ptr<IndexSearcher>(new IndexSearcher(ram_directory.get()));
 
-        disk_searcher = shared_ptr<IndexSearcher>(new IndexSearcher(idx_path.c_str()));
+        disk_searcher = shared_ptr<IndexSearcher>(new IndexSearcher(disk_reader.get()));
         last_refresh  = -1;
 
@@ -144 +144 @@
 CLuceneIndex::~CLuceneIndex()
 {
-    //monitor_thread->stop();
     sync();
 }
@@ -187 +186 @@
 {
 
-    assert(!key.empty());
-
+    if(key.empty()){
+        ThrudexException ex;
+        ex.what = "Empty key";
+        throw ex;
+    }
 
     Guard g( mutex );
-
 
     //always put into memory (we will merge to disk later)
@@ -198 +199 @@
     shared_ptr<IndexModifier> l_modifier     = modifier;
     shared_ptr<set<string> >  l_disk_deletes = disk_deletes;
+    shared_ptr<UpdateFilter>  l_disk_filter  = disk_filter;
+
+    wstring wkey = build_wstring(key);
 
     //if update to a doc in memory old copy first
     if( ram_bloom->contains( key ) ){
 
-        wstring wkey = build_wstring(key);
         Term *t = new Term(DOC_KEY, wkey.c_str() );
 
@@ -212 +215 @@
 
 
-    //If this exists already on disk
-    //remove it
+    //If this exists already on disk remove it
     if( l_disk_bloom->contains( key ) ){
         l_disk_deletes->insert( key );
+
+        if(!syncing)
+            l_disk_filter->skip(wkey);
     }
 
     l_modifier->addDocument(doc);
-
     l_ram_bloom->insert( key );
 
@@ -233 +237 @@
     shared_ptr<IndexModifier> l_modifier     = modifier;
     shared_ptr<set<string> >  l_disk_deletes = disk_deletes;
-
+    shared_ptr<UpdateFilter>  l_disk_filter  = disk_filter;
+
+    wstring wkey = build_wstring(key);
 
     //Since we don't want to write to disk
@@ -240 +246 @@
         LOG4CXX_DEBUG(logger, "Removed "+key);
         l_disk_deletes->insert( key );
+
+        if(!syncing)
+            l_disk_filter->skip(wkey);
+
         last_modified = Util::currentTime();
     }
@@ -246 +256 @@
     if(l_ram_bloom->contains( key )){
 
-        wstring wkey = build_wstring(key);
         Term      *t = new Term(DOC_KEY, wkey.c_str() );
 
@@ -266 +275 @@
     shared_ptr<CLuceneRAMDirectory> l_ram_prev_directory = ram_prev_directory;
 
-    shared_ptr<MultiSearcher> l_searcher = this->getSearcher();
+    shared_ptr<MultiSearcher> l_searcher    = this->getSearcher();
+    shared_ptr<UpdateFilter>  l_disk_filter = disk_filter;
 
     Query *query;
@@ -301 +311 @@
 
     if( q.sortby.empty() ){
-        h = l_searcher->search(query);
+        h = l_searcher->search(query, l_disk_filter.get());
     } else {
 
@@ -313 +323 @@
 
         try {
-            h = l_searcher->search(query,lsort);
+            h = l_searcher->search(query,l_disk_filter.get(),lsort);
         } catch(CLuceneError &e) {
 
             LOG4CXX_WARN(logger, "Sort failed, falling back on regular search");
-            h = l_searcher->search(query);
+            h = l_searcher->search(query,l_disk_filter.get());
         }
     }
@@ -419 +429 @@
 
     LOG4CXX_DEBUG(logger,"Syncing Started");
-
+    string idx_path = index_root + "/" + index_name;
 
     shared_ptr<IndexModifier>       l_ram_modifier;
@@ -426 +436 @@
     shared_ptr<set<string> >        l_disk_deletes;
     shared_ptr<bloom_filter>        l_disk_bloom;
+    shared_ptr<UpdateFilter>        l_update_filter;
 
 
@@ -459 +470 @@
 
     //Now we start by deleting any updated docs from disk
-    string idx_path = index_root + "/" + index_name;
 
     shared_ptr<IndexReader> disk_reader(IndexReader::open(idx_path.c_str()));
@@ -498 +508 @@
         LOG4CXX_DEBUG(logger,"Merged");
     }
+
     //Search new index (big perf hit so get it over now)
-
-    shared_ptr<IndexSearcher> l_disk_searcher(new IndexSearcher(idx_path.c_str()));
+    shared_ptr<IndexReader>   l_disk_reader(IndexReader::open(idx_path.c_str()));
+    shared_ptr<IndexSearcher> l_disk_searcher(new IndexSearcher(l_disk_reader.get()));
+    shared_ptr<UpdateFilter>  l_disk_filter(new UpdateFilter(l_disk_reader));
+
+
     wstring q = wstring(DOC_KEY)+wstring(L":1234");
 
@@ -514 +528 @@
         Guard g(mutex);
 
+        //Add any new deletes to the filter
+        set<string>::iterator it;
+        for( it=disk_deletes->begin(); it!=disk_deletes->end(); ++it){
+            wstring wkey = build_wstring(*it);
+            l_disk_filter->skip(wkey);
+        }
+
+        disk_reader   = l_disk_reader;
+        disk_filter   = l_disk_filter;
+        disk_searcher = l_disk_searcher;
+
+
+        last_synched = Util::currentTime();
+
         syncing = false; //this flag alters the search code to include prev searcher
-
-        disk_searcher = l_disk_searcher;
-
-        last_synched = Util::currentTime();
     }
 

trunk/thrudex/src/CLuceneIndex.h

@@ -25 +25 @@
 
 class bloom_filter;
+class UpdateFilter;
 
 #define DOC_KEY L"_doc_key_"
@@ -80 +81 @@
     bool                                             syncing;
 
+    boost::shared_ptr<lucene::index::IndexReader>    disk_reader;
+    boost::shared_ptr<UpdateFilter>                  disk_filter;
     boost::shared_ptr<lucene::search::IndexSearcher> disk_searcher;
     boost::shared_ptr<bloom_filter>                  disk_bloom;

trunk/thrudex/src/Makefile.am

@@ -10 +10 @@
                   CLuceneRAMDirectory.cpp               \
                   CLuceneIndex.cpp                      \
+                  UpdateFilter.cpp                      \
                   main.cpp
 

trunk/thrudex/src/thrudex.conf

@@ -16 +16 @@
 
 # Set root logger level to DEBUG and its only appender to A1.
-#log4j.rootLogger=DEBUG, A1
-log4j.rootLogger=INFO, A1
+log4j.rootLogger=DEBUG, A1
+#log4j.rootLogger=INFO, A1
 
 # A1 is set to be a ConsoleAppender.